Files
govoplan-core/tests/test_connector_runtime_contract.py

71 lines
2.5 KiB
Python

from __future__ import annotations
import unittest
from datetime import UTC, datetime
from govoplan_core.core.connector_runtime import (
CONNECTOR_RUNTIME_CONTRACT_VERSION,
ConnectorContractError,
ConnectorDryRunResult,
ConnectorEffectPreview,
ConnectorEndpoint,
summarize_connector_effects,
)
class ConnectorRuntimeContractTests(unittest.TestCase):
def test_endpoint_rejects_embedded_credentials(self) -> None:
with self.assertRaisesRegex(ConnectorContractError, "must not contain credentials"):
ConnectorEndpoint(url="ldaps://user:secret@directory.example.test")
def test_dry_run_is_bounded_consistent_and_apply_gated(self) -> None:
effects = (
ConnectorEffectPreview(
effect="create",
source_object_ref="ldap:uid=one",
sample={"display_name": "Example Person"},
),
ConnectorEffectPreview(
effect="unchanged",
source_object_ref="ldap:uid=two",
target_object_ref="contact-2",
),
)
result = ConnectorDryRunResult(
contract_version=CONNECTOR_RUNTIME_CONTRACT_VERSION,
source_ref="source-1",
source_revision="revision-1",
source_fingerprint="a" * 64,
input_hash="b" * 64,
generated_at=datetime(2026, 8, 2, tzinfo=UTC),
summary=summarize_connector_effects(effects),
effects=effects,
apply_token="plan-1",
)
self.assertEqual(1, result.summary.creates)
self.assertEqual(1, result.summary.unchanged)
self.assertTrue(result.can_apply)
def test_summary_must_describe_exact_returned_effects(self) -> None:
with self.assertRaisesRegex(ConnectorContractError, "summary counts"):
ConnectorDryRunResult(
contract_version=CONNECTOR_RUNTIME_CONTRACT_VERSION,
source_ref="source-1",
source_revision="revision-1",
source_fingerprint="a" * 64,
input_hash="b" * 64,
generated_at=datetime(2026, 8, 2, tzinfo=UTC),
summary=summarize_connector_effects(()),
effects=(
ConnectorEffectPreview(
effect="create",
source_object_ref="source:one",
),
),
)
if __name__ == "__main__":
unittest.main()