296 lines
11 KiB
Python
296 lines
11 KiB
Python
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()
|