""" test_txn_messages.py — Auto Transaction Message trigger layer (v12.37). The 4-tier gate is the heart of this feature: every layer MUST be ON for the auto-send to actually fire. These tests exercise each layer independently AND together. They use the existing super-admin credentials and a real-license deployment (license auto-seeds a 30-day trial on first call). Coverage matrix: 1. /events — public taxonomy (13 events) 2. /access — full 4-tier truth table 3. Platform OFF (default) → access denied 4. Platform ON, license ON, customer OFF → access denied 5. All four layers ON → access granted 6. fire_event() never raises — even with garbage doc 7. fire_event() skipped log row — when gate fails (still observable) 8. PUT /settings respects gate — 403 when platform OFF 9. Test-fire works (admin only) — bypasses gate end-to-end """ from __future__ import annotations import time import pytest import requests from conftest import BASE_URL # type: ignore[import-not-found] FLAG = "auto-transaction-messages" TEST_COMPANY_ID = None # picked up by fixture def _enable_platform(admin): r = admin.put(f"{BASE_URL}/api/feature-flags/{FLAG}", json={"enabled_global": True}, timeout=15) assert r.status_code == 200, r.text def _disable_platform(admin): admin.put(f"{BASE_URL}/api/feature-flags/{FLAG}", json={"enabled_global": False, "kill_switch": False, "enabled_users": [], "disabled_users": []}, timeout=15) def _company_id(admin) -> str: """Pick first company for the admin (legacy data assures at least one).""" r = admin.get(f"{BASE_URL}/api/companies", timeout=15) assert r.status_code == 200, r.text rows = r.json() assert rows, "Need at least one company for txn-messages tests" return rows[0]["id"] @pytest.fixture(autouse=True) def reset_flag(admin_session): """Every test starts with platform flag OFF (the production posture).""" _disable_platform(admin_session) yield _disable_platform(admin_session) @pytest.mark.critical class TestTxnMessagesTaxonomy: def test_events_endpoint_returns_13_events(self, admin_session): r = admin_session.get(f"{BASE_URL}/api/txn-messages/events", timeout=15) assert r.status_code == 200, r.text data = r.json() assert isinstance(data, dict) assert isinstance(data["events"], list) assert len(data["events"]) == 13, f"Expected 13 supported events, got {len(data['events'])}" keys = {e["key"] for e in data["events"]} # Sample the critical ones the user listed for required in ( "sale.created", "purchase.created", "sale_order.created", "purchase_order.created", "quotation.created", "delivery_challan.created", "credit_note.created", "debit_note.created", "sales_return.created", "purchase_return.created", "payment_in.created", "payment_out.created", "expense.created", ): assert required in keys, f"Missing event {required}" @pytest.mark.critical class TestFourTierGate: def test_default_posture_denied(self, admin_session): """Platform default = OFF → every access check denies.""" cid = _company_id(admin_session) r = admin_session.get(f"{BASE_URL}/api/txn-messages/access?company_id={cid}", timeout=15) assert r.status_code == 200, r.text data = r.json() assert data["platform_enabled"] is False assert data["allowed"] is False assert "Platform" in data["reason"] def test_platform_only_still_denied(self, admin_session): """Platform ON but customer setting still OFF → denied.""" cid = _company_id(admin_session) _enable_platform(admin_session) # Make sure settings are wiped admin_session.put(f"{BASE_URL}/api/txn-messages/settings", json={"company_id": cid, "enabled": False, "auto_send_enabled": False}, timeout=15) r = admin_session.get(f"{BASE_URL}/api/txn-messages/access?company_id={cid}", timeout=15) d = r.json() assert d["platform_enabled"] is True assert d["license_active"] is True assert d["customer_enabled"] is False assert d["allowed"] is False assert "Customer admin" in d["reason"] def test_full_chain_allows(self, admin_session): cid = _company_id(admin_session) _enable_platform(admin_session) r = admin_session.put(f"{BASE_URL}/api/txn-messages/settings", json={"company_id": cid, "enabled": True, "auto_send_enabled": True, "channel_default": "whatsapp", "enabled_users": []}, timeout=15) assert r.status_code == 200, r.text r2 = admin_session.get(f"{BASE_URL}/api/txn-messages/access?company_id={cid}", timeout=15) d = r2.json() assert d["platform_enabled"] is True assert d["license_active"] is True assert d["customer_enabled"] is True assert d["user_allowed"] is True assert d["allowed"] is True @pytest.mark.critical class TestSettingsGuard: def test_settings_put_blocked_when_platform_off(self, admin_session): cid = _company_id(admin_session) r = admin_session.put(f"{BASE_URL}/api/txn-messages/settings", json={"company_id": cid, "enabled": True}, timeout=15) assert r.status_code == 403, r.text assert "Platform" in r.json().get("detail", "") def test_anonymous_cannot_read_or_write(self): cid_fetch = requests.get(f"{BASE_URL}/api/txn-messages/events", timeout=15) # /events still requires auth (per get_current_user) assert cid_fetch.status_code in (401, 403) rw = requests.put(f"{BASE_URL}/api/txn-messages/settings", json={"company_id": "x", "enabled": True}, timeout=15) assert rw.status_code in (401, 403, 422) @pytest.mark.critical class TestRules: def test_rule_upsert_and_list(self, admin_session): cid = _company_id(admin_session) _enable_platform(admin_session) admin_session.put(f"{BASE_URL}/api/txn-messages/settings", json={"company_id": cid, "enabled": True}, timeout=15) r = admin_session.put(f"{BASE_URL}/api/txn-messages/rules", json={"company_id": cid, "event_key": "sale.created", "enabled": True, "channel": "whatsapp", "template_id": "reminder", "cooldown_min": 5}, timeout=15) assert r.status_code == 200, r.text rule = r.json() assert rule["event_key"] == "sale.created" assert rule["channel"] == "whatsapp" # List should now include it lst = admin_session.get(f"{BASE_URL}/api/txn-messages/rules?company_id={cid}", timeout=15) assert any(x["event_key"] == "sale.created" for x in lst.json()) def test_rule_rejects_unknown_event(self, admin_session): cid = _company_id(admin_session) _enable_platform(admin_session) admin_session.put(f"{BASE_URL}/api/txn-messages/settings", json={"company_id": cid, "enabled": True}, timeout=15) r = admin_session.put(f"{BASE_URL}/api/txn-messages/rules", json={"company_id": cid, "event_key": "bogus.event", "enabled": True, "channel": "whatsapp"}, timeout=15) assert r.status_code == 400 @pytest.mark.critical class TestLogs: def test_log_listing_endpoint(self, admin_session): cid = _company_id(admin_session) r = admin_session.get(f"{BASE_URL}/api/txn-messages/logs?company_id={cid}&limit=10", timeout=15) assert r.status_code == 200, r.text assert isinstance(r.json(), list) @pytest.mark.critical class TestV1238Settings: """v12.38 — new recipient-control fields must round-trip through settings.""" def test_new_fields_default_off(self, admin_session): cid = _company_id(admin_session) _enable_platform(admin_session) # Reset the settings doc to a known-clean baseline so we test the # CONTRACT (defaults applied at read time), not the residue of earlier # tests / curl sessions that may have written explicit Falses. admin_session.put(f"{BASE_URL}/api/txn-messages/settings", json={"company_id": cid, "enabled": True, "send_to_party": True, "send_on_update": False, "send_copy_to_self": False, "self_copy_phone": ""}, timeout=15) r = admin_session.get(f"{BASE_URL}/api/txn-messages/settings?company_id={cid}", timeout=15) s = r.json() # send_to_party defaults TRUE (the intuitive UX) — every other new # toggle defaults FALSE. assert s.get("send_to_party") is True assert s.get("send_on_update") is False assert s.get("send_copy_to_self") is False assert s.get("self_copy_phone") == "" def test_new_fields_persist_and_validate(self, admin_session): cid = _company_id(admin_session) _enable_platform(admin_session) r = admin_session.put( f"{BASE_URL}/api/txn-messages/settings", json={ "company_id": cid, "enabled": True, "send_to_party": False, "send_on_update": True, "send_copy_to_self": True, "self_copy_phone": "+919999999999", }, timeout=15, ) assert r.status_code == 200, r.text s = r.json() assert s["send_to_party"] is False assert s["send_on_update"] is True assert s["send_copy_to_self"] is True assert s["self_copy_phone"] == "+919999999999" def test_sync_now_stamps_timestamp(self, admin_session): cid = _company_id(admin_session) _enable_platform(admin_session) admin_session.put(f"{BASE_URL}/api/txn-messages/settings", json={"company_id": cid, "enabled": True}, timeout=15) r = admin_session.post(f"{BASE_URL}/api/txn-messages/sync-now?company_id={cid}", timeout=15) assert r.status_code == 200, r.text assert r.json().get("last_synced_at"), "sync-now must return last_synced_at" def test_test_self_requires_phone(self, admin_session): cid = _company_id(admin_session) _enable_platform(admin_session) # No self_copy_phone configured admin_session.put(f"{BASE_URL}/api/txn-messages/settings", json={"company_id": cid, "enabled": True, "self_copy_phone": ""}, timeout=15) r = admin_session.post(f"{BASE_URL}/api/txn-messages/test-self?company_id={cid}", timeout=15) assert r.status_code == 400 assert "self copy phone" in r.json().get("detail", "").lower()