Add governed tabular and LDAP address sources

tests/test_ldap_sync.py

@@ -0,0 +1,172 @@
+from __future__ import annotations
+
+import unittest
+
+from sqlalchemy import create_engine
+from sqlalchemy.orm import sessionmaker
+
+from govoplan_addresses.backend.db.models import AddressBook, AddressSyncSource, Contact
+from govoplan_addresses.backend.ldap import (
+    AddressLdapClient,
+    AddressLdapEntry,
+    AddressLdapError,
+    AddressLdapSearchResult,
+)
+from govoplan_addresses.backend.ldap_schemas import AddressLdapSourceCreateRequest
+from govoplan_addresses.backend.service import (
+    create_ldap_sync_source,
+    preview_sync_source,
+    run_sync_source,
+)
+from govoplan_core.db.base import Base
+
+
+class Principal:
+    account_id = "account-1"
+    group_ids = frozenset()
+
+    @property
+    def tenant_id(self) -> str:
+        return "tenant-1"
+
+    def has(self, scope: str) -> bool:
+        return scope in {
+            "addresses:address_book:read",
+            "addresses:address_book:write",
+            "addresses:contact:read",
+            "addresses:contact:write",
+            "addresses:contact:delete",
+            "addresses:sync:read",
+            "addresses:sync:write",
+        }
+
+
+class FakeLdapClient:
+    def __init__(self, entries: list[AddressLdapEntry], *, complete: bool = True) -> None:
+        self.entries = entries
+        self.complete = complete
+
+    def search(self, *, base_dn: str, search_filter: str, attributes: tuple[str, ...], page_size: int, max_entries: int) -> AddressLdapSearchResult:
+        del search_filter, attributes, max_entries
+        return AddressLdapSearchResult(
+            base_dn=base_dn,
+            entries=tuple(self.entries),
+            complete=self.complete,
+            page_size=page_size,
+        )
+
+
+def ldap_entry(
+    key: str,
+    *,
+    revision: str = "20260802090000Z",
+    organization: str = "Analysis Office",
+) -> AddressLdapEntry:
+    return AddressLdapEntry(
+        dn=f"uid={key},ou=people,dc=example,dc=test",
+        attributes={
+            "entryUUID": key,
+            "modifyTimestamp": revision,
+            "displayName": "Ada Lovelace",
+            "givenName": "Ada",
+            "sn": "Lovelace",
+            "mail": "ada@example.test",
+            "o": organization,
+            "memberOf": ["cn=analysts,ou=groups,dc=example,dc=test"],
+        },
+    )
+
+
+class AddressLdapSyncTests(unittest.TestCase):
+    def setUp(self) -> None:
+        engine = create_engine("sqlite+pysqlite:///:memory:")
+        Base.metadata.create_all(engine)
+        self.session = sessionmaker(bind=engine, expire_on_commit=False)()
+        self.principal = Principal()
+        self.book = AddressBook(
+            tenant_id="tenant-1",
+            scope_type="tenant",
+            scope_id="tenant-1",
+            name="Directory",
+            source_kind="local",
+            read_only=False,
+        )
+        self.session.add(self.book)
+        self.session.flush()
+        self.source = create_ldap_sync_source(
+            self.session,
+            self.principal,
+            self.book.id,
+            AddressLdapSourceCreateRequest(
+                url="ldaps://directory.example.test",
+                display_name="Corporate directory",
+                base_dn="ou=people,dc=example,dc=test",
+            ),
+        )
+        self.session.flush()
+
+    def test_preview_is_nonmutating_and_full_sync_is_idempotent(self) -> None:
+        client = FakeLdapClient([ldap_entry("person-1")])
+        preview = preview_sync_source(self.session, self.principal, self.source.id, client=client)
+        self.assertEqual(1, preview.stats.created)
+        self.assertEqual(0, self.session.query(Contact).count())
+
+        first = run_sync_source(self.session, self.principal, self.source.id, client=client)
+        self.assertEqual(1, first.stats.created)
+        contact = self.session.query(Contact).one()
+        self.assertEqual("ldap", contact.source_kind)
+        self.assertEqual("person-1", contact.provenance["ldap"]["source_key"])
+        self.assertEqual("succeeded", self.source.status)
+
+        repeated = run_sync_source(self.session, self.principal, self.source.id, client=client)
+        self.assertEqual(1, repeated.stats.unchanged)
+        self.assertEqual(1, self.session.query(Contact).count())
+
+        changed_client = FakeLdapClient(
+            [ldap_entry("person-1", revision="20260802100000Z", organization="Computing Office")]
+        )
+        changed = run_sync_source(self.session, self.principal, self.source.id, client=changed_client)
+        self.assertEqual(1, changed.stats.updated)
+        self.assertEqual("Computing Office", self.session.query(Contact).one().organization)
+
+    def test_only_complete_scans_plan_authoritative_deletes(self) -> None:
+        run_sync_source(
+            self.session,
+            self.principal,
+            self.source.id,
+            client=FakeLdapClient([ldap_entry("person-1")]),
+        )
+        incomplete = preview_sync_source(
+            self.session,
+            self.principal,
+            self.source.id,
+            client=FakeLdapClient([], complete=False),
+        )
+        self.assertEqual(0, incomplete.stats.deleted)
+        self.assertEqual(1, incomplete.stats.errors)
+        self.assertIsNone(self.session.query(Contact).one().deleted_at)
+
+        complete = run_sync_source(
+            self.session,
+            self.principal,
+            self.source.id,
+            client=FakeLdapClient([], complete=True),
+        )
+        self.assertEqual(1, complete.stats.deleted)
+        self.assertIsNotNone(self.session.query(Contact).one().deleted_at)
+
+    def test_connector_requires_encrypted_transport(self) -> None:
+        with self.assertRaisesRegex(AddressLdapError, "require StartTLS"):
+            AddressLdapClient(url="ldap://directory.example.test", start_tls=False)
+        with self.assertRaisesRegex(AddressLdapError, "must not contain credentials"):
+            AddressLdapClient(url="ldaps://user:secret@directory.example.test")
+
+    def test_source_is_always_read_only(self) -> None:
+        source = self.session.get(AddressSyncSource, self.source.id)
+        self.assertTrue(source.read_only)
+        self.assertEqual("read_only", source.sync_direction)
+        self.assertTrue(source.address_book.read_only)
+
+
+if __name__ == "__main__":
+    unittest.main()