from __future__ import annotations import unittest from govoplan_core.core.dataflows import ( CAPABILITY_DATAFLOW_RUN_LIFECYCLE, DataflowRunDescriptor, DataflowRunLifecycleProvider, dataflow_run_lifecycle, ) from govoplan_core.core.modules import ModuleContext, ModuleManifest from govoplan_core.core.registry import PlatformRegistry class _Provider: def start_run(self, session, principal, *, request): del session, principal return DataflowRunDescriptor( ref="dataflow-run:1", pipeline_ref=request.pipeline_ref, revision=request.revision, status="succeeded", definition_hash="abc", executor_version="test", ) def get_run(self, session, principal, *, run_ref): del session, principal if run_ref != "dataflow-run:1": return None return DataflowRunDescriptor( ref=run_ref, pipeline_ref="pipeline:1", revision=1, status="succeeded", definition_hash="abc", executor_version="test", ) def cancel_run(self, session, principal, *, run_ref): del session, principal return DataflowRunDescriptor( ref=run_ref, pipeline_ref="pipeline:1", revision=1, status="cancelled", definition_hash="abc", executor_version="test", ) class DataflowContractTests(unittest.TestCase): def test_run_lifecycle_is_runtime_checkable_and_resolved(self) -> None: provider = _Provider() self.assertIsInstance(provider, DataflowRunLifecycleProvider) registry = PlatformRegistry() registry.register( ModuleManifest( id="dataflow_contract_test", name="Dataflow contract test", version="test", capability_factories={ CAPABILITY_DATAFLOW_RUN_LIFECYCLE: lambda context: provider, }, ) ) registry.configure_capability_context( ModuleContext(registry=registry, settings=object()) ) self.assertIs(provider, dataflow_run_lifecycle(registry)) self.assertIsNone(dataflow_run_lifecycle(PlatformRegistry())) if __name__ == "__main__": unittest.main()