tests/test_machine_api_auth.py

"""The `gr.api` machine endpoints are gated by caller-token verification (ADR-0004/0018).

A machine call carries no OAuth session, so identity comes from a caller token resolved via
`whoami`. These endpoints are the orchestrator's front door — if they are open, the allow-list
on the UI is decorative.
"""

from __future__ import annotations

import json

import pytest

import gradio_ui


class _Request:
    """Minimal stand-in for `gr.Request` — only `.headers.get` is used."""

    def __init__(self, headers=None):
        self.headers = headers or {}


@pytest.fixture(autouse=True)
def _enforce(monkeypatch, tmp_path):
    monkeypatch.setenv("ACCESS_CONTROL", "enforce")
    monkeypatch.setenv("ALLOWED_IDS", "orchestrator-bot")
    monkeypatch.setenv("LOG_SINK", "local")
    monkeypatch.setenv("LOG_SINK_LOCAL_DIR", str(tmp_path / "run_logs"))
    gradio_ui._WHOAMI_CACHE.clear()
    yield
    gradio_ui._WHOAMI_CACHE.clear()


def _explode(*a, **k):  # pragma: no cover
    raise AssertionError("tool was called despite a denied access decision")


MACHINE_CASES = [
    ("query_variant_status", {"genes": "KRAS", "source": "cbioportal:paad_tcga"}),
    ("variant_by_subtype", {"genes": "KRAS", "source": "cbioportal:paad_tcga"}),
    ("panel", {}),
]


@pytest.mark.parametrize("endpoint,kwargs", MACHINE_CASES)
def test_machine_endpoint_denies_tokenless_call(monkeypatch, endpoint, kwargs):
    """No token header ⇒ denied, fail-closed, before any tool call."""
    monkeypatch.setattr(gradio_ui, "_query_variant_status", _explode)
    monkeypatch.setattr(gradio_ui, "_variant_by_subtype", _explode)

    body = json.loads(getattr(gradio_ui, endpoint)(**kwargs, request=_Request()))

    assert body["status"] == "denied"


@pytest.mark.parametrize("endpoint,kwargs", MACHINE_CASES)
def test_machine_endpoint_denies_unresolvable_token(monkeypatch, endpoint, kwargs):
    """A token `whoami` cannot resolve is an anonymous caller, not a trusted one."""
    monkeypatch.setattr(gradio_ui, "_resolve_token_identity", lambda token: None)
    monkeypatch.setattr(gradio_ui, "_query_variant_status", _explode)
    monkeypatch.setattr(gradio_ui, "_variant_by_subtype", _explode)

    request = _Request({"x-orchestrator-token": "Bearer nonsense"})
    body = json.loads(getattr(gradio_ui, endpoint)(**kwargs, request=request))

    assert body["status"] == "denied"


def test_machine_endpoint_denies_resolvable_but_unlisted_identity(monkeypatch):
    """A real HF account that is not allow-listed is still refused."""
    monkeypatch.setattr(gradio_ui, "_resolve_token_identity", lambda token: "random-person")
    monkeypatch.setattr(gradio_ui, "_query_variant_status", _explode)

    request = _Request({"x-orchestrator-token": "tok"})
    body = json.loads(gradio_ui.query_variant_status(genes="KRAS", source="x", request=request))

    assert body["status"] == "denied"
    assert "random-person" in body["reason"]


def test_machine_endpoint_allows_listed_identity(monkeypatch):
    monkeypatch.setattr(gradio_ui, "_resolve_token_identity", lambda token: "orchestrator-bot")
    monkeypatch.setattr(
        gradio_ui,
        "_query_variant_status",
        lambda genes, source: {"n_samples": 7, "genes": {}, "source": source},
    )

    request = _Request({"x-orchestrator-token": "Bearer good-token"})
    body = json.loads(
        gradio_ui.query_variant_status(
            genes="KRAS,TP53", source="cbioportal:paad_tcga", request=request
        )
    )

    assert body["n_samples"] == 7


def test_bearer_prefix_is_stripped(monkeypatch):
    seen = {}
    monkeypatch.setattr(
        gradio_ui, "_resolve_token_identity", lambda token: seen.setdefault("token", token)
    )
    gradio_ui._machine_caller_identity(_Request({"x-orchestrator-token": "Bearer abc123"}))
    assert seen["token"] == "abc123"


def test_empty_genes_falls_back_to_full_panel():
    """The machine contract sends a comma-separated string; empty means 'the whole panel'."""
    assert gradio_ui._split_genes("") == list(gradio_ui.PANEL)
    assert gradio_ui._split_genes("kras, tp53") == ["KRAS", "TP53"]


# --------------------------------------------------------------------------- #
# Denial DIAGNOSIS — the three failure modes must be distinguishable
# --------------------------------------------------------------------------- #
# Regression guard for a real misdiagnosis (2026-07-27→28). The orchestrator was denied and
# the response said "Please sign in with your HuggingFace account", so the investigation went
# server-side — probing headers, suspecting `whoami`. Nothing was wrong here: the caller had
# simply sent no header at all (its own HF_TOKEN was unset). These three states are
# operationally different and must never again be one indistinguishable message.


