6krrt/tests/test_admin_provider_readonly.py

"""A base-config provider is read-only on every verb, not just delete.

`POST /admin/api/providers/{name}` had no provenance check, which made it
the one way to write to something the portal labels "base config /
read-only". The result could not be undone from the portal: the write forks
the provider into config.local.yaml, and DELETE then refuses if it is the
default_provider, leaving hand-editing that file as the only way out.
"""

from __future__ import annotations

import sqlite3
from pathlib import Path

import pytest
from starlette.testclient import TestClient
from fastapi import FastAPI

import admin
from config import load_config

ROOT = Path(__file__).resolve().parent.parent
SCHEMA_SQL = (ROOT / "config" / "schema.sql").read_text()

BASE_PROVIDER = {
    "base_url": "https://api.base.example/v1",
    "api_key_env": "BASE_KEY",
    "has_energy_telemetry": True,
    "enabled": True,
}


@pytest.fixture
def client(tmp_path):
    """An admin router rooted at a throwaway config dir.

    base_dir is what decides which config.yaml/config.local.yaml the router
    reads and writes, so pointing it at tmp_path isolates the test from the
    machine's real overlay entirely.

    The extra provider is injected into the parsed document rather than
    appended as text: a second ``dispatch_providers:`` key is a YAML
    duplicate, and last-one-wins would silently delete the real providers --
    taking ``dispatch_settings.default_provider`` down with them.
    """
    import yaml

    cfg_dir = tmp_path / "config"
    cfg_dir.mkdir()
    doc = yaml.safe_load((ROOT / "config" / "config.yaml").read_text())
    doc.setdefault("dispatch_providers", {})["basedude"] = dict(BASE_PROVIDER)
    (cfg_dir / "config.yaml").write_text(yaml.safe_dump(doc, sort_keys=False))

    db_path = tmp_path / "t.db"
    conn = sqlite3.connect(db_path)
    conn.executescript(SCHEMA_SQL)
    admin.ensure_admin_tables(conn)
    conn.commit()
    conn.close()

    def _db():
        c = sqlite3.connect(db_path)
        c.row_factory = sqlite3.Row
        return c

    cfg = load_config(cfg_dir / "config.yaml")
    app = FastAPI()
    app.include_router(admin.build_router(cfg, _db, str(tmp_path)),
                       prefix="/admin")
    return TestClient(app), cfg_dir


def _body(**over):
    body = {
        "base_url": "https://api.changed.example/v1",
        "api_key_env": "CHANGED_KEY",
        "has_energy_telemetry": False,
        "enabled": True,
    }
    body.update(over)
    return body


def test_updating_a_base_config_provider_is_refused(client):
    c, cfg_dir = client
    resp = c.post("/admin/api/providers/basedude", json=_body())
    assert resp.status_code == 403, resp.text
    assert "read-only" in resp.text
    # And nothing was written: the overlay must not exist or must not carry it.
    overlay = cfg_dir / "config.local.yaml"
    if overlay.exists():
        assert "basedude" not in overlay.read_text()


def test_the_refusal_does_not_advise_an_impossible_workaround(client):
    """The message must not send the operator somewhere the API refuses.

    The profile endpoints used to say "add an overlay profile with the same
    name to override it", which admin_profile_create rejects with a 409.
    """
    c, _ = client
    detail = c.post("/admin/api/providers/basedude", json=_body()).json()["detail"]
    assert "override it" not in detail
    assert "config.yaml" in detail


def test_a_new_provider_is_still_creatable(client):
    c, cfg_dir = client
    resp = c.post("/admin/api/providers/brandnew", json=_body())
    assert resp.status_code == 200, resp.text
    assert "brandnew" in (cfg_dir / "config.local.yaml").read_text()


def test_an_overlay_provider_is_still_editable(client):
    """The guard keys on base-only, so an overlay entry stays editable."""
    c, cfg_dir = client
    assert c.post("/admin/api/providers/mine", json=_body()).status_code == 200
    resp = c.post("/admin/api/providers/mine",
                  json=_body(base_url="https://api.second.example/v1"))
    assert resp.status_code == 200, resp.text
    assert "api.second.example" in (cfg_dir / "config.local.yaml").read_text()


def test_delete_still_refuses_a_base_config_provider(client):
    """The behaviour update was inconsistent with, unchanged."""
    c, _ = client
    resp = c.delete("/admin/api/providers/basedude")
    assert resp.status_code == 403, resp.text