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": "admin@example.com"} 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({"admin@example.com"})) 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(" Kyle@Example.com, pat@example.com;sam@example.com ,") == { "kyle@example.com", "pat@example.com", "sam@example.com"} 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": "user@example.com"}).json()["is_admin"] is False assert client.get("/api/v1/me").json()["is_admin"] is False @pytest.mark.parametrize("headers", [{}, {"X-Forwarded-Email": "user@example.com"}]) 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": "Admin@Example.COM"}).status_code == 200 assert client.get("/api/v1/admin/keys", headers={"X-Forwarded-Preferred-Username": "admin@example.com"}).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"] == "admin@example.com" 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"] == "admin@example.com" 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": "admin@example.com"} | 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()