From 8b1910b5b772b30c73e99cab53ad2449881c1995 Mon Sep 17 00:00:00 2001 From: Albrecht Degering Date: Tue, 28 Jul 2026 12:42:49 +0200 Subject: [PATCH] feat: add datasource and definition graph contracts --- src/govoplan_core/core/datasources.py | 391 ++++++++++++++++++++ src/govoplan_core/core/definition_graphs.py | 342 +++++++++++++++++ src/govoplan_core/settings.py | 2 +- tests/test_datasource_contract.py | 164 ++++++++ tests/test_definition_graphs.py | 145 ++++++++ webui/package-lock.json | 22 ++ webui/package.json | 7 +- webui/scripts/test-module-permutations.mjs | 5 +- webui/src/definitionGraph.ts | 147 ++++++++ webui/src/platform/modules.ts | 3 +- webui/tests/definition-graph.test.ts | 68 ++++ webui/tsconfig.module-tests.json | 2 + webui/vite.config.ts | 3 + 13 files changed, 1297 insertions(+), 4 deletions(-) create mode 100644 src/govoplan_core/core/datasources.py create mode 100644 src/govoplan_core/core/definition_graphs.py create mode 100644 tests/test_datasource_contract.py create mode 100644 tests/test_definition_graphs.py create mode 100644 webui/src/definitionGraph.ts create mode 100644 webui/tests/definition-graph.test.ts diff --git a/src/govoplan_core/core/datasources.py b/src/govoplan_core/core/datasources.py new file mode 100644 index 0000000..b2017f5 --- /dev/null +++ b/src/govoplan_core/core/datasources.py @@ -0,0 +1,391 @@ +from __future__ import annotations + +from collections.abc import Mapping, Sequence +from dataclasses import dataclass, field +from datetime import datetime +from typing import Literal, Protocol, runtime_checkable + + +CAPABILITY_DATASOURCE_CATALOGUE = "datasources.catalogue" +CAPABILITY_DATASOURCE_LIFECYCLE = "datasources.lifecycle" +CAPABILITY_DATASOURCE_ORIGINS = "connectors.datasourceOrigins" + +DatasourceMode = Literal["live", "cached", "static"] +DatasourceKind = Literal[ + "upload", + "database", + "http", + "rest", + "directory", + "file", + "feed", + "custom", +] +DatasourceShape = Literal["tabular", "document", "binary", "directory", "stream"] +DatasourceConsistency = Literal["current", "live", "frozen"] + + +class DatasourceError(ValueError): + """Stable base error for provider-neutral datasource operations.""" + + +class DatasourceNotFoundError(DatasourceError): + pass + + +class DatasourceAccessError(DatasourceError): + pass + + +class DatasourceValidationError(DatasourceError): + pass + + +class DatasourceUnavailableError(DatasourceError): + pass + + +@dataclass(frozen=True, slots=True) +class DatasourceField: + name: str + data_type: str + nullable: bool = True + + +@dataclass(frozen=True, slots=True) +class DatasourceDescriptor: + ref: str + source_name: str + name: str + kind: DatasourceKind + mode: DatasourceMode + shape: DatasourceShape + status: str = "active" + description: str | None = None + provider: str | None = None + provider_ref: str | None = None + schema: tuple[DatasourceField, ...] = () + schema_version: str = "1" + fingerprint: str = "" + current_materialization_ref: str | None = None + row_count: int | None = None + byte_count: int | None = None + updated_at: datetime | None = None + capabilities: tuple[str, ...] = ("read",) + provenance: Mapping[str, object] = field(default_factory=dict) + metadata: Mapping[str, object] = field(default_factory=dict) + + +@dataclass(frozen=True, slots=True) +class DatasourceMaterialization: + ref: str + datasource_ref: str + revision: int + state: str + fingerprint: str + schema: tuple[DatasourceField, ...] = () + row_count: int | None = None + byte_count: int | None = None + frozen_at: datetime | None = None + frozen_label: str | None = None + source_timestamp: datetime | None = None + created_at: datetime | None = None + provenance: Mapping[str, object] = field(default_factory=dict) + metadata: Mapping[str, object] = field(default_factory=dict) + + +@dataclass(frozen=True, slots=True) +class DatasourceStage: + ref: str + name: str + source_name: str + kind: DatasourceKind + mode: DatasourceMode + shape: DatasourceShape + state: str + target_datasource_ref: str | None = None + fingerprint: str = "" + schema: tuple[DatasourceField, ...] = () + row_count: int | None = None + byte_count: int | None = None + validation: Mapping[str, object] = field(default_factory=dict) + created_at: datetime | None = None + promoted_at: datetime | None = None + promoted_materialization_ref: str | None = None + provenance: Mapping[str, object] = field(default_factory=dict) + metadata: Mapping[str, object] = field(default_factory=dict) + + +@dataclass(frozen=True, slots=True) +class DatasourceReadRequest: + datasource_ref: str + materialization_ref: str | None = None + consistency: DatasourceConsistency = "current" + limit: int = 250 + offset: int = 0 + columns: tuple[str, ...] = () + expected_fingerprint: str | None = None + + +@dataclass(frozen=True, slots=True) +class DatasourceReadResult: + datasource: DatasourceDescriptor + rows: tuple[Mapping[str, object], ...] + total_rows: int + truncated: bool + materialization: DatasourceMaterialization | None = None + + +@dataclass(frozen=True, slots=True) +class DatasourceStageInput: + name: str + source_name: str + kind: DatasourceKind + mode: DatasourceMode + shape: DatasourceShape + rows: tuple[Mapping[str, object], ...] + description: str | None = None + target_datasource_ref: str | None = None + provider: str | None = None + provider_ref: str | None = None + provenance: Mapping[str, object] = field(default_factory=dict) + metadata: Mapping[str, object] = field(default_factory=dict) + + +@dataclass(frozen=True, slots=True) +class DatasourceOrigin: + ref: str + source_name: str + name: str + kind: DatasourceKind + shape: DatasourceShape + supported_modes: tuple[DatasourceMode, ...] + provider: str + description: str | None = None + schema: tuple[DatasourceField, ...] = () + schema_version: str = "1" + fingerprint: str = "" + row_count: int | None = None + byte_count: int | None = None + updated_at: datetime | None = None + capabilities: tuple[str, ...] = ("read",) + metadata: Mapping[str, object] = field(default_factory=dict) + + +@dataclass(frozen=True, slots=True) +class DatasourceOriginReadRequest: + origin_ref: str + limit: int = 250 + offset: int = 0 + columns: tuple[str, ...] = () + expected_fingerprint: str | None = None + + +@dataclass(frozen=True, slots=True) +class DatasourceOriginReadResult: + origin: DatasourceOrigin + rows: tuple[Mapping[str, object], ...] + total_rows: int + truncated: bool + + +@runtime_checkable +class DatasourceCatalogueProvider(Protocol): + def list_datasources( + self, + session: object, + principal: object, + *, + query: str = "", + limit: int = 100, + ) -> Sequence[DatasourceDescriptor]: + ... + + def get_datasource( + self, + session: object, + principal: object, + *, + datasource_ref: str, + ) -> DatasourceDescriptor | None: + ... + + def read_datasource( + self, + session: object, + principal: object, + *, + request: DatasourceReadRequest, + ) -> DatasourceReadResult: + ... + + def list_materializations( + self, + session: object, + principal: object, + *, + datasource_ref: str, + ) -> Sequence[DatasourceMaterialization]: + ... + + +@runtime_checkable +class DatasourceLifecycleProvider(Protocol): + def list_stages( + self, + session: object, + principal: object, + *, + limit: int = 100, + ) -> Sequence[DatasourceStage]: + ... + + def create_stage( + self, + session: object, + principal: object, + *, + stage: DatasourceStageInput, + ) -> DatasourceStage: + ... + + def promote_stage( + self, + session: object, + principal: object, + *, + stage_ref: str, + freeze: bool = False, + frozen_label: str | None = None, + ) -> tuple[DatasourceDescriptor, DatasourceMaterialization]: + ... + + def register_origin( + self, + session: object, + principal: object, + *, + origin_ref: str, + name: str, + source_name: str, + mode: DatasourceMode, + description: str | None = None, + ) -> DatasourceDescriptor: + ... + + def refresh_datasource( + self, + session: object, + principal: object, + *, + datasource_ref: str, + ) -> tuple[DatasourceDescriptor, DatasourceMaterialization]: + ... + + def freeze_datasource( + self, + session: object, + principal: object, + *, + datasource_ref: str, + label: str | None = None, + ) -> DatasourceMaterialization: + ... + + def retire_datasource( + self, + session: object, + principal: object, + *, + datasource_ref: str, + ) -> DatasourceDescriptor: + ... + + +@runtime_checkable +class DatasourceOriginProvider(Protocol): + def list_origins( + self, + session: object, + principal: object, + *, + query: str = "", + limit: int = 100, + ) -> Sequence[DatasourceOrigin]: + ... + + def get_origin( + self, + session: object, + principal: object, + *, + origin_ref: str, + ) -> DatasourceOrigin | None: + ... + + def read_origin( + self, + session: object, + principal: object, + *, + request: DatasourceOriginReadRequest, + ) -> DatasourceOriginReadResult: + ... + + +def datasource_catalogue(registry: object | None) -> DatasourceCatalogueProvider | None: + capability = _capability(registry, CAPABILITY_DATASOURCE_CATALOGUE) + return capability if isinstance(capability, DatasourceCatalogueProvider) else None + + +def datasource_lifecycle(registry: object | None) -> DatasourceLifecycleProvider | None: + capability = _capability(registry, CAPABILITY_DATASOURCE_LIFECYCLE) + return capability if isinstance(capability, DatasourceLifecycleProvider) else None + + +def datasource_origins(registry: object | None) -> DatasourceOriginProvider | None: + capability = _capability(registry, CAPABILITY_DATASOURCE_ORIGINS) + return capability if isinstance(capability, DatasourceOriginProvider) else None + + +def _capability(registry: object | None, name: str) -> object | None: + if ( + registry is None + or not hasattr(registry, "has_capability") + or not hasattr(registry, "capability") + or not registry.has_capability(name) + ): + return None + return registry.capability(name) + + +__all__ = [ + "CAPABILITY_DATASOURCE_CATALOGUE", + "CAPABILITY_DATASOURCE_LIFECYCLE", + "CAPABILITY_DATASOURCE_ORIGINS", + "DatasourceAccessError", + "DatasourceCatalogueProvider", + "DatasourceConsistency", + "DatasourceDescriptor", + "DatasourceError", + "DatasourceField", + "DatasourceKind", + "DatasourceLifecycleProvider", + "DatasourceMaterialization", + "DatasourceMode", + "DatasourceNotFoundError", + "DatasourceOrigin", + "DatasourceOriginProvider", + "DatasourceOriginReadRequest", + "DatasourceOriginReadResult", + "DatasourceReadRequest", + "DatasourceReadResult", + "DatasourceShape", + "DatasourceStage", + "DatasourceStageInput", + "DatasourceUnavailableError", + "DatasourceValidationError", + "datasource_catalogue", + "datasource_lifecycle", + "datasource_origins", +] diff --git a/src/govoplan_core/core/definition_graphs.py b/src/govoplan_core/core/definition_graphs.py new file mode 100644 index 0000000..3344d23 --- /dev/null +++ b/src/govoplan_core/core/definition_graphs.py @@ -0,0 +1,342 @@ +from __future__ import annotations + +from collections import deque +from collections.abc import Mapping, Sequence +from dataclasses import dataclass, field +from typing import Any + + +@dataclass(frozen=True, slots=True) +class DefinitionPort: + id: str + label: str + required: bool = True + multiple: bool = False + minimum_connections: int = 1 + + +@dataclass(frozen=True, slots=True) +class DefinitionConfigField: + id: str + label: str + kind: str + required: bool = False + description: str | None = None + options: tuple[tuple[str, str], ...] = () + + +@dataclass(frozen=True, slots=True) +class DefinitionNodeType: + type: str + category: str + label: str + description: str + icon: str + input_ports: tuple[DefinitionPort, ...] = () + output_ports: tuple[DefinitionPort, ...] = ( + DefinitionPort(id="output", label="Output"), + ) + config_fields: tuple[DefinitionConfigField, ...] = () + default_config: Mapping[str, Any] = field(default_factory=dict) + metadata: Mapping[str, Any] = field(default_factory=dict) + + +@dataclass(frozen=True, slots=True) +class DefinitionNode: + id: str + type: str + + +@dataclass(frozen=True, slots=True) +class DefinitionEdge: + id: str + source: str + target: str + source_port: str = "output" + target_port: str = "input" + + +@dataclass(frozen=True, slots=True) +class DefinitionNodeCountConstraint: + code: str + label: str + minimum: int = 0 + maximum: int | None = None + node_types: tuple[str, ...] = () + type_prefixes: tuple[str, ...] = () + categories: tuple[str, ...] = () + + def matches(self, node: DefinitionNode, node_type: DefinitionNodeType | None) -> bool: + return ( + node.type in self.node_types + or any(node.type.startswith(prefix) for prefix in self.type_prefixes) + or node_type is not None + and node_type.category in self.categories + ) + + +@dataclass(frozen=True, slots=True) +class DefinitionGraphConstraints: + max_nodes: int = 100 + max_edges: int = 200 + allow_cycles: bool = False + require_connected: bool = True + node_counts: tuple[DefinitionNodeCountConstraint, ...] = () + + +@dataclass(frozen=True, slots=True) +class DefinitionGraphLibrary: + id: str + version: str + category_labels: Mapping[str, str] + node_types: tuple[DefinitionNodeType, ...] + constraints: DefinitionGraphConstraints = field(default_factory=DefinitionGraphConstraints) + + def node_type(self, type_id: str) -> DefinitionNodeType | None: + return next((item for item in self.node_types if item.type == type_id), None) + + +@dataclass(frozen=True, slots=True) +class DefinitionDiagnostic: + severity: str + code: str + message: str + node_id: str | None = None + field: str | None = None + + +def validate_definition_graph( + library: DefinitionGraphLibrary, + *, + nodes: Sequence[DefinitionNode], + edges: Sequence[DefinitionEdge], +) -> tuple[DefinitionDiagnostic, ...]: + diagnostics: list[DefinitionDiagnostic] = [] + constraints = library.constraints + if not nodes: + return (_error("graph.empty", "Add nodes before saving the definition."),) + if len(nodes) > constraints.max_nodes: + diagnostics.append( + _error( + "graph.node_limit", + f"Definitions are limited to {constraints.max_nodes:,} nodes.", + ) + ) + if len(edges) > constraints.max_edges: + diagnostics.append( + _error( + "graph.edge_limit", + f"Definitions are limited to {constraints.max_edges:,} edges.", + ) + ) + + node_by_id = {node.id: node for node in nodes} + if len(node_by_id) != len(nodes): + diagnostics.append(_error("graph.duplicate_node", "Node identifiers must be unique.")) + if len({edge.id for edge in edges}) != len(edges): + diagnostics.append(_error("graph.duplicate_edge", "Edge identifiers must be unique.")) + + incoming: dict[str, list[DefinitionEdge]] = {node_id: [] for node_id in node_by_id} + undirected: dict[str, set[str]] = {node_id: set() for node_id in node_by_id} + topology_edges: list[DefinitionEdge] = [] + for edge in edges: + if edge.source not in node_by_id: + diagnostics.append( + _error( + "edge.unknown_source", + f"Edge {edge.id!r} references an unknown source node.", + ) + ) + continue + if edge.target not in node_by_id: + diagnostics.append( + _error( + "edge.unknown_target", + f"Edge {edge.id!r} references an unknown target node.", + ) + ) + continue + if edge.source == edge.target: + diagnostics.append( + _error( + "edge.self_reference", + "A node cannot connect to itself.", + node_id=edge.source, + ) + ) + continue + topology_edges.append(edge) + source_definition = library.node_type(node_by_id[edge.source].type) + target_definition = library.node_type(node_by_id[edge.target].type) + if source_definition is not None and edge.source_port not in { + port.id for port in source_definition.output_ports + }: + diagnostics.append( + _error( + "edge.unknown_source_port", + f"Node {edge.source!r} has no output port {edge.source_port!r}.", + node_id=edge.source, + ) + ) + continue + if target_definition is not None and edge.target_port not in { + port.id for port in target_definition.input_ports + }: + diagnostics.append( + _error( + "edge.unknown_target_port", + f"Node {edge.target!r} has no input port {edge.target_port!r}.", + node_id=edge.target, + ) + ) + continue + incoming[edge.target].append(edge) + undirected[edge.source].add(edge.target) + undirected[edge.target].add(edge.source) + + for node in nodes: + definition = library.node_type(node.type) + if definition is None: + diagnostics.append( + _error( + "node.unsupported_type", + f"Node type {node.type!r} is not in the {library.id!r} library.", + node_id=node.id, + field="type", + ) + ) + continue + for port in definition.input_ports: + connections = [ + edge for edge in incoming.get(node.id, ()) if edge.target_port == port.id + ] + minimum = port.minimum_connections if port.required else 0 + if len(connections) < minimum: + diagnostics.append( + _error( + "node.input_required", + ( + f"{definition.label} requires {minimum} " + f"{port.label.lower()} connection" + f"{'' if minimum == 1 else 's'}." + ), + node_id=node.id, + ) + ) + if not port.multiple and len(connections) > 1: + diagnostics.append( + _error( + "node.input_multiple", + f"{port.label} accepts only one connection.", + node_id=node.id, + ) + ) + + for count_constraint in constraints.node_counts: + count = sum( + count_constraint.matches(node, library.node_type(node.type)) + for node in nodes + ) + if count < count_constraint.minimum: + diagnostics.append( + _error( + count_constraint.code, + ( + f"A definition needs at least {count_constraint.minimum} " + f"{count_constraint.label} node" + f"{'' if count_constraint.minimum == 1 else 's'}." + ), + ) + ) + if count_constraint.maximum is not None and count > count_constraint.maximum: + diagnostics.append( + _error( + count_constraint.code, + ( + f"A definition allows at most {count_constraint.maximum} " + f"{count_constraint.label} node" + f"{'' if count_constraint.maximum == 1 else 's'}." + ), + ) + ) + + _, cyclic = definition_topological_order(nodes, topology_edges) + if cyclic and not constraints.allow_cycles: + diagnostics.append(_error("graph.cycle", "Definition edges must form an acyclic graph.")) + if constraints.require_connected and len(node_by_id) > 1: + connected = _connected_nodes(next(iter(node_by_id)), undirected) + if len(connected) != len(node_by_id): + diagnostics.append( + _error("graph.disconnected", "Every node must belong to one connected definition.") + ) + return tuple(diagnostics) + + +def definition_topological_order( + nodes: Sequence[DefinitionNode], + edges: Sequence[DefinitionEdge], +) -> tuple[tuple[str, ...], bool]: + node_ids = [node.id for node in nodes] + incoming_count = {node_id: 0 for node_id in node_ids} + outgoing: dict[str, list[str]] = {node_id: [] for node_id in node_ids} + for edge in edges: + if ( + edge.source in outgoing + and edge.target in incoming_count + and edge.source != edge.target + ): + outgoing[edge.source].append(edge.target) + incoming_count[edge.target] += 1 + ready = deque(node_id for node_id in node_ids if incoming_count[node_id] == 0) + ordered: list[str] = [] + while ready: + node_id = ready.popleft() + ordered.append(node_id) + for target in outgoing[node_id]: + incoming_count[target] -= 1 + if incoming_count[target] == 0: + ready.append(target) + return tuple(ordered), len(ordered) != len(node_ids) + + +def _connected_nodes(start: str, adjacency: Mapping[str, set[str]]) -> set[str]: + seen: set[str] = set() + pending = [start] + while pending: + node_id = pending.pop() + if node_id in seen: + continue + seen.add(node_id) + pending.extend(adjacency.get(node_id, ())) + return seen + + +def _error( + code: str, + message: str, + *, + node_id: str | None = None, + field: str | None = None, +) -> DefinitionDiagnostic: + return DefinitionDiagnostic( + severity="error", + code=code, + message=message, + node_id=node_id, + field=field, + ) + + +__all__ = [ + "DefinitionConfigField", + "DefinitionDiagnostic", + "DefinitionEdge", + "DefinitionGraphConstraints", + "DefinitionGraphLibrary", + "DefinitionNode", + "DefinitionNodeCountConstraint", + "DefinitionNodeType", + "DefinitionPort", + "definition_topological_order", + "validate_definition_graph", +] diff --git a/src/govoplan_core/settings.py b/src/govoplan_core/settings.py index da65178..47095ce 100644 --- a/src/govoplan_core/settings.py +++ b/src/govoplan_core/settings.py @@ -17,7 +17,7 @@ class Settings(BaseSettings): tenant_data_mode: str = Field(default="shared", alias="TENANT_DATA_MODE") tenant_db_url_template: str | None = Field(default=None, alias="TENANT_DB_URL_TEMPLATE") tenant_schema_template: str | None = Field(default=None, alias="TENANT_SCHEMA_TEMPLATE") - enabled_modules: str = Field(default="tenancy,organizations,identity,access,admin,dashboard,policy,audit,campaigns,files,mail,calendar,poll,scheduling,connectors,dataflow,notifications,docs,ops", alias="ENABLED_MODULES") + enabled_modules: str = Field(default="tenancy,organizations,identity,access,admin,dashboard,policy,audit,campaigns,files,mail,calendar,poll,scheduling,connectors,datasources,dataflow,workflow,notifications,docs,ops", alias="ENABLED_MODULES") migration_track: str = Field(default="release", alias="GOVOPLAN_MIGRATION_TRACK") redis_url: str = Field(default="redis://redis:6379/0", alias="REDIS_URL") celery_enabled: bool = Field(default=False, alias="CELERY_ENABLED") diff --git a/tests/test_datasource_contract.py b/tests/test_datasource_contract.py new file mode 100644 index 0000000..c0bb7a7 --- /dev/null +++ b/tests/test_datasource_contract.py @@ -0,0 +1,164 @@ +from __future__ import annotations + +import unittest + +from govoplan_core.core.datasources import ( + CAPABILITY_DATASOURCE_CATALOGUE, + CAPABILITY_DATASOURCE_LIFECYCLE, + CAPABILITY_DATASOURCE_ORIGINS, + DatasourceCatalogueProvider, + DatasourceDescriptor, + DatasourceField, + DatasourceLifecycleProvider, + DatasourceMaterialization, + DatasourceOrigin, + DatasourceOriginProvider, + DatasourceOriginReadResult, + DatasourceReadResult, + DatasourceStage, + datasource_catalogue, + datasource_lifecycle, + datasource_origins, +) +from govoplan_core.core.modules import ModuleContext, ModuleManifest +from govoplan_core.core.registry import PlatformRegistry + + +class _Provider: + descriptor = DatasourceDescriptor( + ref="datasource:source-1", + source_name="monthly_cases", + name="Monthly cases", + kind="upload", + mode="static", + shape="tabular", + schema=(DatasourceField(name="case_id", data_type="string", nullable=False),), + fingerprint="abc123", + ) + + def list_datasources(self, session, principal, *, query="", limit=100): + del session, principal, query + return (self.descriptor,)[:limit] + + def get_datasource(self, session, principal, *, datasource_ref): + del session, principal + return self.descriptor if datasource_ref == self.descriptor.ref else None + + def read_datasource(self, session, principal, *, request): + del session, principal, request + return DatasourceReadResult( + datasource=self.descriptor, + rows=({"case_id": "0012"},), + total_rows=1, + truncated=False, + ) + + def list_materializations(self, session, principal, *, datasource_ref): + del session, principal, datasource_ref + return () + + def list_stages(self, session, principal, *, limit=100): + del session, principal, limit + return () + + def create_stage(self, session, principal, *, stage): + del session, principal, stage + return DatasourceStage( + ref="stage:1", + name="Stage", + source_name="stage", + kind="upload", + mode="static", + shape="tabular", + state="ready", + ) + + def promote_stage(self, session, principal, *, stage_ref, freeze=False, frozen_label=None): + del session, principal, stage_ref, freeze, frozen_label + return self.descriptor, DatasourceMaterialization( + ref="materialization:1", + datasource_ref=self.descriptor.ref, + revision=1, + state="published", + fingerprint=self.descriptor.fingerprint, + ) + + def register_origin(self, session, principal, **kwargs): + del session, principal, kwargs + return self.descriptor + + def refresh_datasource(self, session, principal, *, datasource_ref): + del session, principal, datasource_ref + return self.promote_stage(object(), object(), stage_ref="stage:1") + + def freeze_datasource(self, session, principal, *, datasource_ref, label=None): + del session, principal, datasource_ref, label + return self.promote_stage(object(), object(), stage_ref="stage:1")[1] + + def retire_datasource(self, session, principal, *, datasource_ref): + del session, principal, datasource_ref + return self.descriptor + + def list_origins(self, session, principal, *, query="", limit=100): + del session, principal, query + return ( + DatasourceOrigin( + ref="origin:1", + source_name="origin", + name="Origin", + kind="database", + shape="tabular", + supported_modes=("live", "cached"), + provider="test", + ), + )[:limit] + + def get_origin(self, session, principal, *, origin_ref): + return self.list_origins(session, principal)[0] if origin_ref == "origin:1" else None + + def read_origin(self, session, principal, *, request): + origin = self.get_origin(session, principal, origin_ref=request.origin_ref) + return DatasourceOriginReadResult( + origin=origin, + rows=({"id": 1},), + total_rows=1, + truncated=False, + ) + + +class DatasourceContractTests(unittest.TestCase): + def test_capabilities_are_runtime_checkable_and_resolved_without_modules(self) -> None: + provider = _Provider() + self.assertIsInstance(provider, DatasourceCatalogueProvider) + self.assertIsInstance(provider, DatasourceLifecycleProvider) + self.assertIsInstance(provider, DatasourceOriginProvider) + registry = PlatformRegistry() + registry.register( + ModuleManifest( + id="datasource_contract_test", + name="Datasource contract test", + version="test", + capability_factories={ + CAPABILITY_DATASOURCE_CATALOGUE: lambda context: provider, + CAPABILITY_DATASOURCE_LIFECYCLE: lambda context: provider, + CAPABILITY_DATASOURCE_ORIGINS: lambda context: provider, + }, + ) + ) + registry.configure_capability_context(ModuleContext(registry=registry, settings=object())) + + self.assertIs(provider, datasource_catalogue(registry)) + self.assertIs(provider, datasource_lifecycle(registry)) + self.assertIs(provider, datasource_origins(registry)) + self.assertIsNone(datasource_catalogue(PlatformRegistry())) + + def test_descriptor_distinguishes_mode_kind_shape_and_materialization(self) -> None: + descriptor = _Provider.descriptor + + self.assertEqual("static", descriptor.mode) + self.assertEqual("upload", descriptor.kind) + self.assertEqual("tabular", descriptor.shape) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_definition_graphs.py b/tests/test_definition_graphs.py new file mode 100644 index 0000000..144f2b4 --- /dev/null +++ b/tests/test_definition_graphs.py @@ -0,0 +1,145 @@ +from __future__ import annotations + +import unittest + +from govoplan_core.core.definition_graphs import ( + DefinitionEdge, + DefinitionGraphConstraints, + DefinitionGraphLibrary, + DefinitionNode, + DefinitionNodeCountConstraint, + DefinitionNodeType, + DefinitionPort, + validate_definition_graph, +) + + +def library(*, allow_cycles: bool) -> DefinitionGraphLibrary: + return DefinitionGraphLibrary( + id="test", + version="0.1.0", + category_labels={"control": "Control"}, + node_types=( + DefinitionNodeType( + type="start", + category="control", + label="Start", + description="Start", + icon="play", + ), + DefinitionNodeType( + type="step", + category="control", + label="Step", + description="Step", + icon="square", + input_ports=(DefinitionPort(id="input", label="Input"),), + ), + DefinitionNodeType( + type="end", + category="control", + label="End", + description="End", + icon="circle-stop", + input_ports=(DefinitionPort(id="input", label="Input"),), + output_ports=(), + ), + ), + constraints=DefinitionGraphConstraints( + allow_cycles=allow_cycles, + node_counts=( + DefinitionNodeCountConstraint( + code="graph.start_count", + label="start", + minimum=1, + maximum=1, + node_types=("start",), + ), + DefinitionNodeCountConstraint( + code="graph.end_count", + label="end", + minimum=1, + node_types=("end",), + ), + ), + ), + ) + + +class DefinitionGraphTests(unittest.TestCase): + def test_library_driven_ports_and_counts_validate_a_definition(self) -> None: + diagnostics = validate_definition_graph( + library(allow_cycles=False), + nodes=( + DefinitionNode(id="start", type="start"), + DefinitionNode(id="step", type="step"), + DefinitionNode(id="end", type="end"), + ), + edges=( + DefinitionEdge(id="edge-1", source="start", target="step"), + DefinitionEdge(id="edge-2", source="step", target="end"), + ), + ) + + self.assertEqual((), diagnostics) + + def test_constraints_change_cycle_semantics_without_changing_graph_shape(self) -> None: + nodes = ( + DefinitionNode(id="start", type="start"), + DefinitionNode(id="step", type="step"), + DefinitionNode(id="retry", type="step"), + DefinitionNode(id="end", type="end"), + ) + edges = ( + DefinitionEdge(id="edge-1", source="start", target="step"), + DefinitionEdge(id="edge-2", source="step", target="retry"), + DefinitionEdge(id="edge-3", source="retry", target="step"), + DefinitionEdge(id="edge-4", source="retry", target="end"), + ) + + dag_codes = { + item.code + for item in validate_definition_graph( + library(allow_cycles=False), + nodes=nodes, + edges=edges, + ) + } + workflow_codes = { + item.code + for item in validate_definition_graph( + library(allow_cycles=True), + nodes=nodes, + edges=edges, + ) + } + + self.assertIn("graph.cycle", dag_codes) + self.assertNotIn("graph.cycle", workflow_codes) + + def test_unknown_ports_and_missing_required_inputs_are_explicit(self) -> None: + diagnostics = validate_definition_graph( + library(allow_cycles=False), + nodes=( + DefinitionNode(id="start", type="start"), + DefinitionNode(id="end", type="end"), + ), + edges=( + DefinitionEdge( + id="edge", + source="start", + target="end", + target_port="missing", + ), + ), + ) + + self.assertTrue( + {"edge.unknown_target_port", "node.input_required"}.issubset( + {item.code for item in diagnostics} + ) + ) + + +if __name__ == "__main__": + unittest.main() diff --git a/webui/package-lock.json b/webui/package-lock.json index 63beaf5..210d719 100644 --- a/webui/package-lock.json +++ b/webui/package-lock.json @@ -16,6 +16,7 @@ "@govoplan/campaign-webui": "file:../../govoplan-campaign/webui", "@govoplan/dashboard-webui": "file:../../govoplan-dashboard/webui", "@govoplan/dataflow-webui": "file:../../govoplan-dataflow/webui", + "@govoplan/datasources-webui": "file:../../govoplan-datasources/webui", "@govoplan/docs-webui": "file:../../govoplan-docs/webui", "@govoplan/files-webui": "file:../../govoplan-files/webui", "@govoplan/idm-webui": "file:../../govoplan-idm/webui", @@ -191,6 +192,23 @@ } } }, + "../../govoplan-datasources/webui": { + "name": "@govoplan/datasources-webui", + "version": "0.1.14", + "peerDependencies": { + "@govoplan/core-webui": "^0.1.14", + "lucide-react": "^1.23.0", + "react": "^19.0.0", + "react-dom": "^19.0.0", + "react-router-dom": "^7.1.1", + "typescript": "^5.7.2" + }, + "peerDependenciesMeta": { + "@govoplan/core-webui": { + "optional": true + } + } + }, "../../govoplan-docs/webui": { "name": "@govoplan/docs-webui", "version": "0.1.10", @@ -1115,6 +1133,10 @@ "resolved": "../../govoplan-dataflow/webui", "link": true }, + "node_modules/@govoplan/datasources-webui": { + "resolved": "../../govoplan-datasources/webui", + "link": true + }, "node_modules/@govoplan/docs-webui": { "resolved": "../../govoplan-docs/webui", "link": true diff --git a/webui/package.json b/webui/package.json index a48a07f..50553b4 100644 --- a/webui/package.json +++ b/webui/package.json @@ -14,6 +14,10 @@ "./app": { "types": "./src/app.ts", "import": "./src/app.ts" + }, + "./definition-graph": { + "types": "./src/definitionGraph.ts", + "import": "./src/definitionGraph.ts" } }, "scripts": { @@ -26,7 +30,7 @@ "test:dialog-focus": "rm -rf .component-test-build && mkdir -p .component-test-build && printf '{\"type\":\"commonjs\"}\\n' > .component-test-build/package.json && tsc -p tsconfig.component-tests.json && node .component-test-build/tests/dialog-focus.test.js && node scripts/test-dialog-focus-structure.mjs", "test:explorer-tree": "rm -rf .component-test-build && mkdir -p .component-test-build && printf '{\"type\":\"commonjs\"}\\n' > .component-test-build/package.json && tsc -p tsconfig.component-tests.json && node .component-test-build/tests/explorer-tree.test.js", "test:icon-button": "rm -rf .component-test-build && mkdir -p .component-test-build && printf '{\"type\":\"commonjs\"}\\n' > .component-test-build/package.json && tsc -p tsconfig.component-tests.json && node .component-test-build/tests/icon-button.test.js", - "test:module-capabilities": "rm -rf .module-test-build && mkdir -p .module-test-build && printf '{\"type\":\"commonjs\"}\n' > .module-test-build/package.json && tsc -p tsconfig.module-tests.json && node .module-test-build/tests/module-capabilities.test.js && node .module-test-build/tests/privacy-policy.test.js && node .module-test-build/tests/help-context.test.js", + "test:module-capabilities": "rm -rf .module-test-build && mkdir -p .module-test-build && printf '{\"type\":\"commonjs\"}\n' > .module-test-build/package.json && tsc -p tsconfig.module-tests.json && node .module-test-build/tests/module-capabilities.test.js && node .module-test-build/tests/privacy-policy.test.js && node .module-test-build/tests/help-context.test.js && node .module-test-build/tests/definition-graph.test.js", "test:module-permutations": "node scripts/test-module-permutations.mjs", "test:mail-components": "rm -rf .component-test-build && mkdir -p .component-test-build && printf '{\"type\":\"commonjs\"}\\n' > .component-test-build/package.json && tsc -p tsconfig.component-tests.json && node .component-test-build/tests/mail-components.test.js", "test:metric-card": "rm -rf .component-test-build && mkdir -p .component-test-build && printf '{\"type\":\"commonjs\"}\\n' > .component-test-build/package.json && tsc -p tsconfig.component-tests.json && node .component-test-build/tests/metric-card.test.js", @@ -42,6 +46,7 @@ "@govoplan/calendar-webui": "file:../../govoplan-calendar/webui", "@govoplan/campaign-webui": "file:../../govoplan-campaign/webui", "@govoplan/dataflow-webui": "file:../../govoplan-dataflow/webui", + "@govoplan/datasources-webui": "file:../../govoplan-datasources/webui", "@govoplan/dashboard-webui": "file:../../govoplan-dashboard/webui", "@govoplan/docs-webui": "file:../../govoplan-docs/webui", "@govoplan/files-webui": "file:../../govoplan-files/webui", diff --git a/webui/scripts/test-module-permutations.mjs b/webui/scripts/test-module-permutations.mjs index 5fa4f75..d1c6b53 100644 --- a/webui/scripts/test-module-permutations.mjs +++ b/webui/scripts/test-module-permutations.mjs @@ -9,6 +9,7 @@ const packageByModule = { campaigns: "@govoplan/campaign-webui", dashboard: "@govoplan/dashboard-webui", dataflow: "@govoplan/dataflow-webui", + datasources: "@govoplan/datasources-webui", docs: "@govoplan/docs-webui", files: "@govoplan/files-webui", idm: "@govoplan/idm-webui", @@ -29,6 +30,8 @@ const cases = [ { name: "admin-with-policy-and-audit", modules: ["access", "admin", "policy", "audit"] }, { name: "dashboard-only", modules: ["dashboard"] }, { name: "dataflow-only", modules: ["dataflow"] }, + { name: "datasources-only", modules: ["datasources"] }, + { name: "dataflow-with-datasources", modules: ["datasources", "dataflow"] }, { name: "calendar-only", modules: ["calendar"] }, { name: "files-only", modules: ["files"] }, { name: "mail-only", modules: ["mail"] }, @@ -41,7 +44,7 @@ const cases = [ { name: "scheduling-only", modules: ["scheduling"] }, { name: "scheduling-with-calendar", modules: ["scheduling", "calendar"] }, { name: "docs-and-ops", modules: ["access", "docs", "ops"] }, - { name: "full-product", modules: ["access", "admin", "addresses", "policy", "audit", "dashboard", "dataflow", "organizations", "idm", "campaigns", "files", "mail", "notifications", "docs", "ops", "calendar", "scheduling"] } + { name: "full-product", modules: ["access", "admin", "addresses", "policy", "audit", "dashboard", "datasources", "dataflow", "organizations", "idm", "campaigns", "files", "mail", "notifications", "docs", "ops", "calendar", "scheduling"] } ]; const npmExec = process.env.npm_execpath; diff --git a/webui/src/definitionGraph.ts b/webui/src/definitionGraph.ts new file mode 100644 index 0000000..54a0313 --- /dev/null +++ b/webui/src/definitionGraph.ts @@ -0,0 +1,147 @@ +export type DefinitionGraphPosition = { + x: number; + y: number; +}; + +export type DefinitionGraphPort = { + id: string; + label: string; + required: boolean; + multiple: boolean; + minimum_connections: number; +}; + +export type DefinitionGraphConfigField = { + id: string; + label: string; + kind: string; + required: boolean; + description?: string | null; + options: [string, string][]; +}; + +export type DefinitionGraphNodeType = { + type: string; + category: string; + category_label: string; + label: string; + description: string; + icon: string; + input_ports: DefinitionGraphPort[]; + output_ports: DefinitionGraphPort[]; + config_fields: DefinitionGraphConfigField[]; + default_config: Record; + metadata?: Record; +}; + +export type DefinitionGraphNode = { + id: string; + type: string; + label: string; + position: DefinitionGraphPosition; + config: Record; +}; + +export type DefinitionGraphEdge = { + id: string; + source: string; + target: string; + source_port?: string; + target_port?: string; +}; + +export type DefinitionGraph = { + schema_version: number; + nodes: DefinitionGraphNode[]; + edges: DefinitionGraphEdge[]; +}; + +export type DefinitionGraphConnection = { + source: string; + target: string; + sourcePort?: string | null; + targetPort?: string | null; +}; + +export function createDefinitionGraphNode( + type: string, + position: DefinitionGraphPosition, + library: DefinitionGraphNodeType[] +): TNode { + const definition = library.find((item) => item.type === type); + return { + id: `${type.replace(/\./g, "-")}-${crypto.randomUUID()}`, + type, + label: definition?.label ?? type, + position, + config: structuredClone(definition?.default_config ?? {}) + } as TNode; +} + +export function definitionConnectionError( + graph: DefinitionGraph, + library: DefinitionGraphNodeType[], + connection: DefinitionGraphConnection, + options: { allowCycles?: boolean } = {} +): string | null { + if (!connection.source || !connection.target) return "Both endpoints are required."; + if (connection.source === connection.target) return "A node cannot connect to itself."; + const source = graph.nodes.find((node) => node.id === connection.source); + const target = graph.nodes.find((node) => node.id === connection.target); + if (!source || !target) return "The connection references an unknown node."; + const sourceDefinition = library.find((item) => item.type === source.type); + const targetDefinition = library.find((item) => item.type === target.type); + if (!sourceDefinition || !targetDefinition) return "The node type is not in this library."; + const sourcePort = connection.sourcePort ?? "output"; + const targetPort = connection.targetPort ?? "input"; + if (!sourceDefinition.output_ports.some((port) => port.id === sourcePort)) { + return "The source port is not available."; + } + const port = targetDefinition.input_ports.find((item) => item.id === targetPort); + if (!port) return "The target port is not available."; + if (graph.edges.some( + (edge) => + edge.source === connection.source + && edge.target === connection.target + && (edge.source_port ?? "output") === sourcePort + && (edge.target_port ?? "input") === targetPort + )) { + return "This connection already exists."; + } + if (!port.multiple && graph.edges.some( + (edge) => + edge.target === connection.target + && (edge.target_port ?? "input") === targetPort + )) { + return "The target port accepts only one connection."; + } + if (!options.allowCycles && wouldCreateDefinitionCycle( + graph, + connection.source, + connection.target + )) { + return "This connection would create a cycle."; + } + return null; +} + +export function wouldCreateDefinitionCycle( + graph: DefinitionGraph, + source: string, + target: string +): boolean { + const outgoing = new Map(); + graph.edges.forEach((edge) => { + outgoing.set(edge.source, [...(outgoing.get(edge.source) ?? []), edge.target]); + }); + const pending = [target]; + const visited = new Set(); + while (pending.length) { + const nodeId = pending.pop(); + if (!nodeId || visited.has(nodeId)) continue; + if (nodeId === source) return true; + visited.add(nodeId); + pending.push(...(outgoing.get(nodeId) ?? [])); + } + return false; +} diff --git a/webui/src/platform/modules.ts b/webui/src/platform/modules.ts index 25a9c53..b088fcb 100644 --- a/webui/src/platform/modules.ts +++ b/webui/src/platform/modules.ts @@ -1,4 +1,4 @@ -import { Activity, Bell, BookUser, Building2, CalendarClock, CalendarDays, ClipboardPenLine, Folder, Form, LayoutDashboard, LayoutTemplate, Mail, Mails, RadioTower, Shield, Users, Waypoints, type LucideIcon } from "lucide-react"; +import { Activity, Bell, BookUser, Building2, CalendarClock, CalendarDays, ClipboardPenLine, DatabaseZap, Folder, Form, LayoutDashboard, LayoutTemplate, Mail, Mails, RadioTower, Shield, Users, Waypoints, type LucideIcon } from "lucide-react"; import installedWebModules from "virtual:govoplan-installed-modules"; import type { AuthInfo, DashboardWidgetContribution, DashboardWidgetsUiCapability, PlatformModuleInfo, PlatformNavItem, PlatformPublicModuleInfo, PlatformWebModule } from "../types"; import { @@ -49,6 +49,7 @@ const iconByName: Record = { campaign: Mails, "clipboard-pen-line": ClipboardPenLine, dashboard: LayoutDashboard, + "database-zap": DatabaseZap, file: Folder, files: Folder, folder: Folder, diff --git a/webui/tests/definition-graph.test.ts b/webui/tests/definition-graph.test.ts new file mode 100644 index 0000000..7e2c6b5 --- /dev/null +++ b/webui/tests/definition-graph.test.ts @@ -0,0 +1,68 @@ +import { + createDefinitionGraphNode, + definitionConnectionError, + type DefinitionGraph, + type DefinitionGraphNodeType +} from "../src/definitionGraph"; + +const library: DefinitionGraphNodeType[] = [ + { + type: "start", + category: "control", + category_label: "Control", + label: "Start", + description: "Start", + icon: "play", + input_ports: [], + output_ports: [ + { id: "output", label: "Output", required: true, multiple: false, minimum_connections: 1 } + ], + config_fields: [], + default_config: { mode: "manual" } + }, + { + type: "end", + category: "control", + category_label: "Control", + label: "End", + description: "End", + icon: "circle-stop", + input_ports: [ + { id: "input", label: "Input", required: true, multiple: false, minimum_connections: 1 } + ], + output_ports: [], + config_fields: [], + default_config: {} + } +]; + +const graph: DefinitionGraph = { + schema_version: 1, + nodes: [ + { id: "start", type: "start", label: "Start", position: { x: 0, y: 0 }, config: {} }, + { id: "end", type: "end", label: "End", position: { x: 100, y: 0 }, config: {} } + ], + edges: [] +}; + +if (definitionConnectionError( + graph, + library, + { source: "start", target: "end" } +) !== null) { + throw new Error("Expected a valid library-driven connection."); +} + +graph.edges.push({ id: "edge", source: "start", target: "end" }); +if (!definitionConnectionError( + graph, + library, + { source: "start", target: "end" } +)) { + throw new Error("Expected duplicate connection validation."); +} + +const created = createDefinitionGraphNode("start", { x: 40, y: 50 }, library); +if (created.config.mode !== "manual" || created.position.x !== 40) { + throw new Error("Expected node creation to clone library defaults."); +} diff --git a/webui/tsconfig.module-tests.json b/webui/tsconfig.module-tests.json index 5a76eb9..df81a54 100644 --- a/webui/tsconfig.module-tests.json +++ b/webui/tsconfig.module-tests.json @@ -18,9 +18,11 @@ "tests/module-capabilities.test.ts", "tests/privacy-policy.test.ts", "tests/help-context.test.ts", + "tests/definition-graph.test.ts", "src/platform/moduleLogic.ts", "src/utils/helpContext.ts", "src/features/privacy/policyLogic.ts", + "src/definitionGraph.ts", "src/types.ts" ] } diff --git a/webui/vite.config.ts b/webui/vite.config.ts index ae74ea9..cc4e5ed 100644 --- a/webui/vite.config.ts +++ b/webui/vite.config.ts @@ -18,6 +18,7 @@ const defaultWebModulePackages = [ "@govoplan/calendar-webui", "@govoplan/campaign-webui", "@govoplan/dataflow-webui", + "@govoplan/datasources-webui", "@govoplan/dashboard-webui", "@govoplan/docs-webui", "@govoplan/files-webui", @@ -89,6 +90,7 @@ export default defineConfig({ dedupe: ["react", "react-dom", "react-router-dom"], alias: [ { find: "@govoplan/core-webui/app", replacement: fileURLToPath(new URL("./src/app.ts", import.meta.url)) }, + { find: "@govoplan/core-webui/definition-graph", replacement: fileURLToPath(new URL("./src/definitionGraph.ts", import.meta.url)) }, { find: "@govoplan/core-webui", replacement: fileURLToPath(new URL("./src/index.ts", import.meta.url)) } ] }, @@ -102,6 +104,7 @@ export default defineConfig({ fileURLToPath(new URL('../../govoplan-audit/webui', import.meta.url)), fileURLToPath(new URL('../../govoplan-calendar/webui', import.meta.url)), fileURLToPath(new URL('../../govoplan-dataflow/webui', import.meta.url)), + fileURLToPath(new URL('../../govoplan-datasources/webui', import.meta.url)), fileURLToPath(new URL('../../govoplan-dashboard/webui', import.meta.url)), fileURLToPath(new URL('../../govoplan-docs/webui', import.meta.url)), fileURLToPath(new URL('../../govoplan-files/webui', import.meta.url)),