""" Full-stack backup E2E: HTTP submit (PDF → Postgres + MinIO) then admin ZIP download. Requires PostgreSQL with migrations through **009** and reachable **MinIO (S3 API)**. Run (host → docker-compose ports): export INITIATIVE_DATABASE_URL="postgresql+asyncpg://initiative:initiative_secret@127.0.0.1:15432/initiatives" export S3_ENDPOINT_URL="http://127.0.0.1:19000" export S3_PUBLIC_ENDPOINT_URL="http://127.0.0.1:19000" export S3_ACCESS_KEY="minio_user" export S3_SECRET_KEY="minio_password" export S3_BUCKET_ATTACHMENTS="initiative-attachments" export S3_BUCKET_EXPORTS="initiative-exports" export S3_BUCKET_QUARANTINE="initiative-quarantine" export E2E_BACKUP=1 cd be0 && python -m unittest tests.test_backup_e2e -v Browser E2E: see ``fe0/e2e/backup-admin-download.spec.ts`` (requires the same stack plus ``fe0`` + ``E2E_ADMIN_EMAIL`` on ``AUTH_ADMIN_EMAILS``). """ from __future__ import annotations import io import json import os import unittest import uuid import zipfile from unittest.mock import patch from sqlalchemy import select from tests.auth_register_staff_fixture import register_staff_fields _RUN_DB = os.getenv("INITIATIVE_DATABASE_URL", "").strip().lower().startswith("postgresql") _S3_KEYS = ( "S3_ENDPOINT_URL", "S3_ACCESS_KEY", "S3_SECRET_KEY", "S3_BUCKET_ATTACHMENTS", "S3_BUCKET_EXPORTS", "S3_BUCKET_QUARANTINE", ) _HAS_S3 = all(os.getenv(k, "").strip() for k in _S3_KEYS) _RUN_BACKUP = os.getenv("E2E_BACKUP", "").strip().lower() in ("1", "true", "yes") _TEST_PASSWORD = "Testpass1!" _MIN_PDF = b"%PDF-1.4\n%\xe2\xe3\xcf\xd3\n1 0 obj<<>>endobj\ntrailer<<>>\n%%EOF\n" + b"0" * 120 @unittest.skipUnless( _RUN_DB and _HAS_S3 and _RUN_BACKUP, "Need INITIATIVE_DATABASE_URL, full S3_* env, and E2E_BACKUP=1 (see module docstring).", ) class BackupFullStackApiE2ETests(unittest.IsolatedAsyncioTestCase): async def asyncSetUp(self) -> None: from src.initiative_db import engine as eng await eng.dispose_engine() await eng.init_engine() async def asyncTearDown(self) -> None: from src.initiative_db import engine as eng await eng.dispose_engine() async def _delete_users_by_email(self, emails: list[str]) -> None: from src.initiative_db.engine import get_session from src.initiative_db.models import Initiative, User async with get_session() as session: for email in emails: user = ( await session.execute(select(User).where(User.email == email)) ).scalar_one_or_none() if user is None: continue inis = ( await session.execute(select(Initiative).where(Initiative.owner_id == user.id)) ).scalars().all() for ini in inis: await session.delete(ini) await session.flush() await session.delete(user) await session.commit() async def test_submit_creates_minio_full_pdf_then_backup_zip(self) -> None: from fastapi.testclient import TestClient from main import app from src.initiative_db.engine import get_session from src.initiative_db.models import ApplicationArtifact, Initiative applicant_email = f"e2e-backup-app-{uuid.uuid4().hex[:10]}@ump.edu.vn" admin_email = f"e2e-backup-adm-{uuid.uuid4().hex[:10]}@ump.edu.vn" try: with patch.dict(os.environ, {"AUTH_MAIL_LOG_ONLY": "1"}): with TestClient(app) as client: cap_app: list[str] = [] async def grab_app(_t: str, raw: str) -> None: cap_app.append(raw) with patch("src.auth_api.deliver_registration_otp_email", side_effect=grab_app): r = client.post( "/api/v1/auth/register", json={ "fullName": "E2E Applicant", "email": applicant_email, "password": _TEST_PASSWORD, "passwordConfirm": _TEST_PASSWORD, **register_staff_fields(), }, ) self.assertEqual(r.status_code, 200, r.text) self.assertTrue(cap_app) r = client.post( "/api/v1/auth/verify-otp", json={"email": applicant_email, "otp": cap_app[0]}, ) self.assertEqual(r.status_code, 200, r.text) r = client.post( "/api/v1/auth/login", json={"email": applicant_email, "password": _TEST_PASSWORD}, ) self.assertEqual(r.status_code, 200, r.text) applicant_token = r.json()["accessToken"] r = client.post( "/api/applications/new", headers={"Authorization": f"Bearer {applicant_token}"}, json={"name": "E2E backup row"}, ) self.assertEqual(r.status_code, 200, r.text) shell = r.json().get("application") or {} case_id = str(shell.get("draft_case_id") or "").strip() application_id = str(r.json().get("id") or shell.get("id") or "").strip() self.assertTrue(case_id, shell) self.assertTrue(application_id, shell) from tests.fixtures.minimal_submit_bundle import minimal_tabs_bundle bundle = minimal_tabs_bundle(initiative_name="E2E Backup Initiative") for tab_name in ("report", "application", "contribution"): r = client.post( "/api/v1/application-drafts", headers={"Authorization": f"Bearer {applicant_token}"}, json={"caseId": case_id, "tab": tab_name, "data": bundle[tab_name]}, ) self.assertEqual(r.status_code, 200, r.text) r = client.post( f"/api/v1/application-drafts/{case_id}/evidence", headers={"Authorization": f"Bearer {applicant_token}"}, data={"kind": "technical"}, files={"file": ("mc.pdf", io.BytesIO(_MIN_PDF), "application/pdf")}, ) self.assertEqual(r.status_code, 200, r.text) meta = { "initiativeCaseId": case_id, "initiativeName": "E2E Backup Initiative", "authorName": "Applicant", "authorEmail": applicant_email, "subjectId": "s1", "groupId": "g1", "topicType": "Hồ sơ PDF", } r = client.post( "/api/applications/submit", headers={"Authorization": f"Bearer {applicant_token}"}, files={"file": ("e2e.pdf", io.BytesIO(_MIN_PDF), "application/pdf")}, data={"metadata": json.dumps(meta)}, ) self.assertEqual(r.status_code, 200, r.text, r.text) application_id = str((r.json() or {}).get("id") or application_id) with patch.dict(os.environ, {"AUTH_ADMIN_EMAILS": admin_email}): cap_adm: list[str] = [] async def grab_adm(_t: str, raw: str) -> None: cap_adm.append(raw) with patch( "src.auth_api.deliver_registration_otp_email", side_effect=grab_adm, ): r = client.post( "/api/v1/auth/register", json={ "fullName": "E2E Admin", "email": admin_email, "password": _TEST_PASSWORD, "passwordConfirm": _TEST_PASSWORD, **register_staff_fields(), }, ) self.assertEqual(r.status_code, 200, r.text) self.assertTrue(cap_adm) self.assertIn("admin", r.json()["user"]["roles"]) r = client.post( "/api/v1/auth/verify-otp", json={"email": admin_email, "otp": cap_adm[0]}, ) self.assertEqual(r.status_code, 200, r.text) r = client.post( "/api/v1/auth/login", json={"email": admin_email, "password": _TEST_PASSWORD}, ) self.assertEqual(r.status_code, 200, r.text) admin_token = r.json()["accessToken"] r = client.get( f"/api/applications/{application_id}/backup", headers={"Authorization": f"Bearer {admin_token}"}, ) self.assertEqual(r.status_code, 200, r.text) self.assertEqual(r.headers.get("content-type", "").split(";")[0], "application/zip") buf = io.BytesIO(r.content) with zipfile.ZipFile(buf, "r") as zf: names = zf.namelist() self.assertIn("manifest.json", names) self.assertIn("submitted/full-package.pdf", names) manifest = json.loads(zf.read("manifest.json").decode("utf-8")) self.assertEqual(str(manifest.get("applicationId")), application_id) self.assertIn("initiative_id", manifest) packed = {str(x.get("zip_path")): x for x in manifest.get("files") or []} self.assertIn("submitted/full-package.pdf", packed) self.assertFalse(packed["submitted/full-package.pdf"].get("skipped")) async with get_session() as session: ini = ( await session.execute(select(Initiative).where(Initiative.case_code == case_id)) ).scalar_one() row = ( await session.execute( select(ApplicationArtifact).where( ApplicationArtifact.initiative_id == ini.id, ApplicationArtifact.role == "full_pdf", ) ) ).scalar_one_or_none() self.assertIsNotNone(row) assert row is not None uri = (row.storage_uri or "").strip() self.assertTrue(uri.startswith("initiatives/"), uri) self.assertEqual(row.storage_kind, "minio_exports") finally: await self._delete_users_by_email([applicant_email, admin_email]) if __name__ == "__main__": unittest.main()