dograh/api/tests/test_trigger_path_validation.py

from api.services.workflow.trigger_paths import (
    TRIGGER_PATH_MAX_LENGTH,
    validate_trigger_paths,
)


def test_validate_trigger_paths_rejects_invalid_path_segments():
    workflow_definition = {
        "nodes": [
            {
                "id": "trigger-1",
                "type": "trigger",
                "data": {"trigger_path": "support/west"},
            }
        ],
        "edges": [],
    }

    issues = validate_trigger_paths(workflow_definition)

    assert len(issues) == 1
    assert issues[0].node_id == "trigger-1"
    assert "single URL path segment" in issues[0].message


def test_validate_trigger_paths_rejects_long_and_duplicate_paths():
    long_path = "a" * (TRIGGER_PATH_MAX_LENGTH + 1)
    workflow_definition = {
        "nodes": [
            {
                "id": "trigger-1",
                "type": "trigger",
                "data": {"trigger_path": long_path},
            },
            {
                "id": "trigger-2",
                "type": "trigger",
                "data": {"trigger_path": "sales_agent"},
            },
            {
                "id": "trigger-3",
                "type": "trigger",
                "data": {"trigger_path": "sales_agent"},
            },
        ],
        "edges": [],
    }

    issues = validate_trigger_paths(workflow_definition)
    messages = [issue.message for issue in issues]

    assert (
        f"Trigger path must be {TRIGGER_PATH_MAX_LENGTH} characters or fewer."
        in messages
    )
    assert "Trigger path is duplicated in this workflow." in messages


def test_validate_trigger_paths_detects_duplicate_when_first_node_has_no_id():
    """A duplicate trigger path must be flagged even when the first node sharing
    that path has no ``id`` (``node.get("id")`` is None).

    Regression: the duplicate check previously used ``seen_paths.get(path)`` and
    treated a ``None`` result as "not seen yet", so a first node with a missing
    id (stored as None) made every later node with the same path slip through
    undetected.
    """
    workflow_definition = {
        "nodes": [
            # No "id" key -> node_id resolves to None.
            {"type": "trigger", "data": {"trigger_path": "sales_agent"}},
            {
                "id": "trigger-2",
                "type": "trigger",
                "data": {"trigger_path": "sales_agent"},
            },
        ],
        "edges": [],
    }

    issues = validate_trigger_paths(workflow_definition)
    messages = [issue.message for issue in issues]

    assert "Trigger path is duplicated in this workflow." in messages
    duplicate_issue = next(
        issue
        for issue in issues
        if issue.message == "Trigger path is duplicated in this workflow."
    )
    assert duplicate_issue.node_id == "trigger-2"