def test_no_token_header_is_diagnosed_as_caller_misconfiguration(monkeypatch):
    monkeypatch.setattr(gradio_ui, "_query_variant_status", _explode)
    body = json.loads(gradio_ui.query_variant_status(source="x", request=_Request()))

    assert body["status"] == "denied"
    assert body["machine_auth"] == "no_token_header"
    # names the calling Space's secret, and does NOT tell a machine to sign in
    assert "HF_TOKEN" in body["reason"]
    assert "sign in" not in body["reason"].lower()


def test_unresolvable_token_is_diagnosed_separately_from_a_missing_one(monkeypatch):
    monkeypatch.setattr(gradio_ui, "_resolve_token_identity", lambda token: None)
    monkeypatch.setattr(gradio_ui, "_query_variant_status", _explode)

    body = json.loads(
        gradio_ui.query_variant_status(
            source="x", request=_Request({"x-orchestrator-token": "bad-token"})
        )
    )

    assert body["machine_auth"] == "token_unresolved"


def test_resolved_but_unlisted_identity_is_the_only_real_authorization_denial(monkeypatch):
    """Identity resolved fine — this one IS an allow-list decision, and still names who."""
    monkeypatch.setattr(gradio_ui, "_resolve_token_identity", lambda token: "stranger-bot")
    monkeypatch.setattr(gradio_ui, "_query_variant_status", _explode)

    body = json.loads(
        gradio_ui.query_variant_status(
            source="x", request=_Request({"x-orchestrator-token": "good-token"})
        )
    )

    assert body["machine_auth"] == "not_allowlisted"
    assert "stranger-bot" in body["reason"]


def test_identity_resolution_reports_resolved_on_success(monkeypatch):
    monkeypatch.setattr(gradio_ui, "_resolve_token_identity", lambda token: "orchestrator-bot")
    identity, diagnosis = gradio_ui._machine_caller_identity(
        _Request({"x-orchestrator-token": "good-token"})
    )
    assert (identity, diagnosis) == ("orchestrator-bot", "resolved")


# --------------------------------------------------------------------------- #
# Argument DIAGNOSIS — the same lesson, applied to caller-side type mistakes
# --------------------------------------------------------------------------- #
# Found while verifying the ADR-0007 deploy: passing the obvious-looking
# `genes=["KRAS","TP53"]` returned
#   {"status": "error", "reason": "AttributeError: 'list' object has no attribute 'split'"}
# — indistinguishable from a server fault, so it sends the reader into this repo when the fix
# is in their own call. Exactly the failure mode the denial-diagnosis block above exists for.
def _allowed(monkeypatch):
    monkeypatch.setattr(gradio_ui, "_resolve_token_identity", lambda token: "orchestrator-bot")
    return _Request({"x-orchestrator-token": "tok"})


def test_list_genes_reports_a_caller_side_argument_error(monkeypatch):
    """A list of genes is named as such, with the exact string to send instead."""
    request = _allowed(monkeypatch)
    monkeypatch.setattr(gradio_ui, "_query_variant_status", _explode)  # never runs

    body = json.loads(
        gradio_ui.query_variant_status(
            genes=["KRAS", "TP53"], source="cbioportal:paad_tcga", request=request
        )
    )

    assert body["machine_input"] == "genes_not_a_string"
    assert 'genes="KRAS,TP53"' in body["reason"]  # the fix, spelled out
    assert "caller-side" in body["reason"]
    assert "AttributeError" not in body["reason"]


def test_non_string_source_does_not_raise_before_the_gate(monkeypatch):
    """`registered_source` runs pre-gate; a non-string `source` must not blow up there.

    This was an uncaught exception (a 500), not a result — the one input that escaped
    `_safe_result` entirely because the audit descriptor is built before the auth check.
    """
    request = _allowed(monkeypatch)
    monkeypatch.setattr(gradio_ui, "_query_variant_status", _explode)

    body = json.loads(
        gradio_ui.query_variant_status(genes="KRAS", source=["cbioportal:paad_tcga"], request=request)
    )

    assert body["machine_input"] == "source_not_a_string"


def test_argument_errors_are_reported_only_after_the_auth_gate(monkeypatch):
    """A tokenless caller gets `denied`, not a critique of their arguments."""
    monkeypatch.setattr(gradio_ui, "_query_variant_status", _explode)

    body = json.loads(
        gradio_ui.query_variant_status(genes=["KRAS"], source=1234, request=_Request())
    )

    assert body["status"] == "denied"
    assert "machine_input" not in body


@pytest.mark.parametrize("field", ["subtype_attribute", "modality"])
def test_every_string_argument_is_covered(monkeypatch, field):
    request = _allowed(monkeypatch)
    monkeypatch.setattr(gradio_ui, "_variant_by_subtype", _explode)

    body = json.loads(
        gradio_ui.variant_by_subtype(
            genes="KRAS", source="cbioportal:paad_tcga", request=request, **{field: ["x"]}
        )
    )

    assert body["machine_input"] == f"{field}_not_a_string"


def test_valid_strings_are_unaffected(monkeypatch):
    """The happy path must not acquire a new way to fail."""
    request = _allowed(monkeypatch)
    monkeypatch.setattr(
        gradio_ui, "_query_variant_status", lambda genes, source: {"n_samples": 7, "genes": {}}
    )

    body = json.loads(
        gradio_ui.query_variant_status(
            genes="KRAS,TP53", source="cbioportal:paad_tcga", request=request
        )
    )

    assert body["n_samples"] == 7
    assert "machine_input" not in body