govoplan-connectors/tests/test_provider_state.py

from __future__ import annotations

from datetime import UTC, datetime
import unittest

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

from govoplan_connectors.backend.db.models import (
    ConnectorKnowledgeObject,
    ConnectorKnowledgeProfile,
    ConnectorKnowledgeSyncRun,
    ConnectorSanctionsAcquisitionRun,
    ConnectorSanctionsSnapshot,
    ConnectorTabularSource,
)
from govoplan_connectors.backend.manifest import manifest
from govoplan_connectors.backend.provider_state import (
    KNOWLEDGE_PROVIDER_ID,
    SANCTIONS_PROVIDER_ID,
    TABULAR_PROVIDER_ID,
    knowledge_provider_states,
    sanctions_provider_states,
    tabular_provider_states,
)
from govoplan_core.core.provider_governance import ExternalProviderStateContext
from govoplan_core.db.base import Base


class ConnectorsProviderStateTests(unittest.TestCase):
    def setUp(self) -> None:
        self.engine = create_engine("sqlite+pysqlite:///:memory:", future=True)
        Base.metadata.create_all(
            self.engine,
        )
        self.session = sessionmaker(bind=self.engine, expire_on_commit=False)()

    def tearDown(self) -> None:
        self.session.close()
        self.engine.dispose()

    def test_immutable_tabular_snapshot_reports_ready_state(self) -> None:
        self.session.add(
            ConnectorTabularSource(
                id="tabular-1",
                tenant_id="tenant-1",
                source_name="monthly",
                name="Monthly input",
                status="active",
                schema_version=1,
                schema_=[{"name": "id", "type": "string"}],
                rows=[{"id": "1"}],
                fingerprint="a" * 64,
                row_count=1,
                byte_count=10,
            )
        )
        self.session.commit()

        state = tabular_provider_states(
            ExternalProviderStateContext(session=self.session, tenant_id="tenant-1")
        )[0]

        self.assertEqual("healthy", state.health)
        self.assertEqual("ready", state.recovery)
        self.assertEqual("not_applicable", state.freshness)

    def test_sanctions_state_hashes_binding_and_manifest_registers_state(self) -> None:
        now = datetime.now(UTC)
        run = ConnectorSanctionsAcquisitionRun(
            id="run-1",
            tenant_id="tenant-1",
            provider_id="eu",
            source_id="secret-source-name",
            status="succeeded",
            attempt_count=1,
            started_at=now,
            finished_at=now,
        )
        snapshot = ConnectorSanctionsSnapshot(
            id="snapshot-1",
            tenant_id="tenant-1",
            provider_id="eu",
            publisher="European Union",
            jurisdiction="EU",
            list_type="sanctions",
            source_id="secret-source-name",
            source_version="2026-08-01",
            acquired_at=now,
            source_url="https://source.example.test/list.xml",
            content_type="application/xml",
            byte_count=8,
            sha256="b" * 64,
            parser_version="1",
            connector_run_id=run.id,
            raw_content=b"<list/>",
        )
        run.snapshot_id = snapshot.id
        self.session.add_all((run, snapshot))
        self.session.commit()

        state = sanctions_provider_states(
            ExternalProviderStateContext(session=self.session, tenant_id="tenant-1")
        )[0]

        self.assertEqual("healthy", state.health)
        self.assertEqual("ready", state.recovery)
        rendered = str(state.to_dict())
        self.assertNotIn("secret-source-name", rendered)
        self.assertNotIn("source.example.test", rendered)
        self.assertEqual(
            {TABULAR_PROVIDER_ID, SANCTIONS_PROVIDER_ID, KNOWLEDGE_PROVIDER_ID},
            {item.provider_id for item in manifest.external_provider_state_providers},
        )

    def test_external_knowledge_state_reports_health_without_acl_or_endpoint_data(self) -> None:
        now = datetime.now(UTC)
        profile = ConnectorKnowledgeProfile(
            id="knowledge-profile-1",
            tenant_id="tenant-1",
            configuration_id="configuration-secret",
            status="active",
            product="bluespice",
            product_version="4.5.2",
            desired_maturity="migrate",
            discovered_maturity="migrate",
            source_authority_mode="external_mirror",
            default_visibility="restricted",
            default_acl_tokens=["group:secret-acl"],
            namespace_mappings=[{"secret": "mapping"}],
            capabilities=["read", "synchronize", "migrate"],
            health_status="healthy",
            discovered_at=now,
        )
        page = ConnectorKnowledgeObject(
            id="knowledge-object-1",
            tenant_id="tenant-1",
            profile_id=profile.id,
            object_type="page",
            external_id="42",
            title="Secret page title",
            status="active",
            source_revision="501",
            content_hash="c" * 64,
            visibility="restricted",
            acl_tokens=["group:secret-acl"],
            observed_at=now,
        )
        run = ConnectorKnowledgeSyncRun(
            id="knowledge-run-1",
            tenant_id="tenant-1",
            profile_id=profile.id,
            mode="delta",
            idempotency_key="secret-key",
            request_hash="d" * 64,
            status="completed",
            started_at=now,
            finished_at=now,
        )
        self.session.add_all((profile, page, run))
        self.session.commit()

        state = knowledge_provider_states(
            ExternalProviderStateContext(session=self.session, tenant_id="tenant-1")
        )[0]

        self.assertEqual("healthy", state.health)
        self.assertEqual("ready", state.recovery)
        self.assertEqual(1, state.metrics["active_objects"])
        rendered = str(state.to_dict())
        self.assertNotIn("Secret page title", rendered)
        self.assertNotIn("group:secret-acl", rendered)
        self.assertNotIn("configuration-secret", rendered)


if __name__ == "__main__":
    unittest.main()