209 lines
7.7 KiB
Python
209 lines
7.7 KiB
Python
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()
|