422 lines
15 KiB
Python
422 lines
15 KiB
Python
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass
|
|
from types import SimpleNamespace
|
|
from unittest.mock import patch
|
|
|
|
import pytest
|
|
|
|
from govoplan_campaign.backend.documentation import CAMPAIGN_USER_DOCUMENTATION, documentation_topics
|
|
from govoplan_core.core.modules import DocumentationContext
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class _Principal:
|
|
scopes: frozenset[str]
|
|
tenant_id: str = "tenant-1"
|
|
|
|
def has(self, scope: str) -> bool:
|
|
namespace, resource, _action = scope.split(":", 2)
|
|
return scope in self.scopes or f"{namespace}:{resource}:*" in self.scopes or f"{namespace}:*" in self.scopes or "*:*" in self.scopes
|
|
|
|
|
|
class _Registry:
|
|
def __init__(self, integrations: set[str], *, interface_only: set[str] | None = None, capability_only: set[str] | None = None) -> None:
|
|
interfaces = integrations | (interface_only or set())
|
|
self._capabilities = integrations | (capability_only or set())
|
|
self._manifests = (
|
|
SimpleNamespace(
|
|
provides_interfaces=tuple(SimpleNamespace(name=name) for name in sorted(interfaces)),
|
|
),
|
|
)
|
|
|
|
def manifests(self):
|
|
return self._manifests
|
|
|
|
def has_capability(self, name: str) -> bool:
|
|
return name in self._capabilities
|
|
|
|
|
|
def _topics(scopes: set[str], integrations: set[str] | None = None, *, documentation_type: str = "user"):
|
|
return documentation_topics(
|
|
DocumentationContext(
|
|
registry=_Registry(integrations or set()),
|
|
principal=_Principal(frozenset(scopes)),
|
|
documentation_type=documentation_type, # type: ignore[arg-type]
|
|
)
|
|
)
|
|
|
|
|
|
def test_campaign_runtime_documentation_provider_is_registered() -> None:
|
|
from govoplan_campaign.backend.manifest import get_manifest
|
|
|
|
assert documentation_topics in get_manifest().documentation_providers
|
|
|
|
|
|
def test_runtime_documentation_is_user_only_and_requires_a_campaign_task() -> None:
|
|
assert _topics({"docs:documentation:read"}) == ()
|
|
assert _topics({"campaigns:campaign:read"}, documentation_type="admin") == ()
|
|
|
|
|
|
def test_runtime_documentation_reflects_actor_authority_without_exposing_scopes() -> None:
|
|
topic = _topics(
|
|
{
|
|
"campaigns:campaign:read",
|
|
"campaigns:campaign:create",
|
|
"campaigns:campaign:update",
|
|
"campaigns:campaign:validate",
|
|
"campaigns:campaign:build",
|
|
"campaigns:recipient:read",
|
|
"campaigns:recipient:write",
|
|
"campaigns:recipient:import",
|
|
}
|
|
)[0]
|
|
|
|
configuration = topic.metadata["current_configuration"]
|
|
assert "Role authorization: Create new campaigns." in configuration
|
|
assert "Role authorization: Build exact recipient messages for review." in configuration
|
|
assert not any("queue" in item.lower() for item in configuration)
|
|
assert not any("campaigns:" in item or "files:" in item or "mail:" in item for item in configuration)
|
|
|
|
|
|
def test_runtime_documentation_requires_both_interface_and_capability() -> None:
|
|
integration = "mail.campaign_delivery"
|
|
scopes = {"campaigns:campaign:read", "campaigns:campaign:update", "mail:profile:use"}
|
|
|
|
for registry in (
|
|
_Registry(set(), interface_only={integration}),
|
|
_Registry(set(), capability_only={integration}),
|
|
):
|
|
topic = documentation_topics(
|
|
DocumentationContext(registry=registry, principal=_Principal(frozenset(scopes)), documentation_type="user")
|
|
)[0]
|
|
assert not any("profile picker" in item for item in topic.metadata["current_configuration"])
|
|
|
|
topic = _topics(scopes, {integration})[0]
|
|
assert any("Mail's actor-filtered profile picker" in item for item in topic.metadata["current_configuration"])
|
|
|
|
|
|
def test_mail_delivery_actions_are_not_presented_without_the_mail_contract() -> None:
|
|
scopes = {
|
|
"campaigns:campaign:read",
|
|
"campaigns:campaign:queue",
|
|
"campaigns:campaign:send",
|
|
"campaigns:campaign:retry",
|
|
}
|
|
|
|
without_mail = _topics(scopes)[0]
|
|
with_mail = _topics(scopes, {"mail.campaign_delivery"})[0]
|
|
|
|
assert not any("Queue an eligible" in item for item in without_mail.metadata["current_configuration"])
|
|
assert any("Queue an eligible" in item for item in with_mail.metadata["current_configuration"])
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("integrations", "scopes", "expected"),
|
|
[
|
|
(
|
|
{"files.campaign_attachments"},
|
|
{"campaigns:campaign:read", "campaigns:recipient:read", "files:file:read"},
|
|
"Managed campaign attachments can be previewed",
|
|
),
|
|
(
|
|
{"addresses.lookup"},
|
|
{"campaigns:campaign:read", "campaigns:recipient:read"},
|
|
"Address records can be looked up",
|
|
),
|
|
(
|
|
{"addresses.recipient_source"},
|
|
{
|
|
"campaigns:campaign:read",
|
|
"campaigns:campaign:update",
|
|
"campaigns:recipient:read",
|
|
"campaigns:recipient:write",
|
|
"campaigns:recipient:import",
|
|
},
|
|
"traceable recipient snapshot",
|
|
),
|
|
(
|
|
{"notifications.dispatch"},
|
|
{"campaigns:campaign:read"},
|
|
"status changes",
|
|
),
|
|
],
|
|
)
|
|
def test_runtime_documentation_reflects_optional_composition_permutations(
|
|
integrations: set[str],
|
|
scopes: set[str],
|
|
expected: str,
|
|
) -> None:
|
|
topic = _topics(scopes, integrations)[0]
|
|
assert any(expected in item for item in topic.metadata["current_configuration"])
|
|
|
|
|
|
def test_runtime_documentation_full_composition_uses_only_user_facing_names() -> None:
|
|
integrations = {
|
|
"mail.campaign_delivery",
|
|
"files.campaign_attachments",
|
|
"addresses.lookup",
|
|
"addresses.recipient_source",
|
|
"notifications.dispatch",
|
|
}
|
|
topic = _topics({"*:*"}, integrations)[0]
|
|
rendered = "\n".join(
|
|
(
|
|
topic.title,
|
|
topic.summary,
|
|
topic.body,
|
|
*topic.metadata["current_configuration"],
|
|
*topic.metadata["limitations"],
|
|
)
|
|
)
|
|
|
|
assert "Mail's actor-filtered profile picker" in rendered
|
|
assert "Managed file versions" in rendered
|
|
assert "Address records" in rendered
|
|
assert "In-app notifications" in rendered
|
|
assert topic.metadata["help_contexts"] == ["campaigns.list", "campaign.overview"]
|
|
for technical_name in integrations:
|
|
assert technical_name not in rendered
|
|
assert "0.1." not in rendered
|
|
assert "0.2." not in rendered
|
|
assert "hostname" not in rendered.lower()
|
|
assert "secret" not in rendered.lower()
|
|
|
|
|
|
def test_runtime_documentation_states_the_effective_synchronous_limit() -> None:
|
|
session = SimpleNamespace(
|
|
get=lambda _model, _id: SimpleNamespace(
|
|
settings={
|
|
"campaign_delivery_policy": {
|
|
"synchronous_send_max_recipients": 12,
|
|
}
|
|
}
|
|
)
|
|
)
|
|
with patch.dict("os.environ", {"GOVOPLAN_CAMPAIGN_SYNCHRONOUS_SEND_MAX_RECIPIENTS": "25"}):
|
|
topic = documentation_topics(
|
|
DocumentationContext(
|
|
registry=_Registry({"mail.campaign_delivery"}),
|
|
principal=_Principal(frozenset({"campaigns:campaign:read", "campaigns:campaign:send"})),
|
|
session=session,
|
|
documentation_type="user",
|
|
)
|
|
)[0]
|
|
|
|
assert any(
|
|
"Send now is limited to 12 eligible recipient job(s)" in item
|
|
for item in topic.metadata["current_configuration"]
|
|
)
|
|
|
|
|
|
def _visible_static_topics(
|
|
scopes: set[str],
|
|
*,
|
|
modules: set[str] | None = None,
|
|
capabilities: set[str] | None = None,
|
|
) -> set[str]:
|
|
installed = modules or {"campaigns"}
|
|
available_capabilities = capabilities or set()
|
|
principal = _Principal(frozenset(scopes))
|
|
visible: set[str] = set()
|
|
for topic in CAMPAIGN_USER_DOCUMENTATION:
|
|
condition = topic.conditions[0]
|
|
if not set(condition.required_modules).issubset(installed):
|
|
continue
|
|
if not set(condition.required_capabilities).issubset(available_capabilities):
|
|
continue
|
|
if not all(principal.has(scope) for scope in condition.required_scopes):
|
|
continue
|
|
visible.add(topic.id)
|
|
return visible
|
|
|
|
|
|
def test_campaign_manager_sees_only_authoring_tasks_from_the_static_handbook() -> None:
|
|
visible = _visible_static_topics(
|
|
{
|
|
"campaigns:campaign:read",
|
|
"campaigns:campaign:create",
|
|
"campaigns:campaign:update",
|
|
"campaigns:campaign:copy",
|
|
"campaigns:campaign:validate",
|
|
"campaigns:campaign:build",
|
|
"campaigns:recipient:read",
|
|
"campaigns:recipient:write",
|
|
"campaigns:recipient:import",
|
|
"campaigns:report:read",
|
|
}
|
|
)
|
|
|
|
assert {
|
|
"campaigns.workflow.create-campaign",
|
|
"campaigns.workflow.create-editable-successor",
|
|
"campaigns.workflow.import-recipients",
|
|
"campaigns.workflow.view-delivery-report",
|
|
}.issubset(visible)
|
|
assert "campaigns.workflow.queue-delivery" not in visible
|
|
assert "campaigns.workflow.send-small-controlled-run" not in visible
|
|
assert "campaigns.workflow.export-delivery-report" not in visible
|
|
assert "campaigns.workflow.share-campaign" not in visible
|
|
assert "campaigns.workflow.archive-campaign" not in visible
|
|
assert "campaigns.workflow.delete-untouched-draft" not in visible
|
|
assert "campaigns.workflow.create-campaign" not in _visible_static_topics({"campaigns:campaign:create"})
|
|
|
|
|
|
def test_sender_sees_queue_and_send_only_with_the_mail_contract_and_profile_authority() -> None:
|
|
scopes = {
|
|
"campaigns:campaign:read",
|
|
"campaigns:campaign:queue",
|
|
"campaigns:campaign:send",
|
|
"campaigns:recipient:read",
|
|
"campaigns:report:read",
|
|
"mail:profile:use",
|
|
}
|
|
|
|
without_mail = _visible_static_topics(scopes)
|
|
with_mail = _visible_static_topics(
|
|
scopes,
|
|
modules={"campaigns", "mail"},
|
|
capabilities={"mail.campaign_delivery"},
|
|
)
|
|
|
|
assert "campaigns.workflow.queue-delivery" not in without_mail
|
|
assert "campaigns.workflow.send-small-controlled-run" not in without_mail
|
|
assert "campaigns.workflow.queue-delivery" in with_mail
|
|
assert "campaigns.workflow.send-small-controlled-run" in with_mail
|
|
|
|
queue_topic = next(
|
|
topic for topic in CAMPAIGN_USER_DOCUMENTATION
|
|
if topic.id == "campaigns.workflow.queue-delivery"
|
|
)
|
|
assert any(link.href == "/operator" for link in queue_topic.links)
|
|
|
|
|
|
def test_connected_authoring_tasks_require_their_declared_contracts_and_permissions() -> None:
|
|
attachment_scopes = {
|
|
"campaigns:campaign:read",
|
|
"campaigns:campaign:update",
|
|
"campaigns:campaign:validate",
|
|
"campaigns:recipient:read",
|
|
"files:file:read",
|
|
"files:file:share",
|
|
}
|
|
source_scopes = {
|
|
"campaigns:campaign:read",
|
|
"campaigns:campaign:update",
|
|
"campaigns:recipient:read",
|
|
"campaigns:recipient:write",
|
|
"campaigns:recipient:import",
|
|
}
|
|
|
|
files_visible = _visible_static_topics(
|
|
attachment_scopes,
|
|
modules={"campaigns", "files"},
|
|
capabilities={"files.campaign_attachments"},
|
|
)
|
|
addresses_visible = _visible_static_topics(
|
|
source_scopes,
|
|
modules={"campaigns", "addresses"},
|
|
capabilities={"addresses.recipient_source"},
|
|
)
|
|
|
|
assert "campaigns.workflow.use-managed-attachments" in files_visible
|
|
assert "campaigns.workflow.import-address-source" in addresses_visible
|
|
assert "campaigns.workflow.use-managed-attachments" not in _visible_static_topics(attachment_scopes)
|
|
assert "campaigns.workflow.import-address-source" not in _visible_static_topics(source_scopes)
|
|
|
|
|
|
def test_report_export_and_lifecycle_tasks_are_independently_permission_gated() -> None:
|
|
exporter = _visible_static_topics(
|
|
{
|
|
"campaigns:campaign:read",
|
|
"campaigns:report:read",
|
|
"campaigns:report:export",
|
|
"campaigns:recipient:read",
|
|
"campaigns:recipient:export",
|
|
}
|
|
)
|
|
custodian = _visible_static_topics(
|
|
{
|
|
"campaigns:campaign:read",
|
|
"campaigns:campaign:share",
|
|
"campaigns:campaign:archive",
|
|
"campaigns:campaign:delete",
|
|
}
|
|
)
|
|
|
|
assert "campaigns.workflow.export-delivery-report" in exporter
|
|
assert "campaigns.workflow.view-delivery-report" in exporter
|
|
assert "campaigns.workflow.share-campaign" in custodian
|
|
assert "campaigns.workflow.archive-campaign" in custodian
|
|
assert "campaigns.workflow.delete-untouched-draft" in custodian
|
|
assert "campaigns.workflow.export-delivery-report" not in custodian
|
|
|
|
|
|
def test_aggregate_report_task_never_implies_recipient_detail_or_export_authority() -> None:
|
|
visible = _visible_static_topics({"campaigns:report:read"})
|
|
|
|
assert visible == {"campaigns.workflow.view-aggregate-delivery-report"}
|
|
topic = next(
|
|
item for item in CAMPAIGN_USER_DOCUMENTATION
|
|
if item.id == "campaigns.workflow.view-aggregate-delivery-report"
|
|
)
|
|
assert topic.metadata["route"] == "/reports"
|
|
assert "export" in topic.metadata["verification"].lower()
|
|
|
|
|
|
def test_static_campaign_handbook_has_unique_ids_help_contexts_and_no_planned_resend_claim() -> None:
|
|
from govoplan_campaign.backend.manifest import get_manifest
|
|
|
|
topics = get_manifest().documentation
|
|
ids = [topic.id for topic in topics]
|
|
rendered_static = "\n".join(
|
|
(topic.title + "\n" + topic.summary + "\n" + topic.body).lower()
|
|
for topic in CAMPAIGN_USER_DOCUMENTATION
|
|
)
|
|
known_help_contexts = {
|
|
"campaigns.list",
|
|
"campaign.overview",
|
|
"campaign.settings",
|
|
"campaign.fields",
|
|
"campaign.template",
|
|
"campaign.attachments",
|
|
"campaign.recipients",
|
|
"campaign.recipient-data",
|
|
"campaign.server-settings",
|
|
"campaign.global-settings",
|
|
"campaign.review-send",
|
|
"campaign.report",
|
|
"campaign.audit",
|
|
"campaign.json",
|
|
}
|
|
|
|
assert len(ids) == len(set(ids))
|
|
assert "single resend" not in rendered_static
|
|
for topic in CAMPAIGN_USER_DOCUMENTATION:
|
|
assert topic.metadata["kind"] == "workflow"
|
|
assert topic.metadata["prerequisites"]
|
|
assert topic.metadata["steps"]
|
|
assert topic.metadata["outcome"]
|
|
assert topic.metadata["verification"]
|
|
assert set(topic.metadata["help_contexts"]).issubset(known_help_contexts)
|
|
|
|
existing_user_workflows = {
|
|
topic.id: topic
|
|
for topic in topics
|
|
if topic.id
|
|
in {
|
|
"campaigns.mail-profile-user-journey",
|
|
"campaigns.workflow.prepare-validate-and-build",
|
|
"campaigns.workflow.complete-review",
|
|
"campaigns.workflow.retry-and-reconcile",
|
|
}
|
|
}
|
|
assert set(existing_user_workflows) == {
|
|
"campaigns.mail-profile-user-journey",
|
|
"campaigns.workflow.prepare-validate-and-build",
|
|
"campaigns.workflow.complete-review",
|
|
"campaigns.workflow.retry-and-reconcile",
|
|
}
|
|
for topic in existing_user_workflows.values():
|
|
assert topic.metadata["help_contexts"]
|