security: make Campaign mutation audits atomic

tests/test_campaign_audit_atomicity.py

@@ -0,0 +1,119 @@
+from __future__ import annotations
+
+from types import SimpleNamespace
+from unittest.mock import MagicMock, patch
+
+import pytest
+from fastapi import HTTPException
+
+from govoplan_campaign.backend import router
+from govoplan_campaign.backend.schemas import CampaignUpdateRequest, CampaignVersionUpdateRequest
+
+
+def _principal() -> SimpleNamespace:
+    return SimpleNamespace(
+        tenant_id="tenant-1",
+        user=SimpleNamespace(id="user-1"),
+        api_key=None,
+    )
+
+
+def test_version_update_rolls_back_when_its_audit_record_cannot_be_written() -> None:
+    session = MagicMock()
+    principal = _principal()
+    version = SimpleNamespace(
+        id="version-1",
+        raw_json={},
+        current_flow="manual",
+        current_step="recipients",
+    )
+
+    def mutate(*_args, **kwargs):
+        assert kwargs["commit"] is False
+        session.flush()
+        return version
+
+    with (
+        patch.object(router, "_get_campaign_for_principal"),
+        patch.object(router, "_get_version_for_tenant", return_value=version),
+        patch.object(router, "update_campaign_version", side_effect=mutate),
+        patch.object(router, "audit_from_principal", side_effect=RuntimeError("audit unavailable")),
+    ):
+        with pytest.raises(HTTPException, match="audit unavailable") as captured:
+            router._update_campaign_version_detail_response(  # noqa: SLF001 - transaction regression test
+                session,
+                principal,  # type: ignore[arg-type]
+                "campaign-1",
+                "version-1",
+                CampaignVersionUpdateRequest(current_step="recipients"),
+                autosave=True,
+                audit_action="campaign.version_autosaved",
+            )
+
+    assert captured.value.status_code == 422
+    session.commit.assert_not_called()
+    session.rollback.assert_called_once_with()
+
+
+def test_version_fork_rolls_back_when_its_audit_record_cannot_be_written() -> None:
+    session = MagicMock()
+    principal = _principal()
+    campaign = SimpleNamespace(id="campaign-1")
+    source = SimpleNamespace(id="version-1", campaign_id="campaign-1", raw_json={})
+    forked = SimpleNamespace(id="version-2", campaign_id="campaign-1", version_number=2)
+
+    def mutate(*_args, **kwargs):
+        assert kwargs["commit"] is False
+        session.flush()
+        return forked
+
+    with (
+        patch.object(router, "_get_campaign_for_principal", return_value=campaign),
+        patch.object(router, "_require_permission"),
+        patch.object(router, "_get_version_for_tenant", return_value=source),
+        patch.object(router, "_get_campaign_for_tenant", return_value=campaign),
+        patch.object(router, "fork_campaign_version_for_edit", side_effect=mutate),
+        patch.object(router, "audit_from_principal", side_effect=RuntimeError("audit unavailable")),
+    ):
+        with pytest.raises(RuntimeError, match="audit unavailable"):
+            router.fork_version_for_edit(
+                "campaign-1",
+                "version-1",
+                CampaignVersionUpdateRequest(),
+                session=session,
+                principal=principal,  # type: ignore[arg-type]
+            )
+
+    session.commit.assert_not_called()
+    session.rollback.assert_called_once_with()
+
+
+def test_metadata_update_rolls_back_when_its_audit_record_cannot_be_written() -> None:
+    session = MagicMock()
+    principal = _principal()
+    campaign = SimpleNamespace(
+        id="campaign-1",
+        tenant_id="tenant-1",
+        external_id="C-1",
+        name="Old name",
+        description=None,
+        status="draft",
+        current_version_id=None,
+    )
+
+    with (
+        patch.object(router, "_get_campaign_for_principal", return_value=campaign),
+        patch.object(router, "_sync_campaign_metadata_to_current_version"),
+        patch.object(router, "audit_from_principal", side_effect=RuntimeError("audit unavailable")),
+    ):
+        with pytest.raises(RuntimeError, match="audit unavailable"):
+            router.update_campaign_metadata_endpoint(
+                "campaign-1",
+                CampaignUpdateRequest(name="New name"),
+                session=session,
+                principal=principal,  # type: ignore[arg-type]
+            )
+
+    session.flush.assert_called_once_with()
+    session.commit.assert_not_called()
+    session.rollback.assert_called_once_with()