"""MailShift backend API regression tests.""" import time import uuid import pytest import requests from conftest import API, mig_payload # ---------- Health ---------- class TestHealth: def test_root(self, api_client): r = api_client.get(f"{API}/", timeout=30) assert r.status_code == 200 assert "MailShift" in r.json().get("message", "") # ---------- Auth ---------- class TestAuth: def test_login_success(self, api_client, test_credentials): r = api_client.post(f"{API}/auth/login", json=test_credentials, timeout=30) assert r.status_code == 200, r.text d = r.json() assert isinstance(d["token"], str) and len(d["token"]) > 20 assert d["user"]["email"] == test_credentials["email"] assert d["user"]["role"] == "admin" assert "password_hash" not in d["user"] def test_login_bad_password(self, api_client, test_credentials): r = api_client.post(f"{API}/auth/login", json={"email": test_credentials["email"], "password": "wrong-xyz"}, timeout=30) assert r.status_code == 401 assert "detail" in r.json() def test_login_unknown_user(self, api_client): r = api_client.post(f"{API}/auth/login", json={"email": "nobody_qa@example.com", "password": "whatever"}, timeout=30) assert r.status_code == 401 def test_login_invalid_email_format(self, api_client): r = api_client.post(f"{API}/auth/login", json={"email": "notanemail", "password": "x"}, timeout=30) assert r.status_code == 422 def test_register_and_login(self, api_client): email = f"TEST_qa_{uuid.uuid4().hex[:8]}@example.com" r = api_client.post(f"{API}/auth/register", json={"email": email, "password": "secret123", "name": "QA User"}, timeout=30) assert r.status_code == 200, r.text d = r.json() assert d["user"]["email"] == email.lower() assert d["user"]["role"] == "user" token = d["token"] # /auth/me works with returned token me = api_client.get(f"{API}/auth/me", headers={"Authorization": f"Bearer {token}"}, timeout=30) assert me.status_code == 200 assert me.json()["email"] == email.lower() # login with same creds li = api_client.post(f"{API}/auth/login", json={"email": email, "password": "secret123"}, timeout=30) assert li.status_code == 200 # duplicate registration blocked dup = api_client.post(f"{API}/auth/register", json={"email": email, "password": "secret123", "name": "x"}, timeout=30) assert dup.status_code == 400 def test_register_short_password(self, api_client): r = api_client.post(f"{API}/auth/register", json={"email": f"TEST_s{uuid.uuid4().hex[:6]}@example.com", "password": "123", "name": "x"}, timeout=30) assert r.status_code == 422 def test_bcrypt_hash_format(self, test_credentials): import os import pymysql from dotenv import dotenv_values env = dotenv_values("/app/backend/.env") conn = pymysql.connect(host=env["MYSQL_HOST"], port=int(env["MYSQL_PORT"]), user=env["MYSQL_USER"], password=env["MYSQL_PASSWORD"], database=env["MYSQL_DB"]) try: with conn.cursor() as cur: cur.execute("SELECT password_hash FROM users WHERE email=%s", (test_credentials["email"],)) row = cur.fetchone() assert row, "admin user not seeded" assert row[0].startswith("$2b$"), f"unexpected hash prefix: {row[0][:6]}" finally: conn.close() # ---------- Auth guard ---------- class TestAuthGuard: PROTECTED = [ ("get", "/providers"), ("get", "/stats"), ("get", "/migrations"), ("get", "/auth/me"), ("post", "/test-connection"), ("get", "/migrations/does-not-exist"), ("get", "/migrations/x/folders"), ("get", "/migrations/x/logs"), ("post", "/migrations/x/start"), ("post", "/migrations/x/pause"), ("post", "/migrations/x/resume"), ("post", "/migrations/x/cancel"), ("delete", "/migrations/x"), ("post", "/migrations"), ("post", "/migrations/bulk"), ] @pytest.mark.parametrize("method,path", PROTECTED) def test_requires_auth(self, method, path): r = requests.request(method, f"{API}{path}", json={}, timeout=30) assert r.status_code == 401, f"{method} {path} -> {r.status_code}" def test_invalid_token(self, api_client): r = api_client.get(f"{API}/auth/me", headers={"Authorization": "Bearer garbage.token.x"}, timeout=30) assert r.status_code == 401 # ---------- Providers ---------- class TestProviders: def test_provider_presets(self, client): r = client.get(f"{API}/providers", timeout=30) assert r.status_code == 200 data = r.json() assert len(data) >= 5 ids = {p["id"] for p in data} assert {"gmail", "office365", "custom"}.issubset(ids) gmail = next(p for p in data if p["id"] == "gmail") assert gmail["host"] == "imap.gmail.com" assert gmail["port"] == 993 assert gmail["ssl"] is True # ---------- Test connection ---------- class TestConnection: def test_connection_success_local_imap(self, client): r = client.post(f"{API}/test-connection", json={"host": "127.0.0.1", "port": 143, "email": "user1", "password": "pass1", "ssl": False}, timeout=60) assert r.status_code == 200, r.text d = r.json() assert d["ok"] is True, d assert d["folder_count"] >= 2, d def test_connection_bad_password(self, client): r = client.post(f"{API}/test-connection", json={"host": "127.0.0.1", "port": 143, "email": "user1", "password": "WRONG", "ssl": False}, timeout=60) assert r.status_code == 200 d = r.json() assert d["ok"] is False assert d["message"] def test_connection_bad_host(self, client): r = client.post(f"{API}/test-connection", json={"host": "no-such-host.invalid", "port": 993, "email": "a", "password": "b", "ssl": True}, timeout=90) assert r.status_code == 200, r.text d = r.json() assert d["ok"] is False assert isinstance(d["message"], str) and len(d["message"]) > 0 # ---------- Migration CRUD + real migration ---------- class TestMigrations: def test_create_get_and_persist(self, client, created_ids): payload = mig_payload("TEST_crud_mig") r = client.post(f"{API}/migrations", json=payload, timeout=60) assert r.status_code == 200, r.text m = r.json() assert isinstance(m["id"], str) created_ids.append(m["id"]) assert m["name"] == "TEST_crud_mig" assert m["status"] == "pending" assert m["source_email"] == "user1" assert m["dest_email"] == "user2" assert m["source_ssl"] is False assert "source_password" not in m and "source_password_enc" not in m assert "_id" not in m g = client.get(f"{API}/migrations/{m['id']}", timeout=30) assert g.status_code == 200 gd = g.json() assert gd["name"] == "TEST_crud_mig" assert gd["is_running"] is False lst = client.get(f"{API}/migrations", timeout=30) assert lst.status_code == 200 assert m["id"] in [x["id"] for x in lst.json()] def test_get_unknown_migration_404(self, client): r = client.get(f"{API}/migrations/{uuid.uuid4()}", timeout=30) assert r.status_code == 404 def test_create_validation_error(self, client): r = client.post(f"{API}/migrations", json={"name": "x"}, timeout=30) assert r.status_code == 422 def test_real_migration_end_to_end(self, client, created_ids): r = client.post(f"{API}/migrations", json=mig_payload("TEST_e2e_mig"), timeout=60) assert r.status_code == 200, r.text mid = r.json()["id"] created_ids.append(mid) s = client.post(f"{API}/migrations/{mid}/start", timeout=30) assert s.status_code == 200, s.text # worker-pool queue: start now reports 'queued' (worker picks it up immediately) assert s.json()["status"] in ("queued", "running") # duplicate start rejected time.sleep(0.3) dup = client.post(f"{API}/migrations/{mid}/start", timeout=30) assert dup.status_code in (400, 200) status, data = None, {} for _ in range(60): time.sleep(2) g = client.get(f"{API}/migrations/{mid}", timeout=30) assert g.status_code == 200 data = g.json() status = data["status"] if status in ("completed", "failed"): break assert status == "completed", f"final status={status} error={data.get('error')}" assert data["total_folders"] == 2, data assert data["migrated_folders"] == 2, data assert data["total_emails"] == 8, data assert data["migrated_emails"] == 8, data assert data["failed_emails"] == 0, data assert data["completed_at"] is not None assert data["started_at"] is not None f = client.get(f"{API}/migrations/{mid}/folders", timeout=30) assert f.status_code == 200 folders = f.json() names = {x["name"] for x in folders} assert "INBOX" in names, names assert any("Work" in n for n in names), names inbox = next(x for x in folders if x["name"] == "INBOX") assert inbox["total"] == 5 and inbox["migrated"] == 5, inbox assert all(x["status"] == "done" for x in folders), folders lg = client.get(f"{API}/migrations/{mid}/logs", timeout=30) assert lg.status_code == 200 logs = lg.json() assert len(logs) > 3 assert any("Migration completed successfully" in x["message"] for x in logs), logs[-3:] # incremental fetch (after=) last_id = logs[-1]["id"] inc = client.get(f"{API}/migrations/{mid}/logs", params={"after": last_id}, timeout=30) assert inc.status_code == 200 assert inc.json() == [] st = client.get(f"{API}/stats", timeout=30) assert st.status_code == 200 sd = st.json() assert sd["completed"] >= 1 assert sd["emails_migrated"] >= 8 def test_delete_and_verify_removal(self, client): r = client.post(f"{API}/migrations", json=mig_payload("TEST_del_mig"), timeout=60) mid = r.json()["id"] d = client.delete(f"{API}/migrations/{mid}", timeout=30) assert d.status_code == 200 g = client.get(f"{API}/migrations/{mid}", timeout=30) assert g.status_code == 404 def test_pause_cancel_state(self, client, created_ids): r = client.post(f"{API}/migrations", json=mig_payload("TEST_pause_mig"), timeout=60) mid = r.json()["id"] created_ids.append(mid) p = client.post(f"{API}/migrations/{mid}/pause", timeout=30) assert p.status_code == 200 and p.json()["status"] == "paused" assert client.get(f"{API}/migrations/{mid}", timeout=30).json()["status"] == "paused" c = client.post(f"{API}/migrations/{mid}/cancel", timeout=30) assert c.status_code == 200 assert client.get(f"{API}/migrations/{mid}", timeout=30).json()["status"] == "failed" def test_bulk_create(self, client, created_ids): body = {"migrations": [mig_payload("TEST_bulk_A"), mig_payload("TEST_bulk_B")], "auto_start": False} r = client.post(f"{API}/migrations/bulk", json=body, timeout=60) assert r.status_code == 200, r.text d = r.json() assert d["created"] == 2 assert len(d["ids"]) == 2 created_ids.extend(d["ids"]) for mid in d["ids"]: g = client.get(f"{API}/migrations/{mid}", timeout=30) assert g.status_code == 200 assert g.json()["status"] == "pending" # ---------- Ownership isolation ---------- class TestOwnership: def test_other_user_cannot_access(self, client, created_ids): r = client.post(f"{API}/migrations", json=mig_payload("TEST_owner_mig"), timeout=60) mid = r.json()["id"] created_ids.append(mid) email = f"TEST_other_{uuid.uuid4().hex[:8]}@example.com" reg = requests.post(f"{API}/auth/register", json={"email": email, "password": "secret123", "name": "Other"}, timeout=30) assert reg.status_code == 200 other = {"Authorization": f"Bearer {reg.json()['token']}"} assert requests.get(f"{API}/migrations/{mid}", headers=other, timeout=30).status_code == 404 assert requests.delete(f"{API}/migrations/{mid}", headers=other, timeout=30).status_code == 404 assert requests.get(f"{API}/migrations", headers=other, timeout=30).json() == []