"""Versioned staff authorization: effective live truth controls install readiness.""" from copy import deepcopy import json from types import SimpleNamespace from unittest.mock import AsyncMock, MagicMock, patch from uuid import uuid4 import pytest from src.domain.feature_packs.crm_authorization import ( CRM_AUTHORIZATION_CONFLICT, CrmAuthorizationError, classify_policy, ensure_installed_policy, staff_policy, ) from src.domain.feature_packs.service import PackInstallService from tests.test_phase11_pack_migration_materialization import _captured_composition, _load_migration def lead_composition(): migration = _load_migration("alembic/versions/143_lead_staff_authorization.py") raw = _captured_composition("alembic/versions/132_lead_lifecycle_pack.py", "lead-lifecycle") return migration.upgrade_composition(raw) class LiveCloud: def __init__(self, project, policy="absent"): schema = deepcopy(lead_composition()["entities"][0]["schema"]) schema.pop("access_policy") schema["fields"].append({"key": "custom", "type": "text", "indexed": True}) schema["custom_metadata"] = {"retain": [1, 2]} if policy != "absent": schema["access_policy"] = policy self.entity = {"project_id": str(project), "entity_key": "lead", "status": "active", "version": 4, "schema_json": json.dumps(schema), "rls_enabled": 0, "owner_managed": 0, "owner_singleton": 0, "settings_json": '{"retain":true}'} self.patch_calls = [] self.reads = 0 self.failure = None async def create_project_entity(self, *args): return {"already_exists": True} async def get_project_entity(self, project, entity): self.reads += 1 if self.failure == "verification" and self.reads > 1: raise RuntimeError("read unavailable") return {"entity": deepcopy(self.entity), "schema_update_precondition": "entity_version"} async def patch_project_entity(self, project, entity, body, *, expected_version): self.patch_calls.append(deepcopy(body)) assert expected_version == self.entity["version"] if self.failure == "patch": raise RuntimeError("patch unavailable") if self.failure != "dropped": self.entity["schema_json"] = json.dumps(body["schema"]) self.entity["version"] += 1 async def get_project_schema(self, project): return {"entities": [{"key": "lead", "fields": json.loads(self.entity["schema_json"])["fields"]}]} async def install_fixture(cloud, project, *, version="1.1.0", composition=None, bookkeeping=None): pack = SimpleNamespace(id=uuid4(), slug="lead-lifecycle", status="active", version=version, composition=composition or lead_composition(), input_schema=None) svc = PackInstallService(AsyncMock(), MagicMock(), cloud) svc._get_pack_by_slug = AsyncMock(return_value=pack) svc._installed_config = AsyncMock(return_value={}) svc._upsert_install_state = bookkeeping or AsyncMock() with patch("src.domain.projects.cloud_schema_catalog.ProjectCloudSchemaSnapshotRepository.upsert", new_callable=AsyncMock): result = await svc.install_pack(project_id=project, pack_slug=pack.slug, config={}) return result, svc._upsert_install_state def test_catalog_upgrade_is_idempotent_and_preserves_historical_input(): raw = _captured_composition("alembic/versions/132_lead_lifecycle_pack.py", "lead-lifecycle") original = deepcopy(raw) migration = _load_migration("alembic/versions/143_lead_staff_authorization.py") upgraded = migration.upgrade_composition(raw) assert raw == original assert migration.upgrade_composition(upgraded) == upgraded assert upgraded["entities"][0]["schema"]["access_policy"] == staff_policy() assert upgraded["compatibility"]["implementation_version"] == "1.1.0" assert not upgraded["entities"][0].get("rls_enabled") assert not upgraded["entities"][0].get("owner_managed") @pytest.mark.parametrize("value", [None, [], "unknown", {}, {"version": 2, "roles": {}}, {"version": True, "roles": {}}, {"version": 1, "roles": []}]) def test_malformed_explicit_policy_is_not_absence(value): assert classify_policy({"access_policy": value}) == "malformed_policy" def test_semantically_equal_explicit_denials_are_noop(): policy = staff_policy() policy["roles"]["member"] = {a: {"effect": "deny"} for a in ("create", "read", "update", "delete")} assert classify_policy({"access_policy": policy}) == "canonical_policy" policy["roles"]["admin"]["read"].pop("scope") assert classify_policy({"access_policy": policy}) == "conflicting_policy" @pytest.mark.asyncio async def test_live_merge_preserves_everything_except_policy_and_rerun_is_noop(): project = uuid4() cloud = LiveCloud(project) before = json.loads(cloud.entity["schema_json"]) result, state = await install_fixture(cloud, project) assert result["status"] == "installed" assert state.await_args.kwargs["status"] == "installed" after = json.loads(cloud.entity["schema_json"]) assert after.pop("access_policy") == staff_policy() assert after == before assert cloud.entity["rls_enabled"] == cloud.entity["owner_managed"] == 0 result, _ = await install_fixture(cloud, project) assert result["ok"] assert len(cloud.patch_calls) == 1 @pytest.mark.parametrize("policy", [None, {}, {"version": 1, "roles": {}}, {"version": 1, "roles": {"member": {"read": {"effect": "allow", "scope": "all"}}}}]) @pytest.mark.asyncio async def test_custom_or_malformed_policy_blocks_repeatedly_without_mutation(policy): project = uuid4() cloud = LiveCloud(project, policy) before = deepcopy(cloud.entity) for _ in range(2): result, state = await install_fixture(cloud, project) assert not result["ok"] assert result["blocker"]["validation_code"] == CRM_AUTHORIZATION_CONFLICT assert state.await_args.kwargs["status"] == "blocked" assert cloud.entity == before assert cloud.patch_calls == [] @pytest.mark.parametrize("failure", ["patch", "verification", "dropped"]) @pytest.mark.asyncio async def test_delivery_or_verification_failure_never_reports_ready_and_retry_converges(failure): project = uuid4() cloud = LiveCloud(project) cloud.failure = failure result, state = await install_fixture(cloud, project) assert result["status"] == "blocked" assert state.await_args.kwargs["status"] == "blocked" cloud.failure = None result, _ = await install_fixture(cloud, project) assert result["ok"] assert len(cloud.patch_calls) == (1 if failure == "verification" else 2) @pytest.mark.asyncio async def test_interruption_after_patch_retries_without_rollback(): project = uuid4() cloud = LiveCloud(project) with pytest.raises(RuntimeError, match="interrupted"): await install_fixture(cloud, project, bookkeeping=AsyncMock(side_effect=RuntimeError("interrupted"))) assert classify_policy(json.loads(cloud.entity["schema_json"])) == "canonical_policy" result, _ = await install_fixture(cloud, project) assert result["ok"] and len(cloud.patch_calls) == 1 @pytest.mark.asyncio async def test_old_gateway_without_conditional_support_fails_before_write(): project = uuid4() cloud = LiveCloud(project) cloud.get_project_entity = AsyncMock(return_value={"entity": cloud.entity}) with pytest.raises(CrmAuthorizationError, match="conditional"): await ensure_installed_policy(cloud, str(project), "lead", allow_legacy_upgrade=True) assert not cloud.patch_calls @pytest.mark.asyncio async def test_stale_catalog_cannot_report_new_pipeline_ready(): project = uuid4() cloud = LiveCloud(project) result, _ = await install_fixture(cloud, project, version="1.0.0") assert result["status"] == "blocked" and cloud.reads == 0 and not cloud.patch_calls @pytest.mark.asyncio async def test_failed_upgrade_does_not_advance_verified_pack_version(): old = SimpleNamespace(pack_version="1.0.0", verified_at=None, installed_at=None, config={}) db = AsyncMock() db.execute.return_value = MagicMock(scalar_one_or_none=MagicMock(return_value=old)) svc = PackInstallService(db, MagicMock(), MagicMock()) pack = SimpleNamespace(id=uuid4(), slug="lead-lifecycle", version="1.1.0") await svc._upsert_install_state(project_id=uuid4(), pack=pack, config={}, status="blocked", blocker={}) assert old.pack_version == "1.0.0" assert old.status == "blocked"