DataOps-platform/tests/test_n8n_client_config.py

from pathlib import Path

import pytest

from app.core.data_factory.n8n_client import N8nClient, N8nClientError


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


def test_n8n_client_contains_no_committed_production_credentials():
    source = (ROOT / "app/core/data_factory/n8n_client.py").read_text(encoding="utf-8")
    config_source = (ROOT / "app/config/config.py").read_text(encoding="utf-8")

    combined = source + config_source
    assert "n8n.citupro.com" not in combined
    assert "DEFAULT_N8N_API_KEY" not in combined
    assert "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9" not in combined


@pytest.mark.parametrize(
    ("api_url", "api_key", "message"),
    [
        ("", "local-key", "N8N_API_URL"),
        ("http://n8n:5678", "", "N8N_API_KEY"),
        ("https://n8n.example.com", "local-key", "N8N_API_URL"),
        ("http://n8n:5678", "replace-n8n-api-key", "N8N_API_KEY"),
    ],
)
def test_n8n_client_fails_closed_when_configuration_is_missing(
    api_url, api_key, message
):
    client = N8nClient(api_url=api_url, api_key=api_key, timeout=30)

    with pytest.raises(N8nClientError, match=message):
        client._get_headers()