"""Database check for Core → Cloud projection sync (Workstream E4). Needs the scratch URL.""" import json import os from uuid import uuid4 import pytest import pytest_asyncio from sqlalchemy import select, text from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine from src.domain.project_auth import permissions as perms from src.domain.project_auth.membership_projection import reconcile_project_memberships from src.domain.project_auth.membership_projection_sync import sync_project_projections from src.domain.project_auth.models import ProjectMembership, ProjectMembershipProjectionSync from src.domain.project_auth.staff_directory import WorkspaceAppIdentity from src.domain.projects import models as _project_models # noqa: F401 from src.domain.workspaces import models as _workspace_models # noqa: F401 from src.integrations.auth_projection_writer import WriteResult URL = os.environ.get("P2HS_AUTHZ_TEST_DATABASE_URL", "") pytestmark = pytest.mark.skipif(not URL, reason="P2HS_AUTHZ_TEST_DATABASE_URL not set") @pytest_asyncio.fixture async def db(): engine = create_async_engine(URL) async with engine.connect() as conn: trans = await conn.begin() session = AsyncSession(bind=conn, expire_on_commit=False) try: yield session finally: await session.close() await trans.rollback() await engine.dispose() class FakeWriter: def __init__(self, fail: bool = False) -> None: self.fail = fail self.calls: list[tuple[str, str, dict]] = [] async def put_projection(self, project_id, cloud_app_user_id, payload): self.calls.append((project_id, cloud_app_user_id, payload)) if self.fail: return WriteResult(ok=False, status=503, code="http_503") return WriteResult(ok=True, status=200, code=None, changed=True) def _ws(email, *, owner=False): if owner: return WorkspaceAppIdentity(email, "", "owner", "super_admin", True) return WorkspaceAppIdentity(email, "", "admin", "admin", False) async def _project(db): ws, project = uuid4(), uuid4() await db.execute(text("insert into workspaces (id, name, slug) values (:id, 'ws', :s)"), {"id": ws, "s": f"ws-{ws}"}) ptype = (await db.execute(text( "select enumlabel from pg_enum e join pg_type t on t.oid = e.enumtypid " "where t.typname = (select udt_name from information_schema.columns " "where table_name = 'projects' and column_name = 'type') order by enumsortorder limit 1" ))).scalar_one() await db.execute(text(f"insert into projects (id, name, slug, type, workspace_id) " f"values (:id, 'p', :s, '{ptype}', :ws)"), {"id": project, "s": f"p-{project}", "ws": ws}) await db.execute(text( "insert into project_auth_roles (id, project_id, name, created_by, permissions_json, " "assignment_tier, protected) values " "(:id, :p, 'super_admin', 'codegen', cast(:perms as jsonb), NULL, true)"), {"id": uuid4(), "p": project, "perms": json.dumps(list(perms.OWNER_PRESET))}) await db.execute(text( "insert into project_auth_roles (id, project_id, name, created_by, permissions_json, " "assignment_tier, protected) values (:id, :p, 'admin', 'codegen', cast(:perms as jsonb), " "'admin', false)"), {"id": uuid4(), "p": project, "perms": json.dumps(list(perms.ADMIN_PRESET))}) return project async def _member(db, project, email): return (await db.execute(select(ProjectMembership).where( ProjectMembership.project_id == project, ProjectMembership.email_normalized == email))).scalar_one() async def _link(db, project, email, cloud_id): await db.execute(text( "update project_memberships set cloud_app_user_id = :c where project_id = :p " "and email_normalized = :e"), {"c": cloud_id, "p": project, "e": email}) async def _state(db, membership_id): return (await db.execute(select(ProjectMembershipProjectionSync).where( ProjectMembershipProjectionSync.membership_id == membership_id))).scalar_one_or_none() @pytest.mark.asyncio async def test_first_sync_acknowledges_then_unchanged_state_sends_nothing(db): project = await _project(db) await reconcile_project_memberships(db, project, [_ws("owner@x.test", owner=True), _ws("admin@x.test")]) await _link(db, project, "owner@x.test", "usr_owner") await _link(db, project, "admin@x.test", "usr_admin") writer = FakeWriter() first = await sync_project_projections(db, project, writer) assert {o.status for o in first} == {"synced"} assert len(writer.calls) == 2 writer.calls.clear() second = await sync_project_projections(db, project, writer) assert {o.status for o in second} == {"unchanged"} assert writer.calls == [] @pytest.mark.asyncio async def test_cloud_failure_is_pending_and_never_changes_core_authority(db): project = await _project(db) await reconcile_project_memberships(db, project, [_ws("owner@x.test", owner=True), _ws("admin@x.test")]) await _link(db, project, "owner@x.test", "usr_owner") await _link(db, project, "admin@x.test", "usr_admin") admin = await _member(db, project, "admin@x.test") before = (admin.authorization_version, admin.status) outcomes = await sync_project_projections(db, project, FakeWriter(fail=True)) assert any(o.status == "pending" and o.code == "http_503" for o in outcomes) state = await _state(db, admin.id) assert state.status == "pending" and state.acknowledged_digest is None assert state.attempts == 1 and state.last_error == "http_503" await db.refresh(admin) assert (admin.authorization_version, admin.status) == before @pytest.mark.asyncio async def test_recovery_after_failure_acknowledges_the_outstanding_projection(db): project = await _project(db) await reconcile_project_memberships(db, project, [_ws("owner@x.test", owner=True), _ws("admin@x.test")]) await _link(db, project, "owner@x.test", "usr_owner") await _link(db, project, "admin@x.test", "usr_admin") await sync_project_projections(db, project, FakeWriter(fail=True)) writer = FakeWriter() retry = await sync_project_projections(db, project, writer) assert {o.status for o in retry} == {"synced"} admin = await _member(db, project, "admin@x.test") state = await _state(db, admin.id) assert state.status == "synced" and state.acknowledged_digest is not None assert state.last_error is None @pytest.mark.asyncio async def test_a_version_bump_resyncs_the_changed_membership_only(db): project = await _project(db) await reconcile_project_memberships(db, project, [_ws("owner@x.test", owner=True), _ws("admin@x.test")]) await _link(db, project, "owner@x.test", "usr_owner") await _link(db, project, "admin@x.test", "usr_admin") writer = FakeWriter() await sync_project_projections(db, project, writer) # The admin leaves the workspace: a real authority change that bumps only that membership. await reconcile_project_memberships(db, project, [_ws("owner@x.test", owner=True)]) writer.calls.clear() outcomes = await sync_project_projections(db, project, writer) assert [c[1] for c in writer.calls] == ["usr_admin"] payload = writer.calls[0][2] assert payload["status"] == "suspended" and payload["project_permissions"] == [] assert {o.email: o.status for o in outcomes} == {"owner@x.test": "unchanged", "admin@x.test": "synced"} @pytest.mark.asyncio async def test_an_unlinked_identity_waits_and_is_never_sent(db): project = await _project(db) await reconcile_project_memberships(db, project, [_ws("owner@x.test", owner=True), _ws("admin@x.test")]) await _link(db, project, "owner@x.test", "usr_owner") writer = FakeWriter() outcomes = await sync_project_projections(db, project, writer) assert [c[1] for c in writer.calls] == ["usr_owner"] waiting = next(o for o in outcomes if o.email == "admin@x.test") assert waiting.status == "pending" and waiting.code == "no_cloud_identity"