134 lines
4.5 KiB
Python
134 lines
4.5 KiB
Python
from __future__ import annotations
|
|
|
|
from datetime import UTC, datetime
|
|
import unittest
|
|
|
|
from sqlalchemy import create_engine
|
|
from sqlalchemy.orm import sessionmaker
|
|
|
|
from govoplan_addresses.backend.db.models import (
|
|
AddressBook,
|
|
AddressSyncConflict,
|
|
AddressSyncDiagnostic,
|
|
AddressSyncSource,
|
|
)
|
|
from govoplan_addresses.backend.manifest import manifest
|
|
from govoplan_addresses.backend.provider_state import (
|
|
CARDDAV_PROVIDER_ID,
|
|
LDAP_PROVIDER_ID,
|
|
carddav_provider_states,
|
|
ldap_provider_states,
|
|
)
|
|
from govoplan_core.core.provider_governance import ExternalProviderStateContext
|
|
from govoplan_core.db.base import Base
|
|
|
|
|
|
class AddressesProviderStateTests(unittest.TestCase):
|
|
def setUp(self) -> None:
|
|
self.engine = create_engine("sqlite+pysqlite:///:memory:", future=True)
|
|
Base.metadata.create_all(
|
|
self.engine,
|
|
tables=(
|
|
AddressBook.__table__,
|
|
AddressSyncSource.__table__,
|
|
AddressSyncConflict.__table__,
|
|
AddressSyncDiagnostic.__table__,
|
|
),
|
|
)
|
|
self.session = sessionmaker(bind=self.engine, expire_on_commit=False)()
|
|
book = AddressBook(
|
|
id="book-1",
|
|
tenant_id="tenant-1",
|
|
scope_type="tenant",
|
|
scope_id="tenant-1",
|
|
name="Remote",
|
|
)
|
|
self.source = AddressSyncSource(
|
|
id="source-1",
|
|
tenant_id="tenant-1",
|
|
address_book_id=book.id,
|
|
connector_type="carddav",
|
|
display_name="CardDAV",
|
|
external_address_book_ref="https://dav.example.test/addressbook/",
|
|
sync_direction="two_way",
|
|
read_only=False,
|
|
enabled=True,
|
|
status="succeeded",
|
|
last_success_at=datetime.now(UTC),
|
|
)
|
|
self.session.add_all((book, self.source))
|
|
self.session.commit()
|
|
|
|
def tearDown(self) -> None:
|
|
self.session.close()
|
|
self.engine.dispose()
|
|
|
|
def test_state_is_tenant_bounded_secret_free_and_reports_conflict(self) -> None:
|
|
healthy = carddav_provider_states(
|
|
ExternalProviderStateContext(session=self.session, tenant_id="tenant-1")
|
|
)[0]
|
|
|
|
self.assertEqual("healthy", healthy.health)
|
|
self.assertEqual("current", healthy.freshness)
|
|
self.assertEqual("governed_sync", healthy.authority_mode)
|
|
self.assertNotIn("dav.example.test", str(healthy.to_dict()))
|
|
self.assertEqual(
|
|
(),
|
|
carddav_provider_states(
|
|
ExternalProviderStateContext(
|
|
session=self.session,
|
|
tenant_id="tenant-2",
|
|
)
|
|
),
|
|
)
|
|
|
|
self.session.add(
|
|
AddressSyncConflict(
|
|
tenant_id="tenant-1",
|
|
sync_source_id=self.source.id,
|
|
address_book_id="book-1",
|
|
field_path="email",
|
|
status="open",
|
|
)
|
|
)
|
|
self.session.flush()
|
|
|
|
conflicted = carddav_provider_states(
|
|
ExternalProviderStateContext(session=self.session, tenant_id="tenant-1")
|
|
)[0]
|
|
self.assertEqual("pending", conflicted.conflict)
|
|
self.assertEqual("attention", conflicted.recovery)
|
|
|
|
def test_manifest_registers_carddav_declaration_and_state(self) -> None:
|
|
self.assertEqual(CARDDAV_PROVIDER_ID, manifest.external_providers[0].id)
|
|
self.assertEqual(
|
|
CARDDAV_PROVIDER_ID,
|
|
manifest.external_provider_state_providers[0].provider_id,
|
|
)
|
|
self.assertEqual(LDAP_PROVIDER_ID, manifest.external_providers[1].id)
|
|
self.assertEqual(
|
|
LDAP_PROVIDER_ID,
|
|
manifest.external_provider_state_providers[1].provider_id,
|
|
)
|
|
|
|
def test_failed_ldap_source_is_stale_without_exposing_endpoint(self) -> None:
|
|
self.source.connector_type = "ldap"
|
|
self.source.display_name = "Directory"
|
|
self.source.status = "failed"
|
|
self.source.last_error = "connection failed"
|
|
self.session.flush()
|
|
|
|
state = ldap_provider_states(
|
|
ExternalProviderStateContext(session=self.session, tenant_id="tenant-1")
|
|
)[0]
|
|
|
|
self.assertEqual(LDAP_PROVIDER_ID, state.provider_id)
|
|
self.assertEqual("error", state.health)
|
|
self.assertEqual("current", state.freshness)
|
|
self.assertEqual("external_authoritative", state.authority_mode)
|
|
self.assertNotIn("dav.example.test", str(state.to_dict()))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|