feat: integrate governed address contacts

tests/test_address_integration.py

@@ -0,0 +1,141 @@
+from __future__ import annotations
+
+import unittest
+from types import SimpleNamespace
+from unittest.mock import patch
+
+from fastapi import HTTPException
+from pydantic import ValidationError
+
+from govoplan_mail.backend.router import (
+    create_mail_address_contact,
+    list_mail_address_write_targets,
+    lookup_mail_addresses,
+)
+from govoplan_mail.backend.schemas import MailContactCreateRequest
+
+
+class _Session:
+    def __init__(self) -> None:
+        self.commits = 0
+        self.rollbacks = 0
+
+    def commit(self) -> None:
+        self.commits += 1
+
+    def rollback(self) -> None:
+        self.rollbacks += 1
+
+
+class _Writer:
+    def __init__(self, *, allowed: bool = True, read_only: bool = False) -> None:
+        self.allowed = allowed
+        self.read_only = read_only
+        self.created_payload = None
+        self.created_provenance = None
+
+    def list_write_targets(self, _session, _principal, *, operation):
+        return (
+            SimpleNamespace(
+                address_book_id="book-1",
+                address_book_label="Personal contacts",
+                operation=operation,
+                allowed=self.allowed,
+                reason="allowed" if self.allowed else "read_only_source",
+                message="Contact can be added." if self.allowed else "This source is read-only.",
+                scope_type="user",
+                scope_id="user-1",
+                source_kind="local" if self.allowed else "ldap",
+                read_only=self.read_only,
+                required_scopes=("addresses:contacts:write",),
+                provenance={"policy": "addresses"},
+            ),
+        )
+
+    def can_write_to_address_book(self, _session, _principal, *, address_book_id, operation):
+        return self.list_write_targets(_session, _principal, operation=operation)[0]
+
+    def create_contact(self, _session, _principal, *, address_book_id, payload, provenance):
+        self.created_payload = payload
+        self.created_provenance = provenance
+        return SimpleNamespace(
+            contact_id="contact-1",
+            address_book_id=address_book_id,
+            display_name=payload["display_name"],
+            email=payload["emails"][0]["email"],
+            source_kind="local",
+            provenance=provenance,
+        )
+
+
+def _principal():
+    return SimpleNamespace(has=lambda scope: scope == "mail:profile:use")
+
+
+class MailAddressIntegrationTests(unittest.TestCase):
+    def test_optional_capabilities_fail_open_for_mail(self) -> None:
+        with patch("govoplan_mail.backend.router._registry_capability", return_value=None):
+            lookup = lookup_mail_addresses(query="ada", limit=25, session=_Session(), principal=_principal())
+            targets = list_mail_address_write_targets(session=_Session(), principal=_principal())
+
+        self.assertFalse(lookup.available)
+        self.assertEqual(lookup.candidates, [])
+        self.assertFalse(targets.available)
+        self.assertEqual(targets.targets, [])
+
+    def test_write_target_preserves_read_only_decision(self) -> None:
+        writer = _Writer(allowed=False, read_only=True)
+        with patch("govoplan_mail.backend.router._registry_capability", return_value=writer):
+            response = list_mail_address_write_targets(session=_Session(), principal=_principal())
+
+        self.assertTrue(response.available)
+        self.assertFalse(response.targets[0].allowed)
+        self.assertTrue(response.targets[0].read_only)
+        self.assertEqual(response.targets[0].reason, "read_only_source")
+
+    def test_blocked_target_cannot_be_bypassed_by_create(self) -> None:
+        session = _Session()
+        writer = _Writer(allowed=False, read_only=True)
+        with (
+            patch("govoplan_mail.backend.router._registry_capability", return_value=writer),
+            self.assertRaises(HTTPException) as raised,
+        ):
+            create_mail_address_contact(
+                MailContactCreateRequest(
+                    address_book_id="book-1",
+                    display_name="Ada Lovelace",
+                    email="ada@example.test",
+                ),
+                session=session,
+                principal=_principal(),
+            )
+
+        self.assertEqual(raised.exception.status_code, 422)
+        self.assertEqual(session.commits, 0)
+
+    def test_allowed_create_uses_writer_and_records_consumer_provenance(self) -> None:
+        session = _Session()
+        writer = _Writer()
+        with patch("govoplan_mail.backend.router._registry_capability", return_value=writer):
+            result = create_mail_address_contact(
+                MailContactCreateRequest(
+                    address_book_id="book-1",
+                    display_name="Ada Lovelace",
+                    email="ada@example.test",
+                ),
+                session=session,
+                principal=_principal(),
+            )
+
+        self.assertEqual(result.contact_id, "contact-1")
+        self.assertEqual(writer.created_payload["emails"][0]["email"], "ada@example.test")
+        self.assertEqual(writer.created_provenance["consumer_module"], "mail")
+        self.assertEqual(session.commits, 1)
+
+    def test_proxy_rejects_invalid_email_before_calling_writer(self) -> None:
+        with self.assertRaises(ValidationError):
+            MailContactCreateRequest(address_book_id="book-1", email="not-an-email")
+
+
+if __name__ == "__main__":
+    unittest.main()