govoplan-addresses/tests/test_ldap_sync.py

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()