DataOps-platform/tests/test_n8n_migration_inventory.py

from __future__ import annotations

import json
import subprocess
import sys
from pathlib import Path


class FakeN8nClient:
    def __init__(self):
        self.workflow_calls = []
        self.execution_calls = []

    def list_workflows(self, *, limit, cursor=None):
        self.workflow_calls.append(cursor)
        if cursor is None:
            return {
                "data": [
                    {
                        "id": "wf-1",
                        "name": "orders-daily",
                        "active": True,
                        "updatedAt": "2026-07-18T01:00:00Z",
                        "tags": [{"name": "production"}],
                        "nodes": [
                            {
                                "name": "Daily",
                                "type": "n8n-nodes-base.scheduleTrigger",
                                "parameters": {"rule": {"interval": [{"field": "days"}]}},
                            },
                            {
                                "name": "Orders DB",
                                "type": "n8n-nodes-base.postgres",
                                "parameters": {"operation": "executeQuery"},
                                "credentials": {
                                    "postgres": {
                                        "id": "credential-id-must-not-leak",
                                        "name": "orders-reader",
                                        "password": "secret-must-not-leak",
                                    }
                                },
                            },
                        ],
                        "connections": {},
                    }
                ],
                "nextCursor": "next-page",
            }
        return {
            "data": [
                {
                    "id": "wf-2",
                    "name": "community-node-flow",
                    "active": False,
                    "nodes": [
                        {
                            "name": "Unknown",
                            "type": "community.exampleNode",
                            "parameters": {"apiKey": "also-must-not-leak"},
                        }
                    ],
                    "connections": {},
                }
            ]
        }

    def list_executions(
        self,
        *,
        workflow_id,
        limit,
        cursor=None,
        include_data=False,
    ):
        self.execution_calls.append((workflow_id, cursor, include_data))
        if workflow_id == "wf-1":
            return {
                "data": [
                    {
                        "id": "execution-1",
                        "status": "success",
                        "startedAt": "2026-07-17T01:00:00Z",
                        "stoppedAt": "2026-07-17T01:02:00Z",
                        "data": {"password": "execution-secret-must-not-leak"},
                    }
                ]
            }
        return {"data": []}


ROOT = Path(__file__).resolve().parents[1]


def test_inventory_script_can_run_as_a_direct_cli_entrypoint():
    result = subprocess.run(
        [
            sys.executable,
            str(ROOT / "scripts" / "inventory_n8n_workflows.py"),
            "--help",
        ],
        cwd=ROOT,
        capture_output=True,
        text=True,
    )

    assert result.returncode == 0, result.stderr
    assert "credential-safe n8n migration inventory" in result.stdout


def test_inventory_cli_passes_explicit_timeout_without_flask_context(
    monkeypatch, tmp_path
):
    import scripts.inventory_n8n_workflows as inventory

    captured = {}

    class RecordingClient:
        def __init__(self, *, api_url, api_key, timeout):
            captured.update(
                api_url=api_url,
                api_key=api_key,
                timeout=timeout,
            )

    monkeypatch.setattr(inventory, "N8nClient", RecordingClient)
    monkeypatch.setattr(
        inventory,
        "collect_inventory",
        lambda client, bindings: inventory.empty_inventory(reason="test"),
    )
    monkeypatch.setenv("N8N_API_URL", "http://localhost:15678")
    monkeypatch.setenv("N8N_API_KEY", "local-test-key")
    monkeypatch.setenv("N8N_API_TIMEOUT", "17")
    monkeypatch.setattr(
        sys,
        "argv",
        [
            "inventory_n8n_workflows.py",
            "--output",
            str(tmp_path / "inventory.json"),
        ],
    )

    assert inventory.main() == 0
    assert captured == {
        "api_url": "http://localhost:15678",
        "api_key": "local-test-key",
        "timeout": 17,
    }


def test_collect_inventory_paginates_and_never_exports_secret_material():
    from scripts.inventory_n8n_workflows import collect_inventory

    result = collect_inventory(
        FakeN8nClient(),
        bindings={"wf-1": "019b06e4-20e1-7cc2-88b5-bc8e28db3285"},
        generated_at="2026-07-18T02:00:00Z",
    )

    assert result["schema_version"] == "1.0"
    assert result["collection_status"] == "complete"
    assert result["coverage"] == {
        "workflow_count": 2,
        "active_count": 1,
        "recently_executed_count": 1,
        "blocked_count": 1,
        "unbound_count": 1,
    }

    first = result["workflows"][0]
    assert first["workflow_id"] == "wf-1"
    assert first["dataflow_uid"] == "019b06e4-20e1-7cc2-88b5-bc8e28db3285"
    assert first["node_categories"] == ["sql", "trigger"]
    assert first["credential_references"] == [
        {
            "node_name": "Orders DB",
            "credential_type": "postgres",
            "credential_name": "orders-reader",
        }
    ]
    assert first["recent_execution"]["status_counts"] == {"success": 1}
    assert first["migration_assessment"]["production_write"] == "unknown"
    assert first["migration_assessment"]["idempotent"] == "unknown"

    second = result["workflows"][1]
    assert "unsupported_node:community.exampleNode" in second["migration_assessment"][
        "blockers"
    ]
    assert "missing_dataflow_binding" in second["migration_assessment"]["blockers"]

    serialized = json.dumps(result, ensure_ascii=False)
    for forbidden in (
        "credential-id-must-not-leak",
        "secret-must-not-leak",
        "also-must-not-leak",
        "execution-secret-must-not-leak",
    ):
        assert forbidden not in serialized


def test_empty_collection_is_explicitly_pending_instead_of_claiming_coverage():
    from scripts.inventory_n8n_workflows import empty_inventory

    result = empty_inventory(
        reason="N8N_API_KEY is not configured",
        generated_at="2026-07-18T02:00:00Z",
    )

    assert result["collection_status"] == "pending_live_n8n_access"
    assert result["coverage"]["workflow_count"] == 0
    assert result["collection_error"] == "N8N_API_KEY is not configured"