"""RBS REGAL — Tests for Security Engine + Marketing Hub (iteration 10). Covers: • Password strength scorer (strong vs weak) • Forgot/Reset password roundtrip (incl. restoration of admin pw) • Login OTP request/verify (success + wrong code + reuse) • Login history (self) • Admin security alerts/dashboard/resolve • Marketing templates, campaigns (valid + invalid channel), coupons (create/list/redeem/limit/delete) • AI ad generator (LLM) • Marketing stats • Authorization (403 for non-admin on admin endpoints) """ import uuid import pytest from conftest import BASE_URL, ADMIN_EMAIL, ADMIN_PASSWORD, _new_session, _login # ============ Module-scoped state ===================================== _state = { "reset_token": None, "otp_token": None, "otp_code": None, "coupon_code": None, "coupon_id": None, "coupon_limited_code": None, "campaign_id": None, "alert_id": None, "non_admin_email": f"TEST_secuser_{uuid.uuid4().hex[:6]}@example.com", "non_admin_pw": "TestUser@2026X", "non_admin_id": None, } # ============ Password Strength ======================================= def test_password_strength_strong(admin_session): r = admin_session.post(f"{BASE_URL}/api/security/password-strength", json={"password": "Rvasa@#9955"}, timeout=20) assert r.status_code == 200, r.text body = r.json() assert body["score"] >= 3, f"Expected score>=3 got {body}" assert "label" in body assert "suggestions" in body and isinstance(body["suggestions"], list) def test_password_strength_weak(admin_session): r = admin_session.post(f"{BASE_URL}/api/security/password-strength", json={"password": "123"}, timeout=20) assert r.status_code == 200, r.text body = r.json() assert body["score"] == 0, f"Expected score=0 for '123' got {body}" assert isinstance(body["suggestions"], list) and len(body["suggestions"]) > 0 # ============ Forgot Password ======================================== def test_forgot_password_existing_email(): s = _new_session() # public endpoint, no auth needed r = s.post(f"{BASE_URL}/api/security/forgot-password", json={"email": ADMIN_EMAIL}, timeout=20) assert r.status_code == 200, r.text body = r.json() assert body["ok"] is True assert body["reset_token"], f"Expected reset_token, got {body}" assert body["expires_in_minutes"] == 60 _state["reset_token"] = body["reset_token"] def test_forgot_password_nonexistent_email_no_enumeration(): s = _new_session() r = s.post(f"{BASE_URL}/api/security/forgot-password", json={"email": "TEST_nobody_xyz_404@nowhere.example"}, timeout=20) assert r.status_code == 200, r.text body = r.json() assert body["ok"] is True assert body["reset_token"] is None # expires_in_minutes is None when no user assert body.get("expires_in_minutes") in (None, 0) # ============ Reset Password (weak/wrong/expired) ==================== def test_reset_password_weak_rejected(): s = _new_session() r = s.post(f"{BASE_URL}/api/security/reset-password", json={"email": ADMIN_EMAIL, "token": _state["reset_token"] or "x", "new_password": "abc"}, timeout=20) assert r.status_code == 400, r.text assert "weak" in r.text.lower() def test_reset_password_wrong_token(): s = _new_session() r = s.post(f"{BASE_URL}/api/security/reset-password", json={"email": ADMIN_EMAIL, "token": "wrong-token-zzz", "new_password": "StrongPass#2026!"}, timeout=20) assert r.status_code == 400, r.text assert "invalid" in r.text.lower() # ============ Reset Password success + verify + restore ============== def test_reset_password_success_and_restore(): """Reset admin pw to a temp, verify old pw fails + new pw works, then restore.""" assert _state["reset_token"], "Reset token missing — earlier test failed" s = _new_session() new_pw = "TempReset#2026Z" r = s.post(f"{BASE_URL}/api/security/reset-password", json={"email": ADMIN_EMAIL, "token": _state["reset_token"], "new_password": new_pw}, timeout=30) assert r.status_code == 200, f"Reset failed: {r.status_code} {r.text}" # OLD password should now fail r_old = _login(_new_session(), ADMIN_EMAIL, ADMIN_PASSWORD) assert r_old.status_code in (401, 403), f"Old pw should fail, got {r_old.status_code}" # NEW password should work r_new = _login(_new_session(), ADMIN_EMAIL, new_pw) assert r_new.status_code == 200, f"New pw should work: {r_new.text}" # Now restore via fresh reset s2 = _new_session() r2 = s2.post(f"{BASE_URL}/api/security/forgot-password", json={"email": ADMIN_EMAIL}, timeout=20) assert r2.status_code == 200, r2.text restore_token = r2.json()["reset_token"] assert restore_token r3 = s2.post(f"{BASE_URL}/api/security/reset-password", json={"email": ADMIN_EMAIL, "token": restore_token, "new_password": ADMIN_PASSWORD}, timeout=30) assert r3.status_code == 200, f"Restore failed: {r3.text}" # Verify original pw works again r_final = _login(_new_session(), ADMIN_EMAIL, ADMIN_PASSWORD) assert r_final.status_code == 200, f"Admin pw NOT restored! {r_final.text}" # ============ Login OTP ============================================== def test_login_otp_request(): s = _new_session() r = s.post(f"{BASE_URL}/api/security/login-otp/request", json={"email": ADMIN_EMAIL, "password": ADMIN_PASSWORD, "channel": "auto"}, timeout=20) assert r.status_code == 200, r.text body = r.json() assert body["ok"] is True assert body["otp_token"] assert body["expires_in_minutes"] == 10 assert body.get("code_preview"), "Expected code_preview because EXPOSE_OTP_CODE defaults to '1'" _state["otp_token"] = body["otp_token"] _state["otp_code"] = body["code_preview"] def test_login_otp_verify_wrong_code_increments_tries(): assert _state["otp_token"], "Missing otp_token" s = _new_session() r = s.post(f"{BASE_URL}/api/security/login-otp/verify", json={"otp_token": _state["otp_token"], "code": "000000"}, timeout=20) assert r.status_code == 401, r.text assert "wrong" in r.text.lower() or "invalid" in r.text.lower() def test_login_otp_verify_success_sets_cookie(): assert _state["otp_token"] and _state["otp_code"] s = _new_session() r = s.post(f"{BASE_URL}/api/security/login-otp/verify", json={"otp_token": _state["otp_token"], "code": _state["otp_code"]}, timeout=20) assert r.status_code == 200, r.text body = r.json() assert body["email"] == ADMIN_EMAIL assert body["role"] == "admin" # Cookie was set has_cookie = any(c.name == "access_token" for c in s.cookies) assert has_cookie, f"access_token cookie not set. Got: {[c.name for c in s.cookies]}" def test_login_otp_verify_reuse_after_success_fails(): assert _state["otp_token"] and _state["otp_code"] s = _new_session() r = s.post(f"{BASE_URL}/api/security/login-otp/verify", json={"otp_token": _state["otp_token"], "code": _state["otp_code"]}, timeout=20) assert r.status_code == 401, r.text assert "invalid otp session" in r.text.lower() # ============ Login History (self) =================================== def test_login_history_self(admin_session): r = admin_session.get(f"{BASE_URL}/api/security/login-history?limit=20", timeout=20) assert r.status_code == 200, r.text data = r.json() assert isinstance(data, list) # Admin certainly logged in — but list MAY be empty if no audit rows yet matched user_id. # If we have rows, validate shape if data: row = data[0] for k in ("ts", "action"): assert k in row, f"missing {k}: {row}" # ============ Admin Security: alerts / dashboard / resolve =========== def test_admin_security_dashboard(admin_session): r = admin_session.get(f"{BASE_URL}/api/admin/security/dashboard", timeout=20) assert r.status_code == 200, r.text body = r.json() for k in ("alerts_unresolved", "failed_logins_7d", "successful_logins_7d", "locked_accounts", "active_password_resets", "pending_otps"): assert k in body, f"missing key {k}: {body}" assert isinstance(body[k], int) def test_admin_security_alerts_list(admin_session): r = admin_session.get(f"{BASE_URL}/api/admin/security/alerts?only_unresolved=false", timeout=20) assert r.status_code == 200, r.text docs = r.json() assert isinstance(docs, list) if docs: _state["alert_id"] = docs[0].get("id") assert "_id" not in docs[0], "Must not leak Mongo _id" def test_admin_security_alert_resolve(admin_session): if not _state.get("alert_id"): pytest.skip("No alerts to resolve") r = admin_session.post(f"{BASE_URL}/api/admin/security/alerts/{_state['alert_id']}/resolve", timeout=20) assert r.status_code == 200, r.text assert r.json().get("ok") is True # ============ Marketing Templates ==================================== def test_marketing_templates(admin_session): r = admin_session.get(f"{BASE_URL}/api/marketing/templates", timeout=20) assert r.status_code == 200, r.text docs = r.json() assert isinstance(docs, list) assert len(docs) == 7, f"Expected 7 templates, got {len(docs)}" keys = {d["key"] for d in docs} expected = {"festival", "diwali", "thank_you", "reminder", "offer", "new_arrival", "birthday"} assert keys == expected, f"Missing/extra template keys: {keys ^ expected}" # ============ Marketing Coupons ====================================== def test_marketing_coupon_create(admin_session): r = admin_session.post(f"{BASE_URL}/api/marketing/coupons", json={"label": "TEST_Diwali 10%", "discount_pct": 10, "max_redemptions": 50}, timeout=20) assert r.status_code == 200, r.text body = r.json() assert body["label"] == "TEST_Diwali 10%" assert body["discount_pct"] == 10 assert body["max_redemptions"] == 50 assert body.get("code"), "Auto-generated code missing" assert len(body["code"]) >= 6 assert body.get("id") _state["coupon_code"] = body["code"] _state["coupon_id"] = body["id"] def test_marketing_coupons_list_includes_new(admin_session): r = admin_session.get(f"{BASE_URL}/api/marketing/coupons", timeout=20) assert r.status_code == 200, r.text codes = [d["code"] for d in r.json()] assert _state["coupon_code"] in codes def test_marketing_coupon_redeem(admin_session): assert _state["coupon_code"] r = admin_session.post(f"{BASE_URL}/api/marketing/coupons/{_state['coupon_code']}/redeem", timeout=20) assert r.status_code == 200, r.text body = r.json() assert body["ok"] is True assert body["code"] == _state["coupon_code"] assert body["discount_pct"] == 10 def test_marketing_coupon_redeem_limit_reached(admin_session): """Create a coupon with max_redemptions=1, redeem twice — second should fail with 400.""" r = admin_session.post(f"{BASE_URL}/api/marketing/coupons", json={"label": "TEST_LIMITED", "discount_pct": 5, "max_redemptions": 1}, timeout=20) assert r.status_code == 200, r.text code = r.json()["code"] cid = r.json()["id"] _state["coupon_limited_code"] = code _state["coupon_limited_id"] = cid r1 = admin_session.post(f"{BASE_URL}/api/marketing/coupons/{code}/redeem", timeout=20) assert r1.status_code == 200, r1.text r2 = admin_session.post(f"{BASE_URL}/api/marketing/coupons/{code}/redeem", timeout=20) assert r2.status_code == 400, r2.text assert "limit" in r2.text.lower() or "reached" in r2.text.lower() def test_marketing_coupon_delete_admin_only(admin_session): if not _state.get("coupon_id"): pytest.skip("No coupon to delete") r = admin_session.delete(f"{BASE_URL}/api/marketing/coupons/{_state['coupon_id']}", timeout=20) assert r.status_code == 200, r.text # Cleanup the limited coupon too if _state.get("coupon_limited_id"): admin_session.delete(f"{BASE_URL}/api/marketing/coupons/{_state['coupon_limited_id']}", timeout=20) # ============ Marketing Campaigns ==================================== @pytest.fixture(scope="module") def _seeded_customer(admin_session): """Seed a TEST_ customer party so campaign create has at least 1 recipient. Cleans up after module.""" # Get a company_id cr = admin_session.get(f"{BASE_URL}/api/companies", timeout=20) if cr.status_code != 200 or not cr.json(): pytest.skip("No company seeded") company_id = cr.json()[0]["id"] party_payload = {"name": f"TEST_Cust_{uuid.uuid4().hex[:6]}", "type": "customer", "phone": "+919999999999", "email": "TEST_x@example.com"} pr = admin_session.post(f"{BASE_URL}/api/parties?company_id={company_id}", json=party_payload, timeout=20) if pr.status_code not in (200, 201): pytest.skip(f"Could not seed customer party: {pr.status_code} {pr.text[:150]}") pid = pr.json().get("id") yield pid # cleanup if pid: try: admin_session.delete(f"{BASE_URL}/api/parties/{pid}", timeout=10) except Exception: pass def test_marketing_campaign_create(admin_session, _seeded_customer): r = admin_session.post(f"{BASE_URL}/api/marketing/campaign", json={"name": "TEST_Camp_Init", "channel": "whatsapp", "message": "Hi {{customer_name}} — test from RBS REGAL", "audience": "customers"}, timeout=30) assert r.status_code == 200, r.text body = r.json() assert body["ok"] is True assert "recipient_count" in body assert "sent" in body assert "dispatcher_active" in body _state["campaign_id"] = body.get("id") def test_marketing_campaigns_list(admin_session): r = admin_session.get(f"{BASE_URL}/api/marketing/campaigns", timeout=20) assert r.status_code == 200, r.text docs = r.json() assert isinstance(docs, list) if _state.get("campaign_id"): ids = [d["id"] for d in docs] assert _state["campaign_id"] in ids def test_marketing_campaign_invalid_channel(admin_session): r = admin_session.post(f"{BASE_URL}/api/marketing/campaign", json={"name": "TEST_BadChan", "channel": "invalid", "message": "x", "audience": "all"}, timeout=20) assert r.status_code == 400, r.text # ============ AI Ad ================================================= def test_marketing_ai_ad(admin_session): r = admin_session.post(f"{BASE_URL}/api/marketing/ai-ad", json={"product": "Diwali diyas", "style": "festival", "language": "hinglish"}, timeout=60) assert r.status_code == 200, f"AI ad failed: {r.status_code} {r.text[:300]}" body = r.json() for k in ("headline", "body", "cta", "hashtags"): assert k in body, f"missing {k}: {body}" assert isinstance(body["hashtags"], list) # ============ Marketing Stats ======================================= def test_marketing_stats(admin_session): r = admin_session.get(f"{BASE_URL}/api/marketing/stats", timeout=20) assert r.status_code == 200, r.text body = r.json() for k in ("total_campaigns", "messages_sent_30d", "active_coupons", "redemptions"): assert k in body, f"missing {k}: {body}" assert isinstance(body[k], int) # ============ Authorization (non-admin denied) ======================= @pytest.fixture(scope="module") def non_admin_session(admin_session): """Create a 'staff' user via /api/users (admin-only), return a logged-in session.""" email = _state["non_admin_email"] pw = _state["non_admin_pw"] payload = {"email": email, "password": pw, "name": "TEST Sec User", "role": "staff"} r = admin_session.post(f"{BASE_URL}/api/users", json=payload, timeout=20) if r.status_code not in (200, 201): pytest.skip(f"Could not create staff user: {r.status_code} {r.text[:200]}") body = r.json() _state["non_admin_id"] = body.get("id") s = _new_session() lr = _login(s, email, pw) if lr.status_code != 200: pytest.skip(f"Staff login failed: {lr.status_code} {lr.text[:200]}") yield s if _state.get("non_admin_id"): try: admin_session.delete(f"{BASE_URL}/api/users/{_state['non_admin_id']}", timeout=10) except Exception: pass def test_non_admin_denied_security_dashboard(non_admin_session): r = non_admin_session.get(f"{BASE_URL}/api/admin/security/dashboard", timeout=20) assert r.status_code == 403, f"Expected 403 for non-admin, got {r.status_code}: {r.text[:200]}"