246 lines
9.4 KiB
Python
246 lines
9.4 KiB
Python
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import json
|
|
from pathlib import Path
|
|
import unittest
|
|
|
|
from sqlalchemy import create_engine
|
|
from sqlalchemy.orm import Session
|
|
|
|
from govoplan_core.core.configuration_packages import (
|
|
ConfigurationApplyResult,
|
|
ConfigurationExportResult,
|
|
ConfigurationPackageManifest,
|
|
ConfigurationPlanItem,
|
|
ConfigurationPreflightContext,
|
|
ConfigurationPreflightResult,
|
|
ConfigurationProviderDescription,
|
|
configuration_package_claim_issues,
|
|
dry_run_configuration_package,
|
|
)
|
|
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
PACKAGE_ROOT = ROOT / "packages"
|
|
|
|
|
|
class ConfigurationPackageArtifactTests(unittest.TestCase):
|
|
def test_product_package_manifests_are_portable_and_evidence_backed(self) -> None:
|
|
paths = tuple(sorted(PACKAGE_ROOT.glob("*/*/package.json")))
|
|
self.assertGreaterEqual(len(paths), 2)
|
|
package_ids: set[str] = set()
|
|
|
|
for path in paths:
|
|
manifest = ConfigurationPackageManifest.from_mapping(
|
|
json.loads(path.read_text(encoding="utf-8"))
|
|
)
|
|
self.assertNotIn(manifest.package_id, package_ids)
|
|
package_ids.add(manifest.package_id)
|
|
self.assertEqual((), configuration_package_claim_issues(manifest))
|
|
for evidence in manifest.evidence:
|
|
evidence_path = ROOT / evidence.reference
|
|
self.assertTrue(
|
|
evidence_path.is_file(),
|
|
f"Missing evidence {evidence.reference} for {manifest.package_id}",
|
|
)
|
|
if evidence.checksum is not None:
|
|
self.assertEqual(
|
|
evidence.checksum,
|
|
"sha256:" + hashlib.sha256(evidence_path.read_bytes()).hexdigest(),
|
|
f"Stale evidence checksum for {manifest.package_id}: {evidence.reference}",
|
|
)
|
|
for requirement in (
|
|
*manifest.required_modules,
|
|
*manifest.optional_modules,
|
|
):
|
|
repository = ROOT.parent / (
|
|
"govoplan-" + requirement.module_id.replace("_", "-")
|
|
)
|
|
self.assertTrue(
|
|
repository.is_dir(),
|
|
f"Missing repository for {requirement.module_id}",
|
|
)
|
|
|
|
provider_module_ids = tuple(
|
|
sorted({fragment.module_id for fragment in manifest.fragments})
|
|
)
|
|
providers = tuple(_ArtifactProvider(module_id) for module_id in provider_module_ids)
|
|
supplied_data = {
|
|
str(item["key"]): _sample_value(item)
|
|
for item in manifest.data_requirements
|
|
}
|
|
result = dry_run_configuration_package(
|
|
manifest,
|
|
providers,
|
|
ConfigurationPreflightContext(
|
|
supplied_data=supplied_data,
|
|
installed_modules={
|
|
item.module_id: item.version or "workspace"
|
|
for item in manifest.required_modules
|
|
},
|
|
capabilities=frozenset(manifest.required_capabilities),
|
|
),
|
|
)
|
|
self.assertFalse(
|
|
any(item.severity == "blocker" for item in result.diagnostics),
|
|
tuple(item.to_dict() for item in result.diagnostics),
|
|
)
|
|
|
|
self.assertIn("product.governed-communication", package_ids)
|
|
self.assertIn("product.governed-data-assurance", package_ids)
|
|
self.assertIn("product.service-to-decision", package_ids)
|
|
|
|
def test_service_to_decision_package_imports_its_form_as_an_idempotent_local_draft(self) -> None:
|
|
from govoplan_forms.backend.configuration_provider import (
|
|
_apply_definition,
|
|
_preflight_definition,
|
|
)
|
|
from govoplan_forms.backend.db.models import FormDefinitionRevision
|
|
from govoplan_forms.backend.service import get_form_definition
|
|
|
|
package = ConfigurationPackageManifest.from_mapping(json.loads(
|
|
(PACKAGE_ROOT / "product/service-to-decision/package.json").read_text(
|
|
encoding="utf-8"
|
|
)
|
|
))
|
|
fragment = next(
|
|
item
|
|
for item in package.fragments
|
|
if item.module_id == "forms" and item.fragment_type == "definition"
|
|
)
|
|
context = ConfigurationPreflightContext(
|
|
tenant_id="tenant-reference-test",
|
|
operator_user_id="operator-1",
|
|
operator_scopes=frozenset({"system:governance:write"}),
|
|
)
|
|
engine = create_engine("sqlite+pysqlite:///:memory:")
|
|
FormDefinitionRevision.__table__.create(engine)
|
|
session = Session(engine)
|
|
try:
|
|
preflight = _preflight_definition(session, fragment, context)
|
|
applied = _apply_definition(session, fragment, context)
|
|
session.commit()
|
|
replay = _apply_definition(session, fragment, context)
|
|
imported = get_form_definition(
|
|
session,
|
|
type("Principal", (), {"tenant_id": "tenant-reference-test"})(),
|
|
form_id="resident-parking-permit-application",
|
|
)
|
|
finally:
|
|
session.close()
|
|
engine.dispose()
|
|
|
|
self.assertFalse(any(item.severity == "blocker" for item in preflight.diagnostics))
|
|
self.assertEqual("create", preflight.plan[0].action)
|
|
self.assertEqual(1, len(applied.created_refs))
|
|
self.assertEqual({}, replay.created_refs)
|
|
self.assertIsNotNone(imported)
|
|
assert imported is not None
|
|
self.assertEqual("tenant-reference-test", imported.reference.tenant_id)
|
|
self.assertEqual("draft", imported.publication_state)
|
|
self.assertEqual("de", imported.fallback_locale)
|
|
|
|
def test_service_to_decision_package_materializes_its_tenant_workflow_idempotently(self) -> None:
|
|
from govoplan_core.core.configuration_packages import _resolve_fragment_data_references
|
|
from govoplan_workflow_engine.backend.configuration_provider import (
|
|
apply_workflow_definitions,
|
|
preflight_workflow_definitions,
|
|
)
|
|
from govoplan_workflow_engine.backend.db.models import (
|
|
WorkflowDefinition,
|
|
WorkflowDefinitionRevision,
|
|
)
|
|
|
|
package = ConfigurationPackageManifest.from_mapping(json.loads(
|
|
(PACKAGE_ROOT / "product/service-to-decision/package.json").read_text(
|
|
encoding="utf-8"
|
|
)
|
|
))
|
|
fragment = next(
|
|
item
|
|
for item in package.fragments
|
|
if item.module_id == "workflow_engine"
|
|
)
|
|
context = ConfigurationPreflightContext(
|
|
tenant_id="tenant-reference-test",
|
|
supplied_data={
|
|
"responsible_group_slug": "traffic-permits",
|
|
"responsible_group_name": "Traffic permits",
|
|
},
|
|
)
|
|
resolved = _resolve_fragment_data_references(
|
|
fragment,
|
|
context.supplied_data,
|
|
)
|
|
engine = create_engine("sqlite+pysqlite:///:memory:")
|
|
WorkflowDefinition.__table__.create(engine)
|
|
WorkflowDefinitionRevision.__table__.create(engine)
|
|
session = Session(engine)
|
|
try:
|
|
preflight = preflight_workflow_definitions(session, resolved, context)
|
|
applied = apply_workflow_definitions(
|
|
session,
|
|
resolved,
|
|
context,
|
|
registry=None,
|
|
)
|
|
replay = apply_workflow_definitions(
|
|
session,
|
|
resolved,
|
|
context,
|
|
registry=None,
|
|
)
|
|
session.commit()
|
|
finally:
|
|
session.close()
|
|
engine.dispose()
|
|
|
|
self.assertFalse(any(item.severity == "blocker" for item in preflight.diagnostics))
|
|
self.assertEqual("create", preflight.plan[0].action)
|
|
self.assertIn("resident-parking-permit-review", applied.created_refs)
|
|
self.assertEqual({}, replay.created_refs)
|
|
self.assertEqual({}, replay.updated_refs)
|
|
|
|
|
|
class _ArtifactProvider:
|
|
def __init__(self, module_id: str) -> None:
|
|
self.module_id = module_id
|
|
|
|
def describe(self) -> ConfigurationProviderDescription:
|
|
return ConfigurationProviderDescription(module_id=self.module_id)
|
|
|
|
def preflight(self, fragment, context) -> ConfigurationPreflightResult:
|
|
del context
|
|
return ConfigurationPreflightResult(plan=(ConfigurationPlanItem(
|
|
action="create",
|
|
module_id=fragment.module_id,
|
|
fragment_type=fragment.fragment_type,
|
|
fragment_id=fragment.fragment_id,
|
|
),))
|
|
|
|
def apply(self, fragment, supplied_data, context) -> ConfigurationApplyResult:
|
|
del fragment, supplied_data, context
|
|
return ConfigurationApplyResult()
|
|
|
|
def export(self, selection, context) -> ConfigurationExportResult:
|
|
del selection, context
|
|
return ConfigurationExportResult()
|
|
|
|
def health(self, import_result, context):
|
|
del import_result, context
|
|
return ()
|
|
|
|
|
|
def _sample_value(requirement: dict[str, object]) -> object:
|
|
data_type = str(requirement.get("data_type") or requirement.get("type") or "string")
|
|
if data_type == "boolean":
|
|
return False
|
|
if data_type in {"integer", "number"}:
|
|
return 1
|
|
return f"fixture-{requirement['key']}"
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|