feat: add governed LDIF contact imports
This commit is contained in:
@@ -0,0 +1,208 @@
|
||||
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()
|
||||
Reference in New Issue
Block a user