""" test_pilot_activation.py — Pilot Customer Activation Wizard (v12.41). Coverage: 1. /api/pilot returns empty list initially 2. /api/pilot/activate flips all 3 flags' enabled_users[] 3. /api/pilot/me reflects active + show_welcome=True after activation 4. /api/pilot/welcome-ack clears the show_welcome flag (only once) 5. /api/pilot/revoke removes user from all 3 flags 6. /api/pilot/promote sets enabled_global=True on all flags 7. Anonymous cannot call any pilot endpoint 8. Invalid soak_hours rejected with 400 """ from __future__ import annotations import pytest import requests from conftest import BASE_URL # type: ignore[import-not-found] PILOT_FLAGS = ["service-worker", "mutation-queue", "auto-transaction-messages"] def _admin_id(admin): me = admin.get(f"{BASE_URL}/api/auth/me", timeout=15).json() return me.get("id") or me.get("_id") or me.get("user_id") def _flag_users(admin, flag_name): r = admin.get(f"{BASE_URL}/api/feature-flags", timeout=15) for f in r.json().get("flags", []): if f["name"] == flag_name: return f.get("enabled_users") or [] return None @pytest.fixture(autouse=True) def cleanup(admin_session): """Make sure no pilot state leaks between tests.""" uid = _admin_id(admin_session) if uid: admin_session.post(f"{BASE_URL}/api/pilot/revoke", json={"user_id": uid}, timeout=15) for f in PILOT_FLAGS: admin_session.put(f"{BASE_URL}/api/feature-flags/{f}", json={"enabled_global": False, "enabled_users": []}, timeout=15) yield if uid: admin_session.post(f"{BASE_URL}/api/pilot/revoke", json={"user_id": uid}, timeout=15) for f in PILOT_FLAGS: admin_session.put(f"{BASE_URL}/api/feature-flags/{f}", json={"enabled_global": False, "enabled_users": []}, timeout=15) @pytest.mark.critical class TestPilotActivation: def test_activate_flips_all_three_flags(self, admin_session): uid = _admin_id(admin_session) assert uid r = admin_session.post( f"{BASE_URL}/api/pilot/activate", json={"user_id": uid, "soak_hours": 24, "note": "pytest"}, timeout=15, ) assert r.status_code == 200, r.text data = r.json() assert data["ok"] is True assert sorted(data["flags_enabled"]) == sorted(PILOT_FLAGS) # Each flag must now contain the user_id for flag in PILOT_FLAGS: users = _flag_users(admin_session, flag) assert uid in users, f"{flag} should contain {uid}, got {users}" def test_me_shows_welcome_then_ack_clears(self, admin_session): uid = _admin_id(admin_session) admin_session.post(f"{BASE_URL}/api/pilot/activate", json={"user_id": uid, "soak_hours": 48}, timeout=15) # First /me — welcome flag is True r1 = admin_session.get(f"{BASE_URL}/api/pilot/me", timeout=15).json() assert r1["active"] is True assert r1["show_welcome"] is True assert 0 < r1["hours_remaining"] <= 48 # Ack r2 = admin_session.post(f"{BASE_URL}/api/pilot/welcome-ack", timeout=15) assert r2.status_code == 200, r2.text # Second /me — welcome flag is False (no replay) r3 = admin_session.get(f"{BASE_URL}/api/pilot/me", timeout=15).json() assert r3["active"] is True assert r3["show_welcome"] is False def test_revoke_clears_all_three_flags(self, admin_session): uid = _admin_id(admin_session) admin_session.post(f"{BASE_URL}/api/pilot/activate", json={"user_id": uid, "soak_hours": 1}, timeout=15) # Sanity for flag in PILOT_FLAGS: assert uid in (_flag_users(admin_session, flag) or []) # Revoke r = admin_session.post(f"{BASE_URL}/api/pilot/revoke", json={"user_id": uid}, timeout=15) assert r.status_code == 200, r.text for flag in PILOT_FLAGS: assert uid not in (_flag_users(admin_session, flag) or []) def test_promote_sets_enabled_global(self, admin_session): uid = _admin_id(admin_session) admin_session.post(f"{BASE_URL}/api/pilot/activate", json={"user_id": uid, "soak_hours": 1}, timeout=15) try: r = admin_session.post(f"{BASE_URL}/api/pilot/promote", json={"user_id": uid}, timeout=15) assert r.status_code == 200, r.text assert sorted(r.json()["promoted_flags"]) == sorted(PILOT_FLAGS) # Verify enabled_global=True for flag in PILOT_FLAGS: r2 = admin_session.get(f"{BASE_URL}/api/feature-flags", timeout=15) for f in r2.json()["flags"]: if f["name"] == flag: assert f["enabled_global"] is True break finally: for flag in PILOT_FLAGS: admin_session.put(f"{BASE_URL}/api/feature-flags/{flag}", json={"enabled_global": False}, timeout=15) def test_invalid_soak_hours_rejected(self, admin_session): uid = _admin_id(admin_session) for bad in (0, -1, 169, 99999): r = admin_session.post(f"{BASE_URL}/api/pilot/activate", json={"user_id": uid, "soak_hours": bad}, timeout=15) assert r.status_code == 400, f"soak_hours={bad} should be rejected, got {r.status_code}" def test_anonymous_denied(self): r1 = requests.get(f"{BASE_URL}/api/pilot", timeout=15) assert r1.status_code in (401, 403) r2 = requests.post(f"{BASE_URL}/api/pilot/activate", json={"user_id": "x", "soak_hours": 1}, timeout=15) assert r2.status_code in (401, 403, 422) @pytest.mark.critical class TestExpenseAutoNumbering: """v12.41 — Smart Serial for expenses (matches existing invoice behaviour).""" def _company_id(self, admin): r = admin.get(f"{BASE_URL}/api/companies", timeout=15) return r.json()[0]["id"] def test_create_expense_auto_numbers(self, admin_session): cid = self._company_id(admin_session) r = admin_session.post( f"{BASE_URL}/api/expenses?company_id={cid}", json={"category": "OfficeAuto", "amount": 100}, timeout=15, ) assert r.status_code == 200, r.text doc = r.json() assert doc.get("expense_no"), "expense_no must be auto-generated when not provided" assert "EXP" in doc["expense_no"] def test_manual_override_bumps_counter(self, admin_session): cid = self._company_id(admin_session) # Pin a specific number r1 = admin_session.post( f"{BASE_URL}/api/expenses?company_id={cid}", json={"category": "Manual", "amount": 50, "expense_no": "EXP/26/05000"}, timeout=15, ) assert r1.status_code == 200, r1.text assert r1.json()["expense_no"] == "EXP/26/05000" # Next auto should continue from 5001 r2 = admin_session.post( f"{BASE_URL}/api/expenses?company_id={cid}", json={"category": "Next", "amount": 60}, timeout=15, ) assert r2.status_code == 200, r2.text nxt = r2.json()["expense_no"] # Trailing digits must be ≥ 5001 (Smart Continue) import re m = re.search(r"(\d+)\s*$", nxt or "") assert m, f"expense_no should end with digits: {nxt}" assert int(m.group(1)) >= 5001, f"expected ≥ 5001, got {m.group(1)} from {nxt}" def test_duplicate_manual_rejected(self, admin_session): cid = self._company_id(admin_session) # Use a known-unique manual number manual = "EXP/26/UNIQ001" admin_session.post( f"{BASE_URL}/api/expenses?company_id={cid}", json={"category": "DupTest", "amount": 1, "expense_no": manual}, timeout=15, ) # Repeat r = admin_session.post( f"{BASE_URL}/api/expenses?company_id={cid}", json={"category": "DupRetry", "amount": 1, "expense_no": manual}, timeout=15, ) assert r.status_code == 400 assert "already exists" in r.json().get("detail", "").lower()