"""Database integration for Core hold mutations and their audit trail (Workstream F). Runs only when P2HS_AUTHZ_TEST_DATABASE_URL points at a migrated, non-production database. Every write is rolled back. """ 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_holds import HoldError, place_local_hold, release_hold from src.domain.project_auth.membership_projection import reconcile_project_memberships from src.domain.project_auth.models import ProjectAuthAuditEvent, ProjectMembership from src.domain.project_auth.staff_directory import WorkspaceAppIdentity from src.domain.projects import models as _project_models # noqa: F401 registers projects table from src.domain.workspaces import models as _workspace_models # noqa: F401 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() async def _project(db): """A project with the canonical owner (protected, v2 preset) and admin (tier admin) roles.""" import json ws, project = uuid4(), uuid4() await db.execute(text("insert into workspaces (id, name, slug) values (:id, 'ws', :slug)"), {"id": ws, "slug": 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', :slug, '{ptype}', :ws)"), {"id": project, "slug": 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 def _ws(email, *, owner=False, admin=False): if owner: return WorkspaceAppIdentity(email, "", "owner", "super_admin", True) return WorkspaceAppIdentity(email, "", "admin", "admin", False) 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 _add_hold(db, membership_id, source): await db.execute(text( "insert into project_membership_suspensions (membership_id, source, reason_code) " "values (:m, cast(:s as p2hs_project_membership_suspension_source), 'seeded')"), {"m": membership_id, "s": source}) await db.execute(text("update project_memberships set status = 'suspended' where id = :m"), {"m": membership_id}) async def _holds(db, membership_id): rows = await db.execute(text( "select source::text from project_membership_suspensions where membership_id = :m " "order by source::text"), {"m": membership_id}) return [r[0] for r in rows.all()] async def _events(db, project): rows = (await db.execute(select(ProjectAuthAuditEvent).where( ProjectAuthAuditEvent.project_id == project))).scalars().all() return rows async def _roster(db): """Project with owner, admin, staff-like admin peer, and a plain target.""" project = await _project(db) await reconcile_project_memberships(db, project, [ _ws("owner@x.test", owner=True), _ws("admin@x.test", admin=True), _ws("peer@x.test", admin=True), ]) owner = await _member(db, project, "owner@x.test") admin = await _member(db, project, "admin@x.test") peer = await _member(db, project, "peer@x.test") return project, owner, admin, peer @pytest.mark.asyncio async def test_owner_places_and_reinstates_a_local_hold_on_an_admin_peer(db): project, owner, admin, peer = await _roster(db) # Only the owner may suspend an admin (D11); the admin peer is held by the owner here. status = await place_local_hold(db, project, actor_membership_id=owner.id, target_membership_id=peer.id, reason_code="review") assert status == "suspended" assert await _holds(db, peer.id) == ["local"] result = await release_hold(db, project, actor_membership_id=owner.id, target_membership_id=peer.id, source="local") assert result.code == "reinstated" assert await _holds(db, peer.id) == [] @pytest.mark.asyncio async def test_admin_cannot_suspend_an_admin_and_owner_cannot_be_held(db): project, owner, admin, peer = await _roster(db) with pytest.raises(HoldError) as exc: await place_local_hold(db, project, actor_membership_id=admin.id, target_membership_id=peer.id, reason_code="x") assert exc.value.code == "permission_denied" with pytest.raises(HoldError) as exc: await place_local_hold(db, project, actor_membership_id=admin.id, target_membership_id=owner.id, reason_code="x") assert exc.value.code == "role_protected" with pytest.raises(HoldError) as exc: await place_local_hold(db, project, actor_membership_id=owner.id, target_membership_id=owner.id, reason_code="x") assert exc.value.code == "self_suspend_denied" @pytest.mark.asyncio async def test_hold_mutations_write_audit_and_bump_the_target_once(db): project, owner, admin, peer = await _roster(db) before = peer.authorization_version await place_local_hold(db, project, actor_membership_id=owner.id, target_membership_id=peer.id, reason_code="review", correlation_id="corr-1") await release_hold(db, project, actor_membership_id=owner.id, target_membership_id=peer.id, source="local", correlation_id="corr-2") await db.refresh(peer) assert peer.authorization_version == before + 2 events = await _events(db, project) assert sorted(e.event for e in events) == ["project.member.hold_placed", "project.member.hold_released"] assert {e.correlation_id for e in events} == {"corr-1", "corr-2"} @pytest.mark.asyncio async def test_legacy_hold_is_owner_only_and_releases_deliberately(db): project, owner, admin, peer = await _roster(db) await _add_hold(db, peer.id, "legacy") before = peer.authorization_version with pytest.raises(HoldError) as exc: await release_hold(db, project, actor_membership_id=admin.id, target_membership_id=peer.id, source="legacy") assert exc.value.code == "permission_denied" assert await _holds(db, peer.id) == ["legacy"] assert await _events(db, project) == [] result = await release_hold(db, project, actor_membership_id=owner.id, target_membership_id=peer.id, source="legacy", correlation_id="corr-legacy") assert result.code == "reinstated" assert await _holds(db, peer.id) == [] await db.refresh(peer) assert peer.status == "active" assert peer.authorization_version == before + 1 [event] = await _events(db, project) assert event.event == "project.member.legacy_suspension_released" assert event.actor_membership_id == owner.id and event.target_membership_id == peer.id assert event.correlation_id == "corr-legacy" assert event.payload == { "source": "legacy", "previous_sources": ["legacy"], "remaining_sources": [], "authorization_version_before": before, "authorization_version_after": before + 1, } @pytest.mark.asyncio async def test_legacy_release_with_local_hold_remaining_stays_suspended(db): project, owner, admin, peer = await _roster(db) await _add_hold(db, peer.id, "legacy") await _add_hold(db, peer.id, "local") result = await release_hold(db, project, actor_membership_id=owner.id, target_membership_id=peer.id, source="legacy") assert result.code == "membership_still_suspended" assert result.remaining_sources == ("local",) await db.refresh(peer) assert peer.status == "suspended" @pytest.mark.asyncio async def test_local_reinstatement_never_clears_a_legacy_hold(db): project, owner, admin, peer = await _roster(db) await _add_hold(db, peer.id, "legacy") result = await release_hold(db, project, actor_membership_id=owner.id, target_membership_id=peer.id, source="local") assert result.released is False assert result.remaining_sources == ("legacy",) assert await _holds(db, peer.id) == ["legacy"] @pytest.mark.asyncio async def test_workspace_holds_are_never_released_by_a_local_mutation(db): project, owner, admin, peer = await _roster(db) await _add_hold(db, peer.id, "workspace") for actor in (owner, admin): with pytest.raises(HoldError) as exc: await release_hold(db, project, actor_membership_id=actor.id, target_membership_id=peer.id, source="workspace") assert exc.value.code == "hold_source_not_releasable" assert await _holds(db, peer.id) == ["workspace"] @pytest.mark.asyncio async def test_repeated_legacy_release_is_a_noop(db): project, owner, admin, peer = await _roster(db) await _add_hold(db, peer.id, "legacy") await release_hold(db, project, actor_membership_id=owner.id, target_membership_id=peer.id, source="legacy") await db.refresh(peer) version_after_first = peer.authorization_version again = await release_hold(db, project, actor_membership_id=owner.id, target_membership_id=peer.id, source="legacy") assert again.released is False await db.refresh(peer) assert peer.authorization_version == version_after_first assert len(await _events(db, project)) == 1 @pytest.mark.asyncio async def test_actor_from_another_project_is_refused(db): project, owner, admin, peer = await _roster(db) other, other_owner, _, _ = await _roster(db) await _add_hold(db, peer.id, "legacy") with pytest.raises(HoldError) as exc: await release_hold(db, project, actor_membership_id=other_owner.id, target_membership_id=peer.id, source="legacy") assert exc.value.code == "actor_not_found" assert await _holds(db, peer.id) == ["legacy"]