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

"""Code the engine runs is a superuser's to change; everything else is not."""

from collections.abc import Callable
from typing import Any

import pytest
from fastapi.testclient import TestClient

from fluksio.api.deps import CODE_IS_SUPERUSERS
from fluksio.api.routes import modules as modules_route
from fluksio.core.config import settings

PREFIX = f"{settings.API_V1_STR}/flows"
FLOW = "code_rights"


def a_flow(version: int, **changes: Any) -> dict[str, Any]:
    nodes: dict[str, dict[str, Any]] = {
        "fn": {
            "id": "fn",
            "type": "python",
            "params": {"gain": 1},
            "provides": [{"name": "reading", "dtype": "float"}],
        },
        "cmd": {"id": "cmd", "type": "exec", "params": {"command": "echo hi"}},
        "hold": {"id": "hold", "type": "delay", "params": {"delay": 1}},
    }
    for node_id, patch in changes.items():
        nodes[node_id] = {**nodes.get(node_id, {"id": node_id}), **patch}
    return {"name": FLOW, "version": version, "nodes": list(nodes.values())}


@pytest.fixture
def save(
    client: TestClient, superuser_token_headers: dict[str, str]
) -> Callable[..., Any]:
    """Seed the flow as a superuser, then save a change as someone else."""
    client.delete(f"{PREFIX}/{FLOW}", headers=superuser_token_headers)
    seeded = client.post(
        f"{PREFIX}/", headers=superuser_token_headers, json={**a_flow(0), "id": FLOW}
    )
    assert seeded.status_code == 201, seeded.text
    version = seeded.json()["definition"]["version"]

    def put(headers: dict[str, str], **changes: Any) -> Any:
        return client.put(
            f"{PREFIX}/{FLOW}", headers=headers, json=a_flow(version, **changes)
        )

    return put


@pytest.mark.parametrize(
    "changes",
    [
        {"new": {"type": "python"}},
        {"new": {"type": "exec", "params": {"command": "id"}}},
        {"cmd": {"params": {"command": "cat /proc/self/environ"}}},
        {"hold": {"type": "exec", "params": {"command": "id"}}},
        {"fn": {"source_ref": "someone_elses"}},
        {"fn": {"resources": {"env": {"PYTHONPATH": "/data/files"}}}},
    ],
    ids=["add-function", "add-exec", "exec-command", "retype", "repoint", "env"],
)
def test_a_user_cannot_change_code_in_a_flow(
    save: Callable[..., Any],
    normal_user_token_headers: dict[str, str],
    superuser_token_headers: dict[str, str],
    changes: dict[str, Any],
) -> None:
    refused = save(normal_user_token_headers, **changes)
    assert refused.status_code == 403
    assert refused.json()["detail"].startswith(CODE_IS_SUPERUSERS)
    assert save(superuser_token_headers, **changes).status_code == 200


def test_a_user_keeps_everything_but_the_code(
    save: Callable[..., Any], normal_user_token_headers: dict[str, str]
) -> None:
    saved = save(
        normal_user_token_headers,
        fn={
            "title": "Renamed",
            "params": {"gain": 2},
            "requires": [{"name": "trigger", "dtype": "float"}],
        },
        hold={"params": {"delay": 5}},
        late={"type": "delay", "params": {"delay": 2}},
    )
    assert saved.status_code == 200, saved.text


@pytest.mark.usefixtures("save")
def test_duplicating_code_nodes_is_a_superusers(
    client: TestClient,
    normal_user_token_headers: dict[str, str],
    superuser_token_headers: dict[str, str],
) -> None:
    url = f"{PREFIX}/{FLOW}/duplicate"

    refused = client.post(url, headers=normal_user_token_headers)
    assert refused.status_code == 403
    assert refused.json()["detail"].startswith(CODE_IS_SUPERUSERS)

    copied = client.post(url, headers=superuser_token_headers)
    assert copied.status_code == 200, copied.text
    name = copied.json()["definition"]["name"]
    client.delete(f"{PREFIX}/{name}", headers=superuser_token_headers)


@pytest.mark.usefixtures("save")
def test_node_source_is_a_superusers_to_write(
    client: TestClient,
    normal_user_token_headers: dict[str, str],
    superuser_token_headers: dict[str, str],
) -> None:
    url = f"{PREFIX}/{FLOW}/nodes/fn/source"
    body = {"code": "def process():\n    return {'reading': 1.0}\n"}

    refused = client.put(url, headers=normal_user_token_headers, json=body)
    assert refused.status_code == 403
    assert refused.json()["detail"] == CODE_IS_SUPERUSERS
    # Reading it is still anyone's.
    assert client.get(url, headers=normal_user_token_headers).status_code == 200

    saved = client.put(url, headers=superuser_token_headers, json=body)
    assert saved.status_code == 200, saved.text


def test_installing_modules_is_a_superusers(
    client: TestClient,
    monkeypatch: pytest.MonkeyPatch,
    normal_user_token_headers: dict[str, str],
    superuser_token_headers: dict[str, str],
) -> None:
    installs: list[str] = []

    def sync(requirements: str) -> tuple[bool, str]:
        installs.append(requirements)
        return True, ""

    async def retire(*_: Any) -> None:
        return None

    monkeypatch.setattr(modules_route.modules, "sync", sync)
    monkeypatch.setattr(modules_route, "_retire", retire)
    url = f"{settings.API_V1_STR}/modules/apply"
    body = {"requirements": ""}

    refused = client.post(url, headers=normal_user_token_headers, json=body)
    assert refused.status_code == 403
    assert refused.json()["detail"] == CODE_IS_SUPERUSERS
    assert installs == []

    assert client.post(url, headers=superuser_token_headers, json=body).is_success
    assert installs == [""]