Backup-Dashboard/backend/tests/test_admin.py

import sqlite3

import pytest
from fastapi.testclient import TestClient

from app import db
from app.main import create_app
from app.settings import Settings, parse_admins

ADMIN = {"X-Forwarded-Email": "[email protected]"}
PAGE = {"X-Requested-By": "backup-dashboard"}
ADMIN_PAGE = ADMIN | PAGE


@pytest.fixture
def client(tmp_path):
    settings = Settings(db_path=tmp_path / "test.db", admins=frozenset({"[email protected]"}))
    with TestClient(create_app(settings)) as c:
        yield c


def create(client, name="veeam-acme", headers=ADMIN_PAGE):
    return client.post("/api/v1/admin/keys", json={"name": name}, headers=headers)


def ingest(client, secret, tenant="Acme"):
    return client.post(
        "/api/v1/ingest",
        json={"tenant": tenant, "source_system": "Veeam B&R", "job": {"external_id": "j1", "name": "Job"}},
        headers={"Authorization": f"Bearer {secret}"},
    )


def test_parse_admins():
    assert parse_admins(" [email protected], [email protected];[email protected] ,") == {
        "[email protected]", "[email protected]", "[email protected]"}
    assert parse_admins("") == frozenset()


def test_me_says_whether_the_user_is_an_admin(client):
    assert client.get("/api/v1/me", headers=ADMIN).json()["is_admin"] is True
    assert client.get("/api/v1/me", headers={"X-Forwarded-Email": "[email protected]"}).json()["is_admin"] is False
    assert client.get("/api/v1/me").json()["is_admin"] is False


@pytest.mark.parametrize("headers", [{}, {"X-Forwarded-Email": "[email protected]"}])
def test_admin_pages_are_admins_only(client, headers):
    assert client.get("/api/v1/admin/keys", headers=headers).status_code == 403
    assert create(client, headers=headers | PAGE).status_code == 403
    assert client.post("/api/v1/admin/keys/1/revoke", headers=headers | PAGE).status_code == 403


def test_admin_match_ignores_case_and_accepts_the_sign_in_name(client):
    assert client.get("/api/v1/admin/keys", headers={"X-Forwarded-Email": "[email protected]"}).status_code == 200
    assert client.get("/api/v1/admin/keys", headers={"X-Forwarded-Preferred-Username": "[email protected]"}).status_code == 200


def test_changes_need_the_page_header(client):
    # Without it, a page on another site could make a signed-in admin's browser send the request
    assert create(client, headers=ADMIN).status_code == 403
    secret = create(client).json()["secret"]
    key_id = client.get("/api/v1/admin/keys", headers=ADMIN).json()["keys"][0]["id"]
    assert client.post(f"/api/v1/admin/keys/{key_id}/revoke", headers=ADMIN).status_code == 403
    assert ingest(client, secret).status_code == 200  # still active


@pytest.mark.parametrize("name", ["", " ", "-leading-dash", "has space", "a" * 65, "semi;colon"])
def test_rejects_bad_key_names(client, name):
    assert create(client, name=name).status_code == 422


def test_created_key_is_shown_once_and_works_for_ingest(client):
    r = create(client, name="  veeam-acme  ")
    assert r.status_code == 201
    body = r.json()
    assert body["secret"].startswith("bdk_")
    assert body["key"]["name"] == "veeam-acme"
    assert body["key"]["created_by"] == "[email protected]"
    listed = client.get("/api/v1/admin/keys", headers=ADMIN).json()["keys"]
    assert "secret" not in listed[0] and "key_hash" not in listed[0]
    assert ingest(client, body["secret"]).status_code == 200


def test_key_records_when_and_for_which_company_it_was_last_used(client):
    secret = create(client).json()["secret"]
    ingest(client, secret, tenant="Acme Corp")
    key = client.get("/api/v1/admin/keys", headers=ADMIN).json()["keys"][0]
    assert key["last_used_at"] is not None
    assert key["last_used_tenant"] == "Acme Corp"


def test_duplicate_names_are_refused_even_after_revoking(client):
    key_id = create(client).json()["key"]["id"]
    assert create(client).status_code == 409
    client.post(f"/api/v1/admin/keys/{key_id}/revoke", headers=ADMIN_PAGE)
    assert create(client).status_code == 409


def test_revoking_stops_the_key_and_records_who(client):
    created = create(client).json()
    r = client.post(f"/api/v1/admin/keys/{created['key']['id']}/revoke", headers=ADMIN_PAGE)
    assert r.status_code == 200
    assert r.json()["key"]["revoked_by"] == "[email protected]"
    assert r.json()["key"]["revoked_at"] is not None
    assert ingest(client, created["secret"]).status_code == 401


def test_revoking_twice_keeps_the_first_revocation(client):
    key_id = create(client).json()["key"]["id"]
    first = client.post(f"/api/v1/admin/keys/{key_id}/revoke", headers=ADMIN_PAGE).json()["key"]
    again = client.post(f"/api/v1/admin/keys/{key_id}/revoke",
                        headers={"X-Forwarded-Email": "[email protected]"} | PAGE).json()["key"]
    assert again["revoked_at"] == first["revoked_at"]


def test_revoking_an_unknown_key_is_404(client):
    assert client.post("/api/v1/admin/keys/999/revoke", headers=ADMIN_PAGE).status_code == 404


def test_lists_active_keys_first(client):
    for name in ("b-key", "a-key", "c-key"):
        create(client, name=name)
    a_id = next(k["id"] for k in client.get("/api/v1/admin/keys", headers=ADMIN).json()["keys"] if k["name"] == "a-key")
    client.post(f"/api/v1/admin/keys/{a_id}/revoke", headers=ADMIN_PAGE)
    names = [k["name"] for k in client.get("/api/v1/admin/keys", headers=ADMIN).json()["keys"]]
    assert names == ["b-key", "c-key", "a-key"]


def test_upgrades_a_version_1_database_keeping_its_keys(tmp_path):
    path = tmp_path / "old.db"
    conn = sqlite3.connect(path)
    conn.executescript(db.SCHEMA)  # the v1 schema, as deployed before admin pages
    conn.execute("INSERT INTO api_keys (name, key_hash, created_at) VALUES ('old-key', 'abc', '2026-10-01T00:00:00Z')")
    conn.execute("PRAGMA user_version = 1")
    conn.commit()
    conn.close()

    db.init_db(path)
    db.init_db(path)  # and running it again is harmless

    conn = db.connect(path)
    assert conn.execute("PRAGMA user_version").fetchone()[0] == db.SCHEMA_VERSION
    row = dict(conn.execute("SELECT * FROM api_keys").fetchone())
    assert row["name"] == "old-key"
    assert {"created_by", "revoked_by", "last_used_tenant"} <= row.keys()
    conn.close()