"""MailShift iteration 5 — Admin Integrations settings store, Migration report (PDF/CSV) download, and IMAP date-range regression.""" import imaplib import time import uuid import pymysql import pytest import requests from conftest import API, mig_payload from test_new_features import _wait_status, _purge_dest, _dest_email_count DB = dict(host="localhost", port=3306, user="mailshift", password="mailshift_pass_2026", database="mailshift") def _db_exec(sql): conn = pymysql.connect(**DB) try: with conn.cursor() as cur: cur.execute(sql) rows = cur.fetchall() conn.commit() return rows finally: conn.close() # ---------- Admin Integrations ---------- class TestIntegrations: """GET/PUT /api/admin/integrations — masking, persistence, fallback.""" @pytest.fixture(scope="class", autouse=True) def restore_settings(self, client): yield # Wipe every admin-managed override so preview falls back to .env again, # then force the in-process settings cache to reload from the (empty) DB. _db_exec("DELETE FROM app_settings") client.put(f"{API}/admin/integrations", json={}, timeout=30) r = client.get(f"{API}/admin/integrations", timeout=30) assert r.status_code == 200 assert _db_exec("SELECT COUNT(*) FROM app_settings")[0][0] == 0 @pytest.fixture(scope="class") def non_admin(self): email = f"TEST_integ_{uuid.uuid4().hex[:8]}@test.com" r = requests.post(f"{API}/auth/register", json={"email": email, "password": "Passw0rd!123", "name": "TEST integ"}, timeout=30) assert r.status_code in (200, 201), r.text return r.json()["token"] def test_get_structure_as_admin(self, client): r = client.get(f"{API}/admin/integrations", timeout=30) assert r.status_code == 200, r.text d = r.json() for k in ("values", "secrets_set", "status", "redirect_uris"): assert k in d, f"missing {k}" assert "GOOGLE_CLIENT_ID" in d["values"] assert "MS_TENANT" in d["values"] assert "OAUTH_PUBLIC_BASE" in d["values"] # secrets must only be exposed as booleans for sk in ("GOOGLE_CLIENT_SECRET", "MS_CLIENT_SECRET", "RAZORPAY_KEY_SECRET", "RAZORPAY_WEBHOOK_SECRET"): assert sk in d["secrets_set"] assert isinstance(d["secrets_set"][sk], bool) assert sk not in d["values"] assert set(d["status"]) == {"google", "microsoft", "razorpay"} assert d["redirect_uris"]["google"].endswith("/api/oauth/google/callback") assert d["redirect_uris"]["microsoft"].endswith("/api/oauth/microsoft/callback") def test_non_admin_forbidden(self, non_admin): h = {"Authorization": f"Bearer {non_admin}"} assert requests.get(f"{API}/admin/integrations", headers=h, timeout=30).status_code == 403 assert requests.put(f"{API}/admin/integrations", headers=h, json={"MS_TENANT": "hacked"}, timeout=30).status_code == 403 def test_unauthenticated_rejected(self): assert requests.get(f"{API}/admin/integrations", timeout=30).status_code in (401, 403) def test_put_saves_value_and_masks_secret(self, client): r = client.put(f"{API}/admin/integrations", json={ "GOOGLE_CLIENT_ID": "TEST_google_client_id.apps.googleusercontent.com", "GOOGLE_CLIENT_SECRET": "TEST_google_secret_value", "MS_TENANT": "TEST_tenant_abc", }, timeout=30) assert r.status_code == 200, r.text d = r.json() assert d["values"]["GOOGLE_CLIENT_ID"] == "TEST_google_client_id.apps.googleusercontent.com" assert d["values"]["MS_TENANT"] == "TEST_tenant_abc" assert d["secrets_set"]["GOOGLE_CLIENT_SECRET"] is True assert "TEST_google_secret_value" not in r.text # never returned in plaintext # GET verifies persistence g = client.get(f"{API}/admin/integrations", timeout=30).json() assert g["values"]["GOOGLE_CLIENT_ID"] == "TEST_google_client_id.apps.googleusercontent.com" assert g["values"]["MS_TENANT"] == "TEST_tenant_abc" assert g["secrets_set"]["GOOGLE_CLIENT_SECRET"] is True assert g["status"]["google"] is True def test_secret_stored_encrypted_in_db(self): rows = _db_exec("SELECT value FROM app_settings WHERE `key`='GOOGLE_CLIENT_SECRET'") assert rows, "secret not persisted" assert "TEST_google_secret_value" not in str(rows[0][0]) def test_empty_secret_keeps_existing(self, client): r = client.put(f"{API}/admin/integrations", json={"GOOGLE_CLIENT_SECRET": "", "MS_TENANT": "TEST_tenant_xyz"}, timeout=30) assert r.status_code == 200, r.text d = r.json() assert d["secrets_set"]["GOOGLE_CLIENT_SECRET"] is True, "empty secret wiped existing secret" assert d["values"]["MS_TENANT"] == "TEST_tenant_xyz" def test_partial_update_does_not_clear_others(self, client): r = client.put(f"{API}/admin/integrations", json={"MS_CLIENT_ID": "TEST_ms_client"}, timeout=30) assert r.status_code == 200 d = r.json() assert d["values"]["MS_CLIENT_ID"] == "TEST_ms_client" assert d["values"]["GOOGLE_CLIENT_ID"] == "TEST_google_client_id.apps.googleusercontent.com" def test_oauth_public_base_updates_redirect_uris(self, client): original = client.get(f"{API}/admin/integrations", timeout=30).json()["values"]["OAUTH_PUBLIC_BASE"] try: r = client.put(f"{API}/admin/integrations", json={"OAUTH_PUBLIC_BASE": "https://qa-test.example.com"}, timeout=30) assert r.status_code == 200, r.text d = r.json() assert d["redirect_uris"]["google"] == \ "https://qa-test.example.com/api/oauth/google/callback" assert d["redirect_uris"]["microsoft"] == \ "https://qa-test.example.com/api/oauth/microsoft/callback" finally: back = client.put(f"{API}/admin/integrations", json={"OAUTH_PUBLIC_BASE": original or ""}, timeout=30) assert back.status_code == 200 assert "qa-test.example.com" not in back.text def test_unmanaged_key_ignored(self, client): r = client.put(f"{API}/admin/integrations", json={"MS_TENANT": "TEST_tenant_final", "SECRET_KEY": "pwn"}, timeout=30) assert r.status_code in (200, 422), r.text if r.status_code == 200: assert "SECRET_KEY" not in r.json()["values"] rows = _db_exec("SELECT COUNT(*) FROM app_settings WHERE `key`='SECRET_KEY'") assert rows[0][0] == 0 # ---------- Migration report (PDF/CSV) ---------- class TestReport: """GET /api/migrations/{id}/report?fmt=pdf|csv""" @pytest.fixture(scope="class") def completed_mig(self, client, created_ids): r = client.get(f"{API}/migrations", timeout=30) assert r.status_code == 200, r.text done = [m for m in r.json() if m["status"] == "completed"] if done: return done[0]["id"] payload = mig_payload("TEST_report_src") payload["selected_folders"] = ["INBOX"] c = client.post(f"{API}/migrations", json=payload, timeout=60) assert c.status_code in (200, 201), c.text mid = c.json()["id"] created_ids.append(mid) client.post(f"{API}/migrations/{mid}/start", timeout=30) assert _wait_status(client, mid)["status"] == "completed" return mid def test_pdf_report(self, client, completed_mig): r = client.get(f"{API}/migrations/{completed_mig}/report", params={"fmt": "pdf"}, timeout=60) assert r.status_code == 200, r.text[:300] assert r.headers["content-type"].startswith("application/pdf") assert r.content[:5] == b"%PDF-", r.content[:20] assert len(r.content) > 1000 assert "attachment" in r.headers.get("content-disposition", "") assert "-report.pdf" in r.headers.get("content-disposition", "") def test_pdf_is_default_format(self, client, completed_mig): r = client.get(f"{API}/migrations/{completed_mig}/report", timeout=60) assert r.status_code == 200 assert r.content[:5] == b"%PDF-" def test_csv_report(self, client, completed_mig): r = client.get(f"{API}/migrations/{completed_mig}/report", params={"fmt": "csv"}, timeout=60) assert r.status_code == 200, r.text[:300] assert r.headers["content-type"].startswith("text/csv") body = r.text for token in ("Migration Report", "Status", "Emails migrated", "Folder", "Data transferred"): assert token in body, f"missing '{token}' in CSV" assert "-report.csv" in r.headers.get("content-disposition", "") # folder rows present lines = [l for l in body.splitlines() if l.strip()] hdr = [i for i, l in enumerate(lines) if l.startswith("Folder,Migrated,Total,Status")] assert hdr, "folder breakdown header missing" assert len(lines) > hdr[0] + 1, "no folder rows in CSV" def test_unknown_id_404(self, client): r = client.get(f"{API}/migrations/{uuid.uuid4()}/report", params={"fmt": "csv"}, timeout=30) assert r.status_code == 404, r.status_code def test_other_user_cannot_download(self, completed_mig): email = f"TEST_rep_{uuid.uuid4().hex[:8]}@test.com" reg = requests.post(f"{API}/auth/register", json={"email": email, "password": "Passw0rd!123", "name": "TEST rep"}, timeout=30) assert reg.status_code in (200, 201), reg.text h = {"Authorization": f"Bearer {reg.json()['token']}"} r = requests.get(f"{API}/migrations/{completed_mig}/report", headers=h, params={"fmt": "csv"}, timeout=30) assert r.status_code in (403, 404), f"leaked report to non-owner: {r.status_code}" def test_report_requires_auth(self, completed_mig): r = requests.get(f"{API}/migrations/{completed_mig}/report", timeout=30) assert r.status_code in (401, 403) # ---------- Date-range regression over local dovecot ---------- class TestDateRange: """POST /api/migrations with date_from/date_to must apply IMAP SINCE/BEFORE.""" def test_range_migrates_only_three(self, client, created_ids): _purge_dest() assert _dest_email_count() == 0 p = mig_payload("TEST_daterange_3") p["selected_folders"] = ["INBOX"] p["date_from"] = "2024-01-01" p["date_to"] = "2024-06-30" r = client.post(f"{API}/migrations", json=p, timeout=60) assert r.status_code in (200, 201), r.text mid = r.json()["id"] created_ids.append(mid) assert client.post(f"{API}/migrations/{mid}/start", timeout=30).status_code == 200 d = _wait_status(client, mid) assert d["status"] == "completed", d assert d["migrated_emails"] == 3, f"expected 3 in-range emails, got {d['migrated_emails']}" assert d["date_from"] == "2024-01-01" and d["date_to"] == "2024-06-30" assert _dest_email_count() == 3 def test_no_range_migrates_all_five(self, client, created_ids): _purge_dest() assert _dest_email_count() == 0 p = mig_payload("TEST_daterange_all") p["selected_folders"] = ["INBOX"] r = client.post(f"{API}/migrations", json=p, timeout=60) assert r.status_code in (200, 201), r.text mid = r.json()["id"] created_ids.append(mid) assert client.post(f"{API}/migrations/{mid}/start", timeout=30).status_code == 200 d = _wait_status(client, mid) assert d["status"] == "completed", d assert d["migrated_emails"] == 5, f"expected 5 emails, got {d['migrated_emails']}" assert _dest_email_count() == 5 def test_only_date_from(self, client, created_ids): _purge_dest() p = mig_payload("TEST_daterange_from") p["selected_folders"] = ["INBOX"] p["date_from"] = "2025-01-01" r = client.post(f"{API}/migrations", json=p, timeout=60) assert r.status_code in (200, 201), r.text mid = r.json()["id"] created_ids.append(mid) client.post(f"{API}/migrations/{mid}/start", timeout=30) d = _wait_status(client, mid) assert d["status"] == "completed", d assert d["migrated_emails"] == 1, f"expected 1 email since 2025-01-01, got {d['migrated_emails']}"