2 Commits

7 changed files with 322 additions and 11 deletions

View File

@@ -144,10 +144,14 @@ paths, storage keys, worker claim tokens, and raw provider diagnostics require
the dedicated diagnostic permission and must not leak through ordinary campaign,
version, job, or report responses.
The current Campaign Report Web UI additionally requires recipient-read access
and does not yet hide every action control that the actor lacks. The server
still authorizes each action, but an aggregate-only reader UI remains open
work; do not promise that experience from `campaigns:report:read` alone.
Campaign now provides a separate aggregate **Reports** surface for readers with
`campaigns:report:read` and access to the campaign. It loads only the safe
aggregate projections, applies small-cell suppression, and offers no recipient
rows, drill-down, filtering, export, or delivery actions. The recipient-aware
**Campaign Report** still requires recipient-read access and does not yet hide
every action control that the actor lacks. The server authorizes each action,
but permission-aware action visibility on that detailed surface remains open
work; do not confuse it with the aggregate reader experience.
### Deliver and resolve outcomes

View File

@@ -10,7 +10,11 @@ scenario catalogue lives in `examples/README.md`; committed fixture files should
be added under `examples/` only when they validate against the current campaign
schema and are safe to run in non-production environments.
- simple announcement with one active recipient and no attachments
- [`simple-announcement`](../examples/simple-announcement/campaign.json), a
credential-free campaign with one active recipient and no attachments; its
automated acceptance check physically blocks Mail and Files imports, denies
network connections, and validates/builds from an unrelated temporary
workspace
- multi-recipient message with To, CC, BCC, Reply-To, bounce, and disposition
notification fields
- campaign with global attachments and recipient-specific attachment rules
@@ -41,6 +45,8 @@ Before tagging a campaign release:
- Review `examples/README.md` and update the scenario catalogue when a release
adds or removes delivery behavior.
- Run `python -m unittest discover -s tests -p 'test_example_campaigns.py'` and
retain its isolated validate/build result as release evidence.
- Run core module permutation tests with campaign installed both with and
without files/mail.
- Validate and build each maintained example campaign.

View File

@@ -9,7 +9,7 @@ campaign schema and do not require production data.
| Scenario | Required Modules | Release Check |
| --- | --- | --- |
| `simple-announcement` | core, access, campaigns | Validate and build one active recipient without attachments. |
| [`simple-announcement`](simple-announcement/campaign.json) | core, access, campaigns | Validate and build one active recipient without attachments while Mail and Files are absent. |
| `addressing-matrix` | core, access, campaigns | Exercise To, CC, BCC, Reply-To, bounce, and disposition-notification fields. |
| `global-attachment` | core, access, campaigns; optional files | Build one deterministic attachment and verify evidence. |
| `recipient-attachment-rules` | core, access, campaigns; optional files | Match recipient-specific attachment rules and verify per-recipient evidence. |
@@ -37,9 +37,13 @@ campaign schema and do not require production data.
Before a release tag:
1. Run module permutation startup checks from core.
2. Validate every committed example fixture against the current campaign schema.
3. Build exact messages for each fixture.
4. Run the mock-delivery example when the dev mailbox capability is enabled.
5. Run `dev/mail-testbed/run_transport_smoke.py`.
6. Execute the delivery checklist in
2. Run `python -m unittest discover -s tests -p 'test_example_campaigns.py'`
from this repository. The acceptance test copies each maintained fixture to
an unrelated temporary workspace before using Campaign's public loader,
validator, and message builder.
3. Validate every committed example fixture against the current campaign schema.
4. Build exact messages for each fixture.
5. Run the mock-delivery example when the dev mailbox capability is enabled.
6. Run `dev/mail-testbed/run_transport_smoke.py`.
7. Execute the delivery checklist in
`docs/EXAMPLE_CAMPAIGNS_AND_RELEASE_CHECKLIST.md`.

View File

@@ -0,0 +1,54 @@
{
"version": "1.0",
"campaign": {
"id": "simple-announcement",
"name": "Simple announcement",
"description": "Credential-free release fixture for Campaign validation and message building.",
"mode": "test"
},
"fields": [
{
"name": "display_name",
"type": "string",
"label": "Display name",
"required": true
}
],
"recipients": {
"from": [
{
"email": "announcements@example.test",
"name": "GovOPlaN Example",
"type": "to"
}
],
"allow_individual_to": true
},
"template": {
"subject": "Planned service maintenance for ${display_name}",
"text": "Hello ${display_name},\n\nThe example service will be unavailable during the announced maintenance window.\n\nThis message was built locally and was not sent.\n",
"body_mode": "text"
},
"attachments": {
"base_path": ".",
"send_without_attachments_behavior": "continue",
"global": []
},
"entries": {
"inline": [
{
"id": "example-recipient",
"to": [
{
"email": "recipient@example.test",
"name": "Example Recipient",
"type": "to"
}
],
"fields": {
"display_name": "Example Recipient"
}
}
]
}
}

View File

@@ -0,0 +1,23 @@
{
"schema_version": 1,
"id": "simple-announcement",
"campaign_file": "campaign.json",
"required_modules": [
"core",
"access",
"campaigns"
],
"absent_optional_modules": [
"files",
"mail"
],
"external_effects": "forbidden",
"expected": {
"campaign_id": "simple-announcement",
"entries_count": 1,
"built_count": 1,
"queueable_count": 1,
"attachment_count": 0,
"subject": "Planned service maintenance for Example Recipient"
}
}

View File

@@ -1,6 +1,7 @@
from __future__ import annotations
from dataclasses import dataclass
from pathlib import Path
from types import SimpleNamespace
from unittest.mock import patch
@@ -364,6 +365,18 @@ def test_aggregate_report_task_never_implies_recipient_detail_or_export_authorit
assert "export" in topic.metadata["verification"].lower()
def test_handbook_distinguishes_shipped_aggregate_reports_from_detailed_report_gaps() -> None:
handbook = " ".join(
(
Path(__file__).resolve().parents[1] / "docs" / "CAMPAIGN_HANDBOOK.md"
).read_text(encoding="utf-8").lower().split()
)
assert "aggregate-only reader ui remains open" not in handbook
assert "separate aggregate **reports** surface" in handbook
assert "permission-aware action visibility on that detailed surface remains open work" in handbook
def test_static_campaign_handbook_has_unique_ids_help_contexts_and_no_planned_resend_claim() -> None:
from govoplan_campaign.backend.manifest import get_manifest

View File

@@ -0,0 +1,207 @@
from __future__ import annotations
import json
import shutil
import subprocess
import sys
import tempfile
import unittest
from pathlib import Path
REPOSITORY_ROOT = Path(__file__).resolve().parents[1]
FIXTURE_ROOT = REPOSITORY_ROOT / "examples" / "simple-announcement"
_ISOLATED_ACCEPTANCE_PROGRAM = r"""
from __future__ import annotations
import hashlib
import importlib.abc
import json
import socket
import sys
from email import policy
from email.parser import BytesParser
from pathlib import Path
source_root = Path(sys.argv[1]).resolve()
fixture_root = Path(sys.argv[2]).resolve()
sys.path.insert(0, str(source_root))
class AbsentOptionalModuleFinder(importlib.abc.MetaPathFinder):
absent_roots = {"govoplan_files", "govoplan_mail"}
def find_spec(self, fullname, path=None, target=None):
if fullname.partition(".")[0] in self.absent_roots:
raise ModuleNotFoundError(
f"{fullname} is intentionally absent in the Campaign fixture check",
name=fullname,
)
return None
sys.meta_path.insert(0, AbsentOptionalModuleFinder())
def deny_network(*args, **kwargs):
raise AssertionError("the Campaign validate/build fixture must not open a network connection")
class NoNetworkSocket(socket.socket):
def connect(self, address):
deny_network(address)
def connect_ex(self, address):
deny_network(address)
socket.create_connection = deny_network
socket.socket = NoNetworkSocket
from govoplan_campaign.backend.campaign import ( # noqa: E402
load_campaign_config,
load_campaign_json,
validate_campaign_config,
)
from govoplan_campaign.backend.messages import build_campaign_messages # noqa: E402
metadata = json.loads((fixture_root / "fixture.json").read_text(encoding="utf-8"))
assert metadata["required_modules"] == ["core", "access", "campaigns"]
assert metadata["absent_optional_modules"] == ["files", "mail"]
assert metadata["external_effects"] == "forbidden"
campaign_file = fixture_root / metadata["campaign_file"]
raw_campaign = load_campaign_json(campaign_file)
def iter_keys(value):
if isinstance(value, dict):
for key, nested in value.items():
yield key.casefold()
yield from iter_keys(nested)
elif isinstance(value, list):
for nested in value:
yield from iter_keys(nested)
forbidden_key_fragments = ("credential", "imap", "mail_profile", "password", "secret", "smtp")
assert not [
key
for key in iter_keys(raw_campaign)
if any(fragment in key for fragment in forbidden_key_fragments)
]
config = load_campaign_config(campaign_file)
assert config.server.mail_profile_id is None
assert not config.attachments.global_
validation = validate_campaign_config(config, campaign_file=campaign_file, check_files=True)
assert validation.ok
assert validation.error_count == 0
assert validation.warning_count == 0
assert validation.entries_count == metadata["expected"]["entries_count"]
def build(output_name):
return build_campaign_messages(
config,
campaign_file=campaign_file,
output_dir=fixture_root.parent / output_name,
write_eml=True,
)
first = build("build-first")
second = build("build-second")
expected = metadata["expected"]
for result in (first, second):
assert result.report.campaign_id == expected["campaign_id"]
assert result.report.entries_count == expected["entries_count"]
assert result.report.built_count == expected["built_count"]
assert result.report.build_failed_count == 0
assert result.report.queueable_count == expected["queueable_count"]
assert len(result.built_messages) == 1
built = result.built_messages[0]
assert built.mime is not None
assert built.draft.subject == expected["subject"]
assert built.draft.validation_status.value == "ready"
assert built.draft.send_status.value == "draft"
assert built.draft.imap_status.value == "not_requested"
assert built.draft.attachment_count == expected["attachment_count"]
assert not built.draft.attachments
assert not built.draft.issues
assert built.draft.from_ is not None
assert built.draft.from_.email == "announcements@example.test"
assert [address.email for address in built.draft.to] == ["recipient@example.test"]
assert built.mime["Subject"] == expected["subject"]
assert "Hello Example Recipient" in built.mime.get_content()
assert list(built.mime.iter_attachments()) == []
def normalized_eml(path_value):
message = BytesParser(policy=policy.default).parsebytes(Path(path_value).read_bytes())
del message["Date"]
del message["Message-ID"]
return message.as_bytes(policy=policy.default)
first_eml = normalized_eml(first.report.messages[0].eml_path)
second_eml = normalized_eml(second.report.messages[0].eml_path)
assert first_eml == second_eml
assert not any(
name == root or name.startswith(root + ".")
for name in sys.modules
for root in ("govoplan_files", "govoplan_mail")
)
print(json.dumps({
"campaign_id": first.report.campaign_id,
"built_count": first.report.built_count,
"queueable_count": first.report.queueable_count,
"normalized_eml_sha256": hashlib.sha256(first_eml).hexdigest(),
}, sort_keys=True))
"""
class CampaignExampleAcceptanceTests(unittest.TestCase):
def test_simple_announcement_validates_and_builds_without_mail_or_files(self) -> None:
self.assertTrue((FIXTURE_ROOT / "campaign.json").is_file())
self.assertTrue((FIXTURE_ROOT / "fixture.json").is_file())
with tempfile.TemporaryDirectory(prefix="govoplan-campaign-acceptance-", dir="/tmp") as temp_dir:
workspace = Path(temp_dir).resolve()
self.assertNotIn(REPOSITORY_ROOT, workspace.parents)
isolated_fixture = workspace / "simple-announcement"
shutil.copytree(FIXTURE_ROOT, isolated_fixture)
completed = subprocess.run(
[
sys.executable,
"-I",
"-c",
_ISOLATED_ACCEPTANCE_PROGRAM,
str(REPOSITORY_ROOT / "src"),
str(isolated_fixture),
],
cwd=workspace,
check=False,
capture_output=True,
text=True,
timeout=30,
)
self.assertEqual(completed.returncode, 0, completed.stderr or completed.stdout)
evidence = json.loads(completed.stdout)
self.assertEqual(evidence["campaign_id"], "simple-announcement")
self.assertEqual(evidence["built_count"], 1)
self.assertEqual(evidence["queueable_count"], 1)
self.assertRegex(evidence["normalized_eml_sha256"], r"^[0-9a-f]{64}$")
if __name__ == "__main__":
unittest.main()