Complete governed campaign lifecycle actions
This commit is contained in:
@@ -0,0 +1,295 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import unittest
|
||||
from datetime import UTC, datetime
|
||||
from types import SimpleNamespace
|
||||
from unittest.mock import patch
|
||||
|
||||
from fastapi import HTTPException
|
||||
from sqlalchemy import Column, String, Table, create_engine
|
||||
from sqlalchemy.orm import Session, sessionmaker
|
||||
|
||||
from govoplan_campaign.backend.campaign.lifecycle import campaign_lifecycle_policy
|
||||
from govoplan_campaign.backend.db.models import (
|
||||
Campaign,
|
||||
CampaignJob,
|
||||
CampaignShare,
|
||||
CampaignVersion,
|
||||
)
|
||||
from govoplan_campaign.backend.routes.campaigns import (
|
||||
archive_campaign_version,
|
||||
copy_campaign,
|
||||
delete_draft_campaign,
|
||||
)
|
||||
from govoplan_campaign.backend.schemas import (
|
||||
CampaignCopyRequest,
|
||||
CampaignLifecycleMutationRequest,
|
||||
)
|
||||
from govoplan_core.db.base import Base
|
||||
|
||||
|
||||
class _Principal:
|
||||
tenant_id = "tenant-1"
|
||||
user = SimpleNamespace(id="user-1")
|
||||
|
||||
def __init__(self, *scopes: str) -> None:
|
||||
self.scopes = frozenset(scopes)
|
||||
|
||||
def has(self, scope: str) -> bool:
|
||||
return scope in self.scopes or "tenant:*" in self.scopes
|
||||
|
||||
|
||||
class CampaignLifecycleTests(unittest.TestCase):
|
||||
def setUp(self) -> None:
|
||||
self.engine = create_engine("sqlite+pysqlite:///:memory:")
|
||||
access_users = Base.metadata.tables.get("access_users")
|
||||
if access_users is None:
|
||||
access_users = Table(
|
||||
"access_users",
|
||||
Base.metadata,
|
||||
Column("id", String(36), primary_key=True),
|
||||
)
|
||||
access_groups = Base.metadata.tables.get("access_groups")
|
||||
if access_groups is None:
|
||||
access_groups = Table(
|
||||
"access_groups",
|
||||
Base.metadata,
|
||||
Column("id", String(36), primary_key=True),
|
||||
)
|
||||
Base.metadata.create_all(
|
||||
self.engine,
|
||||
tables=[
|
||||
access_users,
|
||||
access_groups,
|
||||
Campaign.__table__,
|
||||
CampaignVersion.__table__,
|
||||
CampaignShare.__table__,
|
||||
CampaignJob.__table__,
|
||||
],
|
||||
)
|
||||
self.SessionLocal = sessionmaker(
|
||||
bind=self.engine,
|
||||
class_=Session,
|
||||
expire_on_commit=False,
|
||||
)
|
||||
with self.SessionLocal() as session:
|
||||
session.execute(access_users.insert().values(id="user-1"))
|
||||
campaign = Campaign(
|
||||
id="campaign-1",
|
||||
tenant_id="tenant-1",
|
||||
created_by_user_id="user-1",
|
||||
owner_user_id="user-1",
|
||||
external_id="campaign-1",
|
||||
name="Campaign",
|
||||
status="draft",
|
||||
current_version_id="version-2",
|
||||
)
|
||||
historical = CampaignVersion(
|
||||
id="version-1",
|
||||
campaign_id=campaign.id,
|
||||
version_number=1,
|
||||
workflow_state="completed",
|
||||
raw_json={"version": "1.0", "campaign": {"id": "campaign-1", "name": "Campaign"}},
|
||||
)
|
||||
current = CampaignVersion(
|
||||
id="version-2",
|
||||
campaign_id=campaign.id,
|
||||
version_number=2,
|
||||
workflow_state="editing",
|
||||
raw_json={"version": "1.0", "campaign": {"id": "campaign-1", "name": "Campaign"}},
|
||||
)
|
||||
session.add_all((campaign, historical, current))
|
||||
session.commit()
|
||||
self.principal = _Principal(
|
||||
"campaigns:campaign:read",
|
||||
"campaigns:campaign:copy",
|
||||
"campaigns:campaign:archive",
|
||||
"campaigns:campaign:delete",
|
||||
"campaigns:recipient:read",
|
||||
)
|
||||
self.addCleanup(self.engine.dispose)
|
||||
|
||||
def _policy(self, session: Session, version_id: str | None = None):
|
||||
campaign = session.get(Campaign, "campaign-1")
|
||||
assert campaign is not None
|
||||
return campaign_lifecycle_policy(
|
||||
session,
|
||||
campaign=campaign,
|
||||
principal=self.principal,
|
||||
version_id=version_id,
|
||||
)
|
||||
|
||||
def test_policy_explains_retained_evidence_and_changes_token(self) -> None:
|
||||
with self.SessionLocal() as session:
|
||||
initial = self._policy(session, "version-1")
|
||||
self.assertTrue(initial["actions"]["delete_campaign"]["allowed"])
|
||||
self.assertTrue(initial["actions"]["archive_version"]["allowed"])
|
||||
|
||||
current = session.get(CampaignVersion, "version-2")
|
||||
assert current is not None
|
||||
current.published_at = datetime.now(UTC)
|
||||
session.commit()
|
||||
changed = self._policy(session, "version-1")
|
||||
|
||||
self.assertNotEqual(initial["state_token"], changed["state_token"])
|
||||
self.assertFalse(changed["actions"]["delete_campaign"]["allowed"])
|
||||
self.assertIn("Audit-relevant", changed["actions"]["delete_campaign"]["reason"])
|
||||
|
||||
def test_active_delivery_blocks_campaign_archival(self) -> None:
|
||||
with self.SessionLocal() as session:
|
||||
session.add(
|
||||
CampaignJob(
|
||||
id="job-1",
|
||||
tenant_id="tenant-1",
|
||||
campaign_id="campaign-1",
|
||||
campaign_version_id="version-2",
|
||||
entry_index=0,
|
||||
queue_status="sending",
|
||||
send_status="sending",
|
||||
)
|
||||
)
|
||||
session.commit()
|
||||
policy = self._policy(session)
|
||||
self.assertFalse(policy["actions"]["archive_campaign"]["allowed"])
|
||||
self.assertIn("Active or uncertain", policy["actions"]["archive_campaign"]["reason"])
|
||||
|
||||
def test_stale_delete_token_is_rejected(self) -> None:
|
||||
with self.SessionLocal() as session:
|
||||
policy = self._policy(session)
|
||||
campaign = session.get(Campaign, "campaign-1")
|
||||
assert campaign is not None
|
||||
campaign.name = "Changed elsewhere"
|
||||
session.commit()
|
||||
|
||||
with self.assertRaises(HTTPException) as raised:
|
||||
delete_draft_campaign(
|
||||
"campaign-1",
|
||||
CampaignLifecycleMutationRequest(
|
||||
expected_state_token=policy["state_token"],
|
||||
),
|
||||
session=session,
|
||||
principal=self.principal,
|
||||
)
|
||||
self.assertEqual(raised.exception.status_code, 409)
|
||||
self.assertIn("state changed", str(raised.exception.detail))
|
||||
|
||||
def test_historical_archival_preserves_version_state_and_content(self) -> None:
|
||||
with self.SessionLocal() as session:
|
||||
policy = self._policy(session, "version-1")
|
||||
|
||||
def commit_audit(active_session: Session, *_args, **_kwargs) -> None:
|
||||
active_session.commit()
|
||||
|
||||
with patch(
|
||||
"govoplan_campaign.backend.routes.campaigns.audit_from_principal",
|
||||
side_effect=commit_audit,
|
||||
):
|
||||
response = archive_campaign_version(
|
||||
"campaign-1",
|
||||
"version-1",
|
||||
CampaignLifecycleMutationRequest(
|
||||
expected_state_token=policy["state_token"],
|
||||
),
|
||||
session=session,
|
||||
principal=self.principal,
|
||||
)
|
||||
|
||||
self.assertIsNotNone(response.archived_at)
|
||||
self.assertEqual(response.archived_by_user_id, "user-1")
|
||||
version = session.get(CampaignVersion, "version-1")
|
||||
assert version is not None
|
||||
self.assertEqual(version.workflow_state, "completed")
|
||||
self.assertEqual(version.raw_json["campaign"]["name"], "Campaign")
|
||||
|
||||
def test_whole_campaign_copy_starts_without_operational_evidence(self) -> None:
|
||||
with self.SessionLocal() as session:
|
||||
session.add_all(
|
||||
(
|
||||
CampaignShare(
|
||||
id="share-1",
|
||||
tenant_id="tenant-1",
|
||||
campaign_id="campaign-1",
|
||||
target_type="user",
|
||||
target_id="user-1",
|
||||
permission="read",
|
||||
),
|
||||
CampaignJob(
|
||||
id="job-1",
|
||||
tenant_id="tenant-1",
|
||||
campaign_id="campaign-1",
|
||||
campaign_version_id="version-2",
|
||||
entry_index=0,
|
||||
queue_status="cancelled",
|
||||
send_status="cancelled",
|
||||
),
|
||||
)
|
||||
)
|
||||
session.commit()
|
||||
policy = self._policy(session, "version-2")
|
||||
captured: dict[str, object] = {}
|
||||
|
||||
def create_copy(active_session: Session, **kwargs):
|
||||
raw_json = kwargs["raw_json"]
|
||||
captured["raw_json"] = raw_json
|
||||
destination = Campaign(
|
||||
id="campaign-copy",
|
||||
tenant_id="tenant-1",
|
||||
created_by_user_id="user-1",
|
||||
owner_user_id="user-1",
|
||||
external_id=raw_json["campaign"]["id"],
|
||||
name=raw_json["campaign"]["name"],
|
||||
status="draft",
|
||||
current_version_id="version-copy",
|
||||
)
|
||||
version = CampaignVersion(
|
||||
id="version-copy",
|
||||
campaign_id=destination.id,
|
||||
version_number=1,
|
||||
raw_json=raw_json,
|
||||
)
|
||||
active_session.add_all((destination, version))
|
||||
active_session.flush()
|
||||
return destination, version
|
||||
|
||||
def commit_audit(active_session: Session, *_args, **_kwargs) -> None:
|
||||
active_session.commit()
|
||||
|
||||
with (
|
||||
patch(
|
||||
"govoplan_campaign.backend.routes.campaigns.create_campaign_version_from_json",
|
||||
side_effect=create_copy,
|
||||
),
|
||||
patch(
|
||||
"govoplan_campaign.backend.routes.campaigns.audit_from_principal",
|
||||
side_effect=commit_audit,
|
||||
),
|
||||
):
|
||||
response = copy_campaign(
|
||||
"campaign-1",
|
||||
CampaignCopyRequest(
|
||||
source_version_id="version-2",
|
||||
expected_state_token=policy["state_token"],
|
||||
),
|
||||
session=session,
|
||||
principal=self.principal,
|
||||
)
|
||||
|
||||
self.assertEqual(response.campaign.external_id, "campaign-1-copy")
|
||||
self.assertEqual(response.campaign.owner_user_id, "user-1")
|
||||
self.assertEqual(captured["raw_json"]["campaign"]["mode"], "draft")
|
||||
self.assertEqual(
|
||||
session.query(CampaignJob)
|
||||
.filter(CampaignJob.campaign_id == "campaign-copy")
|
||||
.count(),
|
||||
0,
|
||||
)
|
||||
self.assertEqual(
|
||||
session.query(CampaignShare)
|
||||
.filter(CampaignShare.campaign_id == "campaign-copy")
|
||||
.count(),
|
||||
0,
|
||||
)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
Reference in New Issue
Block a user