"""Cut over existing generated repos from a local user store to Cloud auth.""" from __future__ import annotations import tempfile import zipfile from pathlib import Path from uuid import uuid4 import pytest from src.domain.codegen.auth_users_adapter import ( auth_users_adapter_files, install_auth_users_adapter, ) from src.domain.codegen.legacy_cloud_cutover import ( CutoverReport, cut_over_generated_repo, import_users_to_cloud, ) from src.domain.codegen.legacy_user_store import DiscoveredUser, scan_generated_repo LOCAL_USERS_ROUTE = """ export default defineEventHandler(async (event) => { const users = [ { id: "usr_seed_muhammad_zakir_pie_solutions_com", email: "muhammad.zakir@pie-solutions.com", name: "Muhammad Zakir", role: "super_admin", password: "not-copied", is_active: true, }, { id: "usr_d7becd1b-1b6c-43a5-ad11-2d2a8f31a4e1", email: "mohsin@pie-solutions.com", name: "Mohsin", role: "admin", is_active: true, }, ]; return { success: true, data: users }; }); """ def _write_nuxt_store(root: Path) -> None: route = root / "server/api/admin/users.ts" route.parent.mkdir(parents=True, exist_ok=True) route.write_text(LOCAL_USERS_ROUTE, encoding="utf-8") def test_scan_finds_seeded_admin_users(): with tempfile.TemporaryDirectory() as raw: root = Path(raw) _write_nuxt_store(root) scan = scan_generated_repo(str(root)) emails = {user.email: user.role for user in scan.users} assert emails["muhammad.zakir@pie-solutions.com"] == "super_admin" assert emails["mohsin@pie-solutions.com"] == "admin" assert "server/api/admin/users.ts" in scan.local_route_files assert scan.password_store_files def test_scan_reads_backed_up_local_route(): with tempfile.TemporaryDirectory() as raw: root = Path(raw) _write_nuxt_store(root) install_auth_users_adapter(str(root), "nuxt_app", replace_local=True) scan = scan_generated_repo(str(root)) emails = {user.email: user.role for user in scan.users} assert emails["mohsin@pie-solutions.com"] == "admin" assert emails["muhammad.zakir@pie-solutions.com"] == "super_admin" def test_install_backs_up_local_route_and_writes_cloud_adapter(): with tempfile.TemporaryDirectory() as raw: root = Path(raw) _write_nuxt_store(root) result = install_auth_users_adapter( str(root), "nuxt_app", replace_local=True, ) written = (root / "server/api/admin/users.ts").read_text(encoding="utf-8") backup = (root / ".p2hs/local-bak/server/api/admin/users.ts").read_text( encoding="utf-8", ) records = (root / "functions/_lib/p2hs-records.mjs").is_file() auth_util = (root / "server/utils/p2hs-auth-users.mjs").is_file() assert "server/api/admin/users.ts" in result["written"] assert any(path.endswith("server/api/admin/users.ts") and "local-bak" in path for path in result["backed_up"]) assert "p2hs-auth-users" in written assert "usr_seed_" in backup assert records assert auth_util def test_install_backs_up_homemade_forgot_password_route(): homemade = ( 'export default defineEventHandler(async () => ' '({ success: true, message: "If an account exists" }));\n' ) with tempfile.TemporaryDirectory() as raw: root = Path(raw) route = root / "server/api/auth/forgot-password.ts" route.parent.mkdir(parents=True, exist_ok=True) route.write_text(homemade, encoding="utf-8") result = install_auth_users_adapter(str(root), "nuxt_app", replace_local=True) written = (root / "server/api/auth/forgot-password.ts").read_text(encoding="utf-8") backup = (root / ".p2hs/local-bak/server/api/auth/forgot-password.ts").read_text( encoding="utf-8", ) adapter = (root / "server/utils/p2hs-auth-password.mjs").read_text(encoding="utf-8") assert "server/api/auth/forgot-password.ts" in result["written"] assert any(path.endswith("server/api/auth/forgot-password.ts") and "local-bak" in path for path in result["backed_up"]) assert "p2hs-auth-password" in written assert "auth/password/reset-request" in adapter assert homemade.strip() in backup def test_install_without_replace_keeps_app_owned_route(): with tempfile.TemporaryDirectory() as raw: root = Path(raw) _write_nuxt_store(root) result = install_auth_users_adapter( str(root), "nuxt_app", replace_local=False, ) body = (root / "server/api/admin/users.ts").read_text(encoding="utf-8") assert "server/api/admin/users.ts" in result["skipped"] assert "usr_seed_" in body @pytest.mark.parametrize("project_type", ["html", "react_app", "nuxt_app"]) def test_auth_adapter_files_cover_pages_starters(project_type): files = auth_users_adapter_files(project_type) assert "functions/_lib/p2hs-auth-users.mjs" in files assert "functions/_lib/p2hs-auth-password.mjs" in files assert "functions/api/admin/users.js" in files assert "functions/api/auth/forgot-password.js" in files assert "functions/api/auth/reset-password.js" in files if project_type == "nuxt_app": assert "server/api/admin/users.ts" in files assert "server/api/auth/forgot-password.ts" in files assert "server/api/auth/reset-password.ts" in files assert "p2hs-auth-password" in files["server/api/auth/forgot-password.ts"] else: assert "server/api/admin/users.ts" not in files assert "server/api/auth/forgot-password.ts" not in files def test_generated_user_adapter_uses_activation_links_not_seeded_passwords(): files = auth_users_adapter_files("nuxt_app") adapter = files["server/utils/p2hs-auth-users.mjs"] assert "Do not send passwords when creating users" in adapter assert "email_verified: false" in adapter assert "requestPasswordReset" in adapter assert "activation_pending" in adapter class _FakeCloud: def __init__(self) -> None: self.upserts: list[dict] = [] self.records: dict[str, list[dict]] = {} async def upsert_app_user( self, project_id, *, email, role, email_verified=True, user_metadata=None, ): row = { "project_id": project_id, "email": email, "role": role, "email_verified": email_verified, "user_metadata": user_metadata or {}, } self.upserts.append(row) return {"id": f"cloud-{email}", **row} async def list_app_users(self, project_id): return list(self.upserts) async def create_entity_records(self, project_id, entity_key, records): created = [] for row in records: item = {"id": f"cloud-{entity_key}-{len(self.records.setdefault(entity_key, [])) + 1}", **row} self.records[entity_key].append(item) created.append(item) return created async def list_entity_records(self, project_id, entity_key): return list(self.records.get(entity_key, [])) async def preserve_legacy_app_user(self, project_id, user, legacy): row = next(u for u in self.upserts if u["email"] == user["email"]) row.update( app_metadata={"p2hs_legacy_user": legacy}, status="active" if legacy["is_active"] else "banned", ) return {**user, **row} @pytest.mark.asyncio async def test_import_upserts_email_and_role_not_password(): cloud = _FakeCloud() users = [ DiscoveredUser("mohsin@pie-solutions.com", "admin", "Mohsin", "cli"), ] results = await import_users_to_cloud(cloud, "proj-1", users, dry_run=False) assert results[0].status == "upserted" assert cloud.upserts[0]["email"] == "mohsin@pie-solutions.com" assert cloud.upserts[0]["role"] == "admin" assert "password" not in cloud.upserts[0] class _FakeProject: def __init__(self) -> None: self.id = uuid4() self.workspace_id = uuid4() self.name = "Student Consulting App" self.type = "nuxt_app" @pytest.mark.asyncio async def test_cutover_apply_imports_and_replaces_route(): with tempfile.TemporaryDirectory() as raw: root = Path(raw) / "checkout" _write_nuxt_store(root) cloud = _FakeCloud() project = _FakeProject() report = await cut_over_generated_repo( db=None, project=project, repo_path=str(root), import_dry_run=False, adapter_dry_run=False, ensure_auth=False, extra_users=[DiscoveredUser("extra@example.com", "member", "", "cli")], cloud=cloud, ) route = (root / "server/api/admin/users.ts").read_text(encoding="utf-8") backup = (root / ".p2hs/local-bak/server/api/admin/users.ts").is_file() with zipfile.ZipFile(report.source_archive) as archive: assert "usr_seed_" in archive.read("server/api/admin/users.ts").decode() assert isinstance(report, CutoverReport) emails = {item.email for item in report.imported} assert "mohsin@pie-solutions.com" in emails assert "extra@example.com" in emails assert all(item.status == "upserted" for item in report.imported) assert "p2hs-auth-users" in route assert not backup @pytest.mark.asyncio async def test_cutover_preserves_routes_when_ensure_auth_fails(monkeypatch): class _Boom: def __init__(self, **kwargs): pass async def ensure_default_auth(self, project_id): raise RuntimeError("type authkeystatus does not exist") monkeypatch.setattr( "src.domain.codegen.legacy_cloud_cutover.ProjectAuthService", _Boom, ) with tempfile.TemporaryDirectory() as raw: root = Path(raw) / "checkout" _write_nuxt_store(root) with pytest.raises(ValueError, match="Cloud auth setup failed"): await cut_over_generated_repo( db=None, project=_FakeProject(), repo_path=str(root), import_dry_run=True, adapter_dry_run=False, ensure_auth=True, cloud=_FakeCloud(), ) route = (root / "server/api/admin/users.ts").read_text(encoding="utf-8") assert "usr_seed_" in route def test_auth_key_status_uses_migrated_pg_enum(): from src.domain.project_auth.models import ProjectAuthKey, ProjectAuthRole assert ProjectAuthKey.__table__.c.status.type.name == "p2hs_project_auth_key_status" assert ProjectAuthRole.__table__.c.created_by.type.name == "p2hs_project_auth_role_creator" @pytest.mark.asyncio @pytest.mark.parametrize("failure", ["empty_write", "missing_read", "wrong_role"]) async def test_failed_import_or_verification_never_replaces_routes(failure): class BrokenCloud(_FakeCloud): async def upsert_app_user(self, *args, **kwargs): if failure == "empty_write": return {} return await super().upsert_app_user(*args, **kwargs) async def list_app_users(self, project_id): if failure == "missing_read": return [] return [{**user, "role": "member"} for user in self.upserts] with tempfile.TemporaryDirectory() as raw: root = Path(raw) / "checkout" _write_nuxt_store(root) with pytest.raises(ValueError, match="preserved"): await cut_over_generated_repo( db=None, project=_FakeProject(), repo_path=str(root), import_dry_run=False, adapter_dry_run=False, ensure_auth=False, cloud=BrokenCloud(), ) assert (root / "server/api/admin/users.ts").read_text() == LOCAL_USERS_ROUTE def test_dynamic_user_routes_are_retired_for_cloud_catchall(): with tempfile.TemporaryDirectory() as raw: root = Path(raw) route = root / "server/api/admin/users/[id]/password.patch.ts" route.parent.mkdir(parents=True) route.write_text("import { updatePassword } from '~/server/utils/user-db';") result = install_auth_users_adapter(str(root), "nuxt_app", replace_local=True) assert "server/api/admin/users/[id]/password.patch.ts" in result["backed_up"][-1] or any( path.endswith("server/api/admin/users/[id]/password.patch.ts") and "local-bak" in path for path in result["backed_up"] ) assert not route.exists() assert (root / "server/api/admin/users/[...path].ts").is_file() @pytest.mark.asyncio async def test_cutover_retires_dual_store_helpers(): from src.domain.codegen.backend_guard import validate_backend_repo from src.domain.codegen.legacy_store_retire import RETIRED_MARKER with tempfile.TemporaryDirectory() as raw: root = Path(raw) / "checkout" _write_nuxt_store(root) user_db = root / "server/utils/user-db.ts" user_db.parent.mkdir(parents=True, exist_ok=True) user_db.write_text( "export const SEED_USERS = [{ email: 'private@example.com' }]\n" "export function seedUserId() { return 'usr_seed_owner' }\n", encoding="utf-8", ) report = await cut_over_generated_repo( db=None, project=_FakeProject(), repo_path=str(root), import_dry_run=False, adapter_dry_run=False, ensure_auth=False, cloud=_FakeCloud(), retire_shadow_stores=True, ) body = user_db.read_text(encoding="utf-8") validate_backend_repo(str(root)) assert "server/utils/user-db.ts" in report.retired_stores assert RETIRED_MARKER in body assert "usr_seed_" not in body assert not (root / "server/utils/user-db.ts.p2hs-local.bak").exists() @pytest.mark.asyncio async def test_workspace_merge_keeps_export_fields_and_cli_users(monkeypatch): from src.domain.codegen.legacy_cloud_cutover import _identities_for_cutover from src.domain.project_auth.staff_directory import ( STAFF_ADMIN_ROLE, SUPER_ADMIN_ROLE, WorkspaceAppIdentity, ) legacy = { "id": "usr_live_zakir", "email": "muhammad.zakir@pie-solutions.com", "role": "super_admin", "is_active": True, } exported = DiscoveredUser( "muhammad.zakir@pie-solutions.com", "super_admin", "Zakir", "live_export", legacy, ) extra = DiscoveredUser("mohsin@pie-solutions.com", "admin", "Mohsin", "cli") async def fake_identities(db, workspace_id, owner): return [ WorkspaceAppIdentity("owner@example.com", "Owner", "owner", SUPER_ADMIN_ROLE, True), WorkspaceAppIdentity( "muhammad.zakir@pie-solutions.com", "Muhammad Zakir", "admin", STAFF_ADMIN_ROLE, False, ), ] class Workspace: owner_id = uuid4() class DB: async def get(self, model, pk): return Workspace() monkeypatch.setattr( "src.domain.project_auth.staff_directory.identities_for_workspace", fake_identities, ) merged = await _identities_for_cutover(DB(), _FakeProject(), [exported, extra]) by_email = {user.email: user for user in merged} assert by_email["muhammad.zakir@pie-solutions.com"].role == "admin" assert by_email["muhammad.zakir@pie-solutions.com"].legacy_data == legacy assert by_email["mohsin@pie-solutions.com"].role == "admin" assert by_email["owner@example.com"].role == "super_admin" @pytest.mark.asyncio async def test_live_export_merges_git_scan_and_keeps_cli_users(): from src.domain.codegen.legacy_data_export import _digest, save_export project = _FakeProject() legacy = { "id": "usr_live_zakir", "email": "zakir@example.com", "name": "Zakir", "role": "admin", "is_active": True, } snapshot = { "version": 1, "project_id": str(project.id), "origin": "https://prev.example", "exported_at": "2026-09-08T00:00:00+00:00", "coverage": "listed_endpoints_only", "collections": { "users": { "path": "/api/admin/users", "count": 1, "rows": [legacy], "pages": [], } }, } snapshot["sha256"] = _digest(snapshot) with tempfile.TemporaryDirectory() as raw: export_path = save_export(snapshot, raw) root = Path(raw) / "checkout" _write_nuxt_store(root) cloud = _FakeCloud() report = await cut_over_generated_repo( db=None, project=project, repo_path=str(root), import_dry_run=False, adapter_dry_run=False, ensure_auth=False, extra_users=[DiscoveredUser("cli@example.com", "admin", "CLI", "cli")], cloud=cloud, data_export=str(export_path), ) emails = {item.email: item.role for item in report.imported} assert emails == { "zakir@example.com": "admin", "muhammad.zakir@pie-solutions.com": "super_admin", "mohsin@pie-solutions.com": "admin", "cli@example.com": "admin", } assert all(item.status == "upserted" for item in report.imported) preserved = next(row for row in cloud.upserts if row["email"] == "zakir@example.com") assert preserved["app_metadata"]["p2hs_legacy_user"] == legacy @pytest.mark.asyncio async def test_cutover_imports_records_rewrites_ids_and_drops_local_store(): from src.domain.codegen.legacy_data_export import _digest, save_export from src.domain.codegen.legacy_store_retire import RETIRED_MARKER project = _FakeProject() user = { "id": "usr-1", "email": "staff@example.com", "role": "admin", "name": "Staff", "is_active": True, } student = {"id": "stu-1", "first_name": "Ada", "email": "ada@example.com"} application = { "id": "app-1", "student_id": "stu-1", "university": "MIT", } snapshot = { "version": 1, "project_id": str(project.id), "origin": "https://prev.example", "exported_at": "2026-09-08T00:00:00+00:00", "coverage": "listed_endpoints_only", "collections": { "users": {"count": 1, "rows": [user], "kind": "users"}, "students": { "count": 1, "rows": [student], "kind": "records", "entity": "students", }, "applications": { "count": 1, "rows": [application], "kind": "records", "entity": "applications", }, }, } snapshot["sha256"] = _digest(snapshot) with tempfile.TemporaryDirectory() as raw: export_path = save_export(snapshot, raw) root = Path(raw) / "checkout" _write_nuxt_store(root) service = root / "server/services/record.service.ts" service.parent.mkdir(parents=True, exist_ok=True) service.write_text( "import { isCloudConfigured } from '~/server/utils/p2hs'\n" "import { createDbRecord } from '~/server/utils/records-db'\n", encoding="utf-8", ) (root / "server/api/records/students/query.post.ts").parent.mkdir(parents=True) (root / "server/api/records/students/query.post.ts").write_text( "export default defineEventHandler(() => ({}))", encoding="utf-8", ) user_db = root / "server/utils/records-db.ts" user_db.parent.mkdir(parents=True, exist_ok=True) user_db.write_text("export async function createDbRecord() { return { record: {} } }\n") cloud = _FakeCloud() report = await cut_over_generated_repo( db=None, project=project, repo_path=str(root), import_dry_run=False, adapter_dry_run=False, ensure_auth=False, cloud=cloud, data_export=str(export_path), retire_shadow_stores=True, ) service_text = service.read_text(encoding="utf-8") stub = user_db.read_text(encoding="utf-8") local_query = root / "server/api/records/students/query.post.ts" assert report.imported_records == {"students": 1, "applications": 1} assert cloud.records["applications"][0]["data"]["student_id"].startswith("cloud-students-") assert "createRecord" in service_text assert "records-db" not in service_text assert RETIRED_MARKER in stub assert not local_query.exists() def test_legacy_entities_json_maps_universities_schema(): from src.domain.codegen.legacy_cloud_cutover import _legacy_entity_declarations with tempfile.TemporaryDirectory() as raw: root = Path(raw) (root / "p2hs").mkdir() (root / "p2hs" / "entities.json").write_text( '{"entities":[{"slug":"universities","name":"Universities","rls_enabled":true,' '"fields":[{"name":"name","type":"text","required":true},' '{"name":"is_active","type":"boolean","default":true}]}]}', encoding="utf-8", ) declared = _legacy_entity_declarations(str(root)) assert declared[0]["key"] == "universities" assert declared[0]["rls_enabled"] is True by_key = {field["key"]: field for field in declared[0]["fields"]} assert by_key["name"]["required"] is True assert by_key["is_active"]["type"] == "boolean" assert "default" not in by_key["is_active"]