92 lines
2.7 KiB
Python
92 lines
2.7 KiB
Python
from __future__ import annotations
|
|
|
|
import unittest
|
|
|
|
from fastapi import FastAPI
|
|
|
|
from govoplan_core.core.lifecycle import ModuleLifecycleManager
|
|
from govoplan_core.core.modules import (
|
|
ModuleInterfaceProvider,
|
|
ModuleManifest,
|
|
)
|
|
from govoplan_core.core.registry import PlatformRegistry
|
|
from govoplan_core.core.workflows import (
|
|
CAPABILITY_WORKFLOW_DEFINITION_CONTRIBUTIONS,
|
|
WorkflowDefinitionContribution,
|
|
)
|
|
from govoplan_core.db.session import configure_database, reset_database
|
|
|
|
|
|
class _ContributionProvider:
|
|
def __init__(self) -> None:
|
|
self.calls = 0
|
|
|
|
def reconcile(self, session, *, tenant_ids=()):
|
|
del session, tenant_ids
|
|
self.calls += 1
|
|
return {"created": 1, "updated": 0}
|
|
|
|
|
|
class WorkflowContributionLifecycleTests(unittest.TestCase):
|
|
def setUp(self) -> None:
|
|
configure_database("sqlite:///:memory:")
|
|
|
|
def tearDown(self) -> None:
|
|
reset_database(dispose=True)
|
|
|
|
def test_active_graph_change_reconciles_module_workflow_baselines(self) -> None:
|
|
provider = _ContributionProvider()
|
|
workflow_engine = ModuleManifest(
|
|
id="workflow_engine",
|
|
name="Workflow Engine",
|
|
version="1.0.0",
|
|
provides_interfaces=(
|
|
ModuleInterfaceProvider(
|
|
name="workflow.definition_contributions",
|
|
version="1.0.0",
|
|
),
|
|
),
|
|
capability_factories={
|
|
CAPABILITY_WORKFLOW_DEFINITION_CONTRIBUTIONS: (
|
|
lambda _context: provider
|
|
),
|
|
},
|
|
)
|
|
domain = ModuleManifest(
|
|
id="permits",
|
|
name="Permits",
|
|
version="1.0.0",
|
|
workflow_definitions=(
|
|
WorkflowDefinitionContribution(
|
|
origin_module_id="permits",
|
|
origin_module_version="1.0.0",
|
|
definition_key="approve",
|
|
name="Approve permit",
|
|
graph={"nodes": [], "edges": []},
|
|
),
|
|
),
|
|
)
|
|
registry = PlatformRegistry()
|
|
lifecycle = ModuleLifecycleManager(
|
|
registry=registry,
|
|
available_modules={
|
|
workflow_engine.id: workflow_engine,
|
|
domain.id: domain,
|
|
},
|
|
settings=None,
|
|
)
|
|
lifecycle.attach_app(FastAPI())
|
|
|
|
result = lifecycle.apply_enabled_modules(
|
|
("workflow_engine", "permits"),
|
|
migrate=False,
|
|
protected_modules=(),
|
|
)
|
|
|
|
self.assertEqual(("permits", "workflow_engine"), result.enabled_modules)
|
|
self.assertEqual(1, provider.calls)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|