from __future__ import annotations import base64 import unittest from sqlalchemy import create_engine from sqlalchemy.orm import sessionmaker from govoplan_addresses.backend.db.models import AddressBook, Contact from govoplan_addresses.backend.import_schemas import ( AddressImportConfiguration, AddressImportPreviewRequest, AddressImportProfileCreateRequest, ) from govoplan_addresses.backend.imports import ( apply_address_import, create_import_profile, import_run_payload, preview_address_import, ) from govoplan_addresses.backend.ldif import parse_ldif_rows 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", } def encoded(value: bytes | str) -> str: raw = value.encode() if isinstance(value, str) else value return base64.b64encode(raw).decode() class AddressLdifImportTests(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="Imported contacts", source_kind="local", read_only=False, ) self.session.add(self.book) def create_profile(self, **configuration_overrides): configuration = AddressImportConfiguration( field_mappings={ "source_key": "dn", "display_name": "cn", "given_name": "givenName", "family_name": "sn", "email": "mail", "phone": "telephoneNumber", "organization": "o", }, **configuration_overrides, ) profile = create_import_profile( self.session, self.principal, AddressImportProfileCreateRequest( scope_type="tenant", name="Directory export", source_format="ldif", configuration=configuration, ), ) self.session.flush() return profile def test_parser_unfolds_and_decodes_text_without_projecting_binary_or_urls(self) -> None: rows, diagnostics = parse_ldif_rows( b"version: 1\n\n" b"# exported contact\n" b"dn: uid=ada,ou=people,dc=example,dc=test\n" b"cn:: QWRhIExvdmVsYWNl\n" b"sn: Love\n" b" lace\n" b"mail: ada@example.test\n" b"mail: ada.work@example.test\n" b"jpegPhoto:: /9j/\n" b"seeAlso:< https://example.test/contact/ada\n", max_entries=10, ) self.assertEqual(1, len(rows)) self.assertEqual(["Ada Lovelace"], rows[0][1]["cn"]) self.assertEqual(["Lovelace"], rows[0][1]["sn"]) self.assertEqual(["ada@example.test", "ada.work@example.test"], rows[0][1]["mail"]) self.assertNotIn("jpegphoto", rows[0][1]) self.assertNotIn("seealso", rows[0][1]) self.assertEqual( {"ldif_binary_value_ignored", "ldif_url_value_ignored"}, {item["code"] for item in diagnostics}, ) def test_preview_apply_and_repeat_preserve_multivalue_provenance(self) -> None: profile = self.create_profile(default_tags=["ldif"]) source = ( "dn: uid=ada,ou=people,dc=example,dc=test\n" "cn: Ada Lovelace\n" "givenName: Ada\n" "sn: Lovelace\n" "mail: ada@example.test\n" "mail: ada.work@example.test\n" "telephoneNumber: +49 30 123\n" "o: Analysis Office\n" ) request = AddressImportPreviewRequest( profile_id=profile.id, filename="contacts.ldif", content_base64=encoded(source), ) run = preview_address_import(self.session, self.principal, self.book.id, request) self.assertEqual(1, run.statistics["create"]) self.assertFalse([item for item in run.diagnostics if item["severity"] == "error"]) apply_address_import(self.session, self.principal, run.id, expected_plan_hash=run.plan_hash) contact = self.session.query(Contact).one() self.assertEqual(["ada@example.test", "ada.work@example.test"], [item.email for item in contact.emails]) self.assertEqual("ldif", contact.source_kind) self.assertEqual("ldif", contact.provenance["import"]["source_format"]) self.assertEqual(64, len(contact.provenance["import"]["source_record_hash"])) self.assertNotIn(source, repr(import_run_payload(run))) repeated = preview_address_import(self.session, self.principal, self.book.id, request) self.assertEqual(1, repeated.statistics["unchanged"]) apply_address_import(self.session, self.principal, repeated.id, expected_plan_hash=repeated.plan_hash) self.assertEqual(1, self.session.query(Contact).count()) def test_change_records_are_rejected_by_default_and_add_requires_explicit_policy(self) -> None: source = ( "dn: uid=ada,ou=people,dc=example,dc=test\n" "changetype: add\n" "cn: Ada Lovelace\n" "mail: ada@example.test\n" ) rejected_profile = self.create_profile() rejected = preview_address_import( self.session, self.principal, self.book.id, AddressImportPreviewRequest( profile_id=rejected_profile.id, filename="changes.ldif", content_base64=encoded(source), ), ) self.assertEqual(0, rejected.row_count) self.assertIn("ldif_change_record_rejected", {item["code"] for item in rejected.diagnostics}) self.assertFalse(import_run_payload(rejected)["can_apply"]) allowed_profile = self.create_profile(ldif_change_record_policy="treat_add_as_entry") allowed = preview_address_import( self.session, self.principal, self.book.id, AddressImportPreviewRequest( profile_id=allowed_profile.id, filename="changes.ldif", content_base64=encoded(source), ), ) self.assertEqual(1, allowed.statistics["create"]) self.assertIn("ldif_add_record_imported", {item["code"] for item in allowed.diagnostics}) def test_invalid_base64_is_a_correction_diagnostic_and_blocks_apply(self) -> None: profile = self.create_profile() run = preview_address_import( self.session, self.principal, self.book.id, AddressImportPreviewRequest( profile_id=profile.id, filename="broken.ldif", content_base64=encoded( "dn: uid=ada,dc=example,dc=test\n" "cn:: this-is-not-base64!\n" "mail: ada@example.test\n" ), ), ) self.assertIn("ldif_invalid_base64", {item["code"] for item in run.diagnostics}) self.assertFalse(import_run_payload(run)["can_apply"]) with self.assertRaisesRegex(ValueError, "error diagnostics"): apply_address_import(self.session, self.principal, run.id, expected_plan_hash=run.plan_hash) if __name__ == "__main__": unittest.main()