GovOPlaN/govoplan-campaign
2
0
Fork 0
Code Issues 2 Pull Requests Actions Packages Projects Releases Wiki Activity

feat: add campaign copying scheduling and residual handling

Browse Source
This commit is contained in:
zemion
2026-08-07 14:54:04 +02:00
parent 696f8f6385
commit c2efd6b7bd
35 changed files with 3359 additions and 39 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/test_campaign_lifecycle.py
+128 -1
View File
@@ -13,10 +13,12 @@ from govoplan_campaign.backend.campaign.lifecycle import campaign_lifecycle_poli
from govoplan_campaign.backend.db.models import (
Campaign,
CampaignJob,
CampaignSchedule,
CampaignShare,
CampaignVersion,
)
from govoplan_campaign.backend.routes.campaigns import (
_campaign_copy_configuration,
archive_campaign_version,
copy_campaign,
delete_draft_campaign,
@@ -26,6 +28,7 @@ from govoplan_campaign.backend.schemas import (
CampaignLifecycleMutationRequest,
)
from govoplan_core.db.base import Base
from govoplan_core.core.change_sequence import ChangeSequenceEntry
class _Principal:
@@ -65,6 +68,8 @@ class CampaignLifecycleTests(unittest.TestCase):
CampaignVersion.__table__,
CampaignShare.__table__,
CampaignJob.__table__,
CampaignSchedule.__table__,
ChangeSequenceEntry.__table__,
],
)
self.SessionLocal = sessionmaker(
@@ -73,7 +78,14 @@ class CampaignLifecycleTests(unittest.TestCase):
expire_on_commit=False,
)
with self.SessionLocal() as session:
session.execute(access_users.insert().values(id="user-1"))
user_values = {"id": "user-1"}
if "tenant_id" in access_users.c:
user_values.update(
tenant_id="tenant-1",
account_id="account-1",
email="user-1@example.test",
)
session.execute(access_users.insert().values(**user_values))
campaign = Campaign(
id="campaign-1",
tenant_id="tenant-1",
@@ -153,6 +165,50 @@ class CampaignLifecycleTests(unittest.TestCase):
self.assertFalse(policy["actions"]["archive_campaign"]["allowed"])
self.assertIn("Active or uncertain", policy["actions"]["archive_campaign"]["reason"])
def test_schedule_evidence_blocks_destructive_lifecycle_actions(self) -> None:
with self.SessionLocal() as session:
session.add(
CampaignSchedule(
id="schedule-1",
tenant_id="tenant-1",
campaign_id="campaign-1",
source_version_id="version-2",
created_by_user_id="user-1",
name="Recurring draft",
recurrence_kind="daily",
interval_count=1,
timezone="UTC",
starts_at=datetime(2026, 8, 8, tzinfo=UTC),
next_fire_at=datetime(2026, 8, 8, tzinfo=UTC),
max_occurrences=2,
copy_options={},
source_snapshot={"schema": "test"},
source_snapshot_hash="a" * 64,
)
)
session.commit()
active_policy = self._policy(session)
self.assertFalse(active_policy["actions"]["archive_campaign"]["allowed"])
self.assertIn(
"Pause active",
active_policy["actions"]["archive_campaign"]["reason"],
)
self.assertFalse(active_policy["actions"]["delete_campaign"]["allowed"])
self.assertIn(
"schedule evidence",
active_policy["actions"]["delete_campaign"]["reason"],
)
schedule = session.get(CampaignSchedule, "schedule-1")
assert schedule is not None
schedule.active = False
schedule.resource_revision += 1
session.commit()
paused_policy = self._policy(session)
self.assertTrue(paused_policy["actions"]["archive_campaign"]["allowed"])
self.assertFalse(paused_policy["actions"]["delete_campaign"]["allowed"])
def test_stale_delete_token_is_rejected(self) -> None:
with self.SessionLocal() as session:
policy = self._policy(session)
@@ -203,6 +259,10 @@ class CampaignLifecycleTests(unittest.TestCase):
def test_whole_campaign_copy_starts_without_operational_evidence(self) -> None:
with self.SessionLocal() as session:
source_campaign = session.get(Campaign, "campaign-1")
assert source_campaign is not None
source_campaign.settings = {"retention": "source-policy"}
source_campaign.mail_profile_policy = {"profile_id": "mail-profile-1"}
session.add_all(
(
CampaignShare(
@@ -269,6 +329,8 @@ class CampaignLifecycleTests(unittest.TestCase):
CampaignCopyRequest(
source_version_id="version-2",
expected_state_token=policy["state_token"],
include_policies=False,
include_mail_profile=True,
),
session=session,
principal=self.principal,
@@ -277,6 +339,13 @@ class CampaignLifecycleTests(unittest.TestCase):
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")
copied_campaign = session.get(Campaign, "campaign-copy")
assert copied_campaign is not None
self.assertEqual({}, copied_campaign.settings)
self.assertEqual(
{"profile_id": "mail-profile-1"},
copied_campaign.mail_profile_policy,
)
self.assertEqual(
session.query(CampaignJob)
.filter(CampaignJob.campaign_id == "campaign-copy")
@@ -290,6 +359,64 @@ class CampaignLifecycleTests(unittest.TestCase):
0,
)
def test_copy_choices_reset_only_selected_configuration_domains(self) -> None:
source = {
"version": "1.0",
"campaign": {"id": "campaign-1", "name": "Campaign"},
"fields": [{"name": "case_id"}],
"global_values": {"sender": "Office"},
"recipients": {"to": [{"email": "team@example.test"}]},
"entries": {
"inline": [
{
"id": "one",
"to": [{"email": "person@example.test"}],
"attachments": [{"base_dir": ".", "file_filter": "one.pdf"}],
}
]
},
"template": {"subject": "Hello", "text": "Body"},
"attachments": {
"global": [{"base_dir": ".", "file_filter": "global.pdf"}]
},
"validation_policy": {"missing_required_attachment": "warn"},
"server": {"mail_profile_id": "profile-1"},
"delivery": {"rate_limit": {"messages_per_minute": 20}},
}
copied = _campaign_copy_configuration(
source,
CampaignCopyRequest(
source_version_id="version-2",
expected_state_token="a" * 64,
include_recipients=True,
include_files=False,
include_policies=False,
include_mail_profile=False,
),
)
self.assertEqual(source["attachments"]["global"][0]["file_filter"], "global.pdf")
self.assertEqual({}, copied["attachments"])
self.assertEqual([], copied["entries"]["inline"][0]["attachments"])
self.assertEqual("person@example.test", copied["entries"]["inline"][0]["to"][0]["email"])
self.assertEqual({}, copied["validation_policy"])
self.assertEqual({}, copied["server"])
self.assertEqual(20, copied["delivery"]["rate_limit"]["messages_per_minute"])
def test_copy_without_recipient_permission_is_allowed_when_recipient_data_is_excluded(self) -> None:
principal = _Principal("campaigns:campaign:read", "campaigns:campaign:copy")
with self.SessionLocal() as session:
campaign = session.get(Campaign, "campaign-1")
assert campaign is not None
policy = campaign_lifecycle_policy(
session,
campaign=campaign,
principal=principal,
version_id="version-2",
)
self.assertTrue(policy["actions"]["copy_campaign"]["allowed"])
if __name__ == "__main__":
unittest.main()