142 lines
5.3 KiB
Python
142 lines
5.3 KiB
Python
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()
|