gateway/tests/unit/workflow/test_workflowFileSchema.py

# Copyright (c) 2026 Patrick Motsch
# All rights reserved.
"""Unit tests for the workflow-file (versioned envelope) schema."""

import pytest

from modules.features.graphicalEditor._workflowFileSchema import (
    WORKFLOW_FILE_KIND,
    WORKFLOW_FILE_SCHEMA_VERSION,
    WorkflowFileSchemaError,
    buildFileFromWorkflow,
    buildFileName,
    envelopeToWorkflowData,
    isWorkflowFileEnvelope,
    normalizeGraph,
    validateFileEnvelope,
)


def _sampleWorkflowRow() -> dict:
    return {
        "id": "wf-123",
        "mandateId": "mand-1",
        "featureInstanceId": "inst-1",
        "currentVersionId": "ver-1",
        "eventId": "evt-1",
        "active": True,
        "label": "Test Workflow",
        "description": "Round-trip sample",
        "tags": ["test"],
        "templateScope": None,
        "sharedReadOnly": False,
        "notifyOnFailure": True,
        "isTemplate": False,
        "graph": {
            "nodes": [
                {"id": "n1", "type": "trigger.manual", "x": 50, "y": 200, "parameters": {}},
                {"id": "n2", "type": "ai.prompt", "position": {"x": 300, "y": 200}, "parameters": {"aiPrompt": "Hi"}},
            ],
            "connections": [
                {"source": "n1", "target": "n2", "sourceOutput": 0, "targetInput": 0},
            ],
        },
        "invocations": [{"id": "inv-1", "kind": "manual", "enabled": True, "title": {"de": "Start"}}],
        "sysCreatedAt": 1700000000.0,
        "sysModifiedAt": 1700000100.0,
    }


def _sampleNodeTypes() -> list:
    return ["trigger.manual", "ai.prompt"]


class TestBuildFile:
    def test_envelopeContainsKindAndSchemaVersion(self):
        envelope = buildFileFromWorkflow(_sampleWorkflowRow())
        assert envelope["$kind"] == WORKFLOW_FILE_KIND
        assert envelope["$schemaVersion"] == WORKFLOW_FILE_SCHEMA_VERSION
        assert "$exportedAt" in envelope

    def test_persistenceFieldsAreStripped(self):
        envelope = buildFileFromWorkflow(_sampleWorkflowRow())
        for forbidden in ("id", "mandateId", "featureInstanceId", "currentVersionId", "eventId", "active", "sysCreatedAt", "sysModifiedAt"):
            assert forbidden not in envelope, f"{forbidden} must not appear in exported file"

    def test_portableFieldsAreCopied(self):
        envelope = buildFileFromWorkflow(_sampleWorkflowRow())
        assert envelope["label"] == "Test Workflow"
        assert envelope["description"] == "Round-trip sample"
        assert envelope["tags"] == ["test"]
        assert envelope["notifyOnFailure"] is True

    def test_graphPositionsAreNormalized(self):
        envelope = buildFileFromWorkflow(_sampleWorkflowRow())
        nodes = envelope["graph"]["nodes"]
        assert nodes[0]["x"] == 50
        assert nodes[0]["y"] == 200
        assert nodes[1]["x"] == 300
        assert nodes[1]["y"] == 200
        assert "position" not in nodes[1]


class TestValidate:
    def test_validEnvelopeReturnsNoErrors(self):
        envelope = buildFileFromWorkflow(_sampleWorkflowRow())
        normalized, warnings = validateFileEnvelope(envelope, knownNodeTypes=_sampleNodeTypes())
        assert normalized["label"] == "Test Workflow"
        assert warnings == []

    def test_missingSchemaVersionRaises(self):
        with pytest.raises(WorkflowFileSchemaError):
            validateFileEnvelope({"label": "x", "graph": {}})

    def test_unsupportedSchemaVersionRaises(self):
        with pytest.raises(WorkflowFileSchemaError):
            validateFileEnvelope({"$schemaVersion": "99.0", "label": "x", "graph": {}})

    def test_missingLabelRaises(self):
        with pytest.raises(WorkflowFileSchemaError):
            validateFileEnvelope({"$schemaVersion": WORKFLOW_FILE_SCHEMA_VERSION, "graph": {}})

    def test_missingGraphRaises(self):
        with pytest.raises(WorkflowFileSchemaError):
            validateFileEnvelope({"$schemaVersion": WORKFLOW_FILE_SCHEMA_VERSION, "label": "x"})

    def test_unknownNodeTypeRaises(self):
        envelope = buildFileFromWorkflow(_sampleWorkflowRow())
        with pytest.raises(WorkflowFileSchemaError):
            validateFileEnvelope(envelope, knownNodeTypes=["trigger.manual"])

    def test_emptyNodesProducesWarning(self):
        envelope = {
            "$schemaVersion": WORKFLOW_FILE_SCHEMA_VERSION,
            "label": "Empty",
            "graph": {"nodes": [], "connections": []},
        }
        _, warnings = validateFileEnvelope(envelope)
        assert any("no nodes" in w.lower() for w in warnings)

    def test_danglingConnectionProducesWarning(self):
        envelope = {
            "$schemaVersion": WORKFLOW_FILE_SCHEMA_VERSION,
            "label": "Bad",
            "graph": {
                "nodes": [{"id": "a", "type": "trigger.manual"}],
                "connections": [{"source": "a", "target": "ghost"}],
            },
        }
        _, warnings = validateFileEnvelope(envelope, knownNodeTypes=["trigger.manual"])
        assert any("ghost" in w for w in warnings)


class TestRoundTrip:
    def test_exportThenImportPreservesGraphStructure(self):
        original = _sampleWorkflowRow()
        envelope = buildFileFromWorkflow(original)
        normalized, _ = validateFileEnvelope(envelope, knownNodeTypes=_sampleNodeTypes())
        data = envelopeToWorkflowData(normalized, mandateId="mand-2", featureInstanceId="inst-2")

        assert data["mandateId"] == "mand-2"
        assert data["featureInstanceId"] == "inst-2"
        assert data["active"] is False, "imports must be inactive by default"
        assert data["label"] == original["label"]
        assert data["description"] == original["description"]
        assert len(data["graph"]["nodes"]) == len(original["graph"]["nodes"])
        assert len(data["graph"]["connections"]) == len(original["graph"]["connections"])
        for forbidden in ("id", "currentVersionId", "eventId"):
            assert forbidden not in data


class TestHelpers:
    def test_isWorkflowFileEnvelopeAcceptsValid(self):
        assert isWorkflowFileEnvelope(buildFileFromWorkflow(_sampleWorkflowRow())) is True

    def test_isWorkflowFileEnvelopeRejectsRandom(self):
        assert isWorkflowFileEnvelope({"foo": "bar"}) is False
        assert isWorkflowFileEnvelope("not-a-dict") is False
        assert isWorkflowFileEnvelope(None) is False

    def test_buildFileNameProducesSafeSlug(self):
        assert buildFileName("PWG: Pilot Workflow!") == "pwg-pilot-workflow.workflow.json"
        assert buildFileName("") == "workflow.workflow.json"

    def test_normalizeGraphHandlesMissingFields(self):
        assert normalizeGraph(None) == {"nodes": [], "connections": []}
        assert normalizeGraph({}) == {"nodes": [], "connections": []}