122 lines
4.5 KiB
Python
122 lines
4.5 KiB
Python
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 route_support
|
|
from govoplan_campaign.backend.routes import campaigns as campaign_routes
|
|
from govoplan_campaign.backend.routes import versions as version_routes
|
|
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(route_support, "_get_campaign_for_principal"),
|
|
patch.object(route_support, "_get_version_for_tenant", return_value=version),
|
|
patch.object(route_support, "update_campaign_version", side_effect=mutate),
|
|
patch.object(route_support, "audit_from_principal", side_effect=RuntimeError("audit unavailable")),
|
|
):
|
|
with pytest.raises(HTTPException, match="audit unavailable") as captured:
|
|
route_support._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(version_routes, "_get_campaign_for_principal", return_value=campaign),
|
|
patch.object(version_routes, "_require_permission"),
|
|
patch.object(version_routes, "_get_version_for_tenant", return_value=source),
|
|
patch.object(version_routes, "_get_campaign_for_tenant", return_value=campaign),
|
|
patch.object(version_routes, "fork_campaign_version_for_edit", side_effect=mutate),
|
|
patch.object(version_routes, "audit_from_principal", side_effect=RuntimeError("audit unavailable")),
|
|
):
|
|
with pytest.raises(RuntimeError, match="audit unavailable"):
|
|
version_routes.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(campaign_routes, "_get_campaign_for_principal", return_value=campaign),
|
|
patch.object(campaign_routes, "_sync_campaign_metadata_to_current_version"),
|
|
patch.object(campaign_routes, "audit_from_principal", side_effect=RuntimeError("audit unavailable")),
|
|
):
|
|
with pytest.raises(RuntimeError, match="audit unavailable"):
|
|
campaign_routes.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()
|