app/backend/tests/api/routes/test_search.py

"""The one index the global search matches against."""

from typing import Any

import pytest
from fastapi.testclient import TestClient

from fluksio.api.routes.search import MASK, setting_text
from fluksio.core.config import settings

PREFIX = f"{settings.API_V1_STR}/search"
FLOWS = f"{settings.API_V1_STR}/flows"
DASHBOARDS = f"{settings.API_V1_STR}/dashboards"
SECRETS = f"{settings.API_V1_STR}/secrets"


def test_search_requires_authentication(client: TestClient) -> None:
    assert client.get(f"{PREFIX}/").status_code == 401


def test_index_reaches_inside_flows_and_dashboards(
    client: TestClient, superuser_token_headers: dict[str, str]
) -> None:
    """A node and a widget are the point: neither is on any list endpoint."""
    client.put(
        f"{FLOWS}/searchable",
        headers=superuser_token_headers,
        json={
            "name": "searchable",
            "title": "Searchable",
            "nodes": [{"id": "sensor", "type": "python", "title": "Hall sensor"}],
        },
    )
    client.put(
        f"{DASHBOARDS}/hall",
        headers=superuser_token_headers,
        json={
            "name": "hall",
            "title": "Hall",
            "widgets": [
                {
                    "id": "temperature",
                    "type": "stat",
                    "title": "Temperature",
                    "layout": {"lg": {"x": 0, "y": 0, "w": 3, "h": 2}},
                    "config": {"message": "hall.temperature", "dtype": "float"},
                }
            ],
            "version": 0,
        },
    )

    entries = client.get(f"{PREFIX}/", headers=superuser_token_headers).json()
    # Keyed on the parent too: an id is only unique within the document it is
    # in, and the other suites seed their own `sensor` and `temperature`.
    found = {
        (entry["category"], entry["parent"], entry["name"]): entry for entry in entries
    }

    assert found[("flow", "", "searchable")]["title"] == "Searchable"
    assert found[("node", "searchable", "sensor")]["title"] == "Hall sensor"
    assert found[("node", "searchable", "sensor")]["kind"] == "python"
    assert found[("dashboard", "", "hall")]["title"] == "Hall"
    assert found[("widget", "hall", "temperature")]["title"] == "Temperature"
    assert found[("widget", "hall", "temperature")]["kind"] == "stat"


def test_index_lists_a_flows_inputs_and_outputs(
    client: TestClient, superuser_token_headers: dict[str, str]
) -> None:
    """Named by the qualified message, which is the label the canvas draws."""
    client.put(
        f"{FLOWS}/fit",
        headers=superuser_token_headers,
        json={
            "name": "fit",
            "mode": "batch",
            "inputs": [{"spec": {"name": "epochs", "dtype": "int"}}],
            "outputs": ["accuracy"],
        },
    )

    entries = client.get(f"{PREFIX}/", headers=superuser_token_headers).json()
    found = {(e["category"], e["name"]): e for e in entries if e["parent"] == "fit"}

    assert found[("input", "fit.epochs")]["title"] == "epochs"
    assert found[("input", "fit.epochs")]["kind"] == "int"
    assert found[("output", "fit.accuracy")]["title"] == "accuracy"


def test_secrets_are_named_only_to_a_superuser(
    client: TestClient,
    superuser_token_headers: dict[str, str],
    normal_user_token_headers: dict[str, str],
) -> None:
    client.put(
        f"{SECRETS}/broker_password",
        headers=superuser_token_headers,
        json={"value": "hunter2"},
    )

    def secrets(headers: dict[str, str]) -> set[str]:
        entries = client.get(f"{PREFIX}/", headers=headers).json()
        return {e["name"] for e in entries if e["category"] == "secret"}

    assert "broker_password" in secrets(superuser_token_headers)
    assert secrets(normal_user_token_headers) == set()


def test_a_setting_never_shows_a_credential() -> None:
    """The masking rules, one by one: nothing secret reaches the index."""
    plain = {"type": "number"}
    secret = {"type": "string", "x-secret": True}

    assert setting_text("interval", 5, plain, False) == "5"
    assert setting_text("at_start", True, None, False) == "true"
    assert setting_text("topic", "", plain, False) == ""
    assert setting_text("topic", None, plain, False) == ""

    # A declared credential: a reference is named for those who may read names,
    # a plain string never appears.
    ref = {"$secret": "ntfy_token"}
    assert setting_text("token", ref, secret, True) == "secret ntfy_token"
    assert setting_text("token", ref, secret, False) == "secret"
    assert setting_text("token", "hunter2", secret, True) == MASK

    # An undeclared one is judged by its key, and a reference stays a reference.
    assert setting_text("api_key", "hunter2", None, True) == MASK
    assert setting_text("api_key", ref, None, False) == "secret"

    # Nested, such as a header: what is neutral stays, what is not goes.
    headers: dict[str, Any] = {
        "Accept": "text/plain",
        "Authorization": "Bearer hunter2",
        "X-Api-Key": {"$secret": "key_name"},
    }
    text = setting_text("headers", headers, {"type": "object"}, False)
    assert "hunter2" not in text and "key_name" not in text
    assert '"Accept":"text/plain"' in text
    assert f'"Authorization":"{MASK}"' in text
    assert '"X-Api-Key":"secret"' in text

    # A long value is clipped, so a document is not an index entry.
    clipped = setting_text("payload", "x" * 500, plain, False)
    assert len(clipped) == 80 and clipped.endswith("\u2026")


def test_index_lists_a_nodes_settings(
    client: TestClient,
    superuser_token_headers: dict[str, str],
    normal_user_token_headers: dict[str, str],
) -> None:
    """A setting is found by its node and name, with its value beside it."""
    client.put(
        f"{FLOWS}/tuned",
        headers=superuser_token_headers,
        json={
            "name": "tuned",
            "nodes": [
                {"id": "gain", "type": "python", "params": {"factor": 2.5}},
                {
                    "id": "hook",
                    "type": "http",
                    "params": {
                        "url": "/tick",
                        "secret": {"$secret": "hook_secret"},
                        "headers": {"Authorization": "Bearer s3cr3t-value"},
                    },
                },
            ],
        },
    )

    def settings_of(headers: dict[str, str]) -> tuple[dict[str, dict], str]:
        response = client.get(f"{PREFIX}/", headers=headers)
        found = {
            e["name"]: e
            for e in response.json()
            if e["category"] == "setting" and e["parent"] == "tuned"
        }
        return found, response.text

    found, _ = settings_of(superuser_token_headers)
    assert found["gain.factor"]["detail"] == "2.5"
    assert found["gain.factor"]["kind"] == "python"
    assert found["hook.url"]["detail"] == "/tick"
    assert found["hook.secret"]["detail"] == "secret hook_secret"

    # Who may not read a secret's name does not get it from here either.
    found, text = settings_of(normal_user_token_headers)
    assert found["hook.secret"]["detail"] == "secret"
    assert "hook_secret" not in text
    assert "s3cr3t-value" not in text


def test_alert_channels_are_named_only_to_a_superuser(
    client: TestClient,
    superuser_token_headers: dict[str, str],
    normal_user_token_headers: dict[str, str],
    tmp_path: Any,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    from fluksio.api.routes.alerts import write_config
    from fluksio.flow.alerts import AlertsConfig, Channel

    monkeypatch.setattr(settings, "ALERTS_FILE", tmp_path / "alerts.json")
    write_config(AlertsConfig(channels=[Channel(name="phone", kind="ntfy")]))

    def channels(headers: dict[str, str]) -> set[str]:
        entries = client.get(f"{PREFIX}/", headers=headers).json()
        return {e["name"] for e in entries if e["category"] == "alert"}

    assert channels(superuser_token_headers) == {"phone"}
    assert channels(normal_user_token_headers) == set()