114 lines
4.7 KiB
Python
114 lines
4.7 KiB
Python
from __future__ import annotations
|
|
|
|
import unittest
|
|
from types import SimpleNamespace
|
|
from unittest.mock import patch
|
|
|
|
from govoplan_core.core.modules import MigrationRetirementPlan, ModuleManifest
|
|
from govoplan_mail.backend.manifest import _mail_retirement_provider, get_manifest
|
|
|
|
|
|
class MailManifestTests(unittest.TestCase):
|
|
def test_manifest_declares_optional_addresses_lookup(self) -> None:
|
|
manifest = get_manifest()
|
|
|
|
self.assertIsInstance(manifest, ModuleManifest)
|
|
self.assertEqual(manifest.id, "mail")
|
|
self.assertIn("addresses", manifest.optional_dependencies)
|
|
self.assertIn("addresses.lookup", {interface.name for interface in manifest.requires_interfaces})
|
|
self.assertIn(
|
|
{
|
|
"name": "campaigns.access",
|
|
"version_min": "0.1.0",
|
|
"version_max_exclusive": "0.2.0",
|
|
"optional": True,
|
|
},
|
|
[
|
|
{
|
|
"name": interface.name,
|
|
"version_min": interface.version_min,
|
|
"version_max_exclusive": interface.version_max_exclusive,
|
|
"optional": interface.optional,
|
|
}
|
|
for interface in manifest.requires_interfaces
|
|
],
|
|
)
|
|
self.assertIn(
|
|
{"name": "mail.campaign_delivery", "version": "0.2.0"},
|
|
[
|
|
{"name": interface.name, "version": interface.version}
|
|
for interface in manifest.provides_interfaces
|
|
],
|
|
)
|
|
topics = {topic.id: topic for topic in manifest.documentation}
|
|
self.assertTrue(
|
|
{
|
|
"mail.workflow.choose-and-test-profile",
|
|
"mail.workflow.read-mailbox",
|
|
"mail.reference.credentials-egress-retirement",
|
|
"mail.reference.campaign-delivery-contract",
|
|
}.issubset(topics)
|
|
)
|
|
ownership = topics["mail.profile-ownership-and-consumers"]
|
|
self.assertEqual(ownership.metadata["kind"], "reference")
|
|
self.assertEqual(ownership.metadata["route"], "/settings?section=mail-profiles")
|
|
self.assertIn("campaigns.mail-profile-user-journey", ownership.metadata["related_topic_ids"])
|
|
|
|
mailbox = topics["mail.workflow.read-mailbox"]
|
|
self.assertEqual(mailbox.metadata["route"], "/mail")
|
|
self.assertEqual(
|
|
mailbox.conditions[0].required_scopes,
|
|
("mail:mailbox:read", "mail:profile:use"),
|
|
)
|
|
|
|
campaign_contract = topics["mail.reference.campaign-delivery-contract"]
|
|
self.assertEqual(campaign_contract.conditions[0].required_modules, ("mail", "campaigns"))
|
|
|
|
def test_retirement_scrubs_credentials_before_dropping_mail_tables(self) -> None:
|
|
events: list[str] = []
|
|
|
|
def table_provider(_session, _module_id):
|
|
return MigrationRetirementPlan(
|
|
supported=True,
|
|
summary="Mail tables",
|
|
destroy_data_supported=True,
|
|
destroy_data_executor=lambda _execute_session, _execute_module_id: events.append("drop"),
|
|
)
|
|
|
|
session = SimpleNamespace(get_bind=lambda: object(), query=lambda *_args: None)
|
|
inspector = SimpleNamespace(has_table=lambda _name: True)
|
|
with (
|
|
patch("govoplan_mail.backend.manifest._mail_table_retirement_provider", table_provider),
|
|
patch("govoplan_mail.backend.manifest.inspect", return_value=inspector),
|
|
patch(
|
|
"govoplan_mail.backend.mail_profiles.delete_mail_profile_credentials_for_retirement",
|
|
side_effect=lambda _session: events.append("scrub"),
|
|
),
|
|
):
|
|
plan = _mail_retirement_provider(session, "mail")
|
|
assert plan.destroy_data_executor is not None
|
|
plan.destroy_data_executor(session, "mail")
|
|
|
|
self.assertEqual(events, ["scrub", "drop"])
|
|
self.assertTrue(any("audit" in warning for warning in plan.destroy_data_warnings))
|
|
|
|
events.clear()
|
|
with (
|
|
patch("govoplan_mail.backend.manifest._mail_table_retirement_provider", table_provider),
|
|
patch("govoplan_mail.backend.manifest.inspect", return_value=inspector),
|
|
patch(
|
|
"govoplan_mail.backend.mail_profiles.delete_mail_profile_credentials_for_retirement",
|
|
side_effect=RuntimeError("audit unavailable"),
|
|
),
|
|
):
|
|
blocked_plan = _mail_retirement_provider(session, "mail")
|
|
assert blocked_plan.destroy_data_executor is not None
|
|
with self.assertRaisesRegex(RuntimeError, "audit unavailable"):
|
|
blocked_plan.destroy_data_executor(session, "mail")
|
|
|
|
self.assertEqual(events, [])
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|