jellomator/tests/test_api.py

import importlib
import os
import sys
import time
from pathlib import Path
from typing import Iterator

import pymysql
import pytest
from fastapi.testclient import TestClient

PROJECT_ROOT = Path(__file__).resolve().parents[1]
if str(PROJECT_ROOT) not in sys.path:
    sys.path.insert(0, str(PROJECT_ROOT))


def wait_for_db(host: str, port: int, user: str, password: str, timeout_seconds: int = 60) -> None:
    deadline = time.time() + timeout_seconds
    while time.time() < deadline:
        try:
            conn = pymysql.connect(host=host, port=port, user=user, password=password, autocommit=True)
            conn.close()
            return
        except Exception:
            time.sleep(1)
    raise RuntimeError("MariaDB did not become ready in time")


@pytest.fixture(scope="session")
def app_module():
    os.environ.setdefault("DB_HOST", "127.0.0.1")
    os.environ.setdefault("DB_PORT", "3306")
    os.environ.setdefault("DB_USER", "jellomator")
    os.environ.setdefault("DB_PASSWORD", "jellomator")
    os.environ.setdefault("DB_NAME", "jellomator_test")
    wait_for_db(
        host=os.environ["DB_HOST"],
        port=int(os.environ["DB_PORT"]),
        user=os.environ["DB_USER"],
        password=os.environ["DB_PASSWORD"],
    )
    module = importlib.import_module("backend.main")
    return module


@pytest.fixture()
def client(app_module) -> Iterator[TestClient]:
    with app_module.db() as conn:
        with conn.cursor() as cur:
            cur.execute("delete from sessions")
            cur.execute("delete from users")
            cur.execute("delete from links")
    app_module.login_attempts.clear()
    app_module.login_lockouts.clear()
    with TestClient(app_module.app) as test_client:
        yield test_client


def test_healthz(client: TestClient):
    resp = client.get("/healthz")
    assert resp.status_code == 200
    assert resp.json() == {"ok": True}


def test_readyz(client: TestClient):
    resp = client.get("/readyz")
    assert resp.status_code == 200
    assert resp.json() == {"ok": True}


def test_setup_and_login(client: TestClient):
    setup_resp = client.post("/api/setup", json={"username": "admin", "password": "123456789012"})
    assert setup_resp.status_code == 200
    assert setup_resp.json() == {"ok": True}

    login_resp = client.post("/api/login", json={"username": "admin", "password": "123456789012"})
    assert login_resp.status_code == 200
    assert login_resp.json() == {"ok": True}
    assert "jellomator_session=" in login_resp.headers.get("set-cookie", "")


def test_link_crud_with_auth(client: TestClient):
    client.post("/api/setup", json={"username": "admin", "password": "123456789012"})
    login_resp = client.post("/api/login", json={"username": "admin", "password": "123456789012"})
    assert login_resp.status_code == 200

    create_resp = client.post(
        "/api/links",
        data={
            "name": "Prowlarr",
            "url": "https://prowlarr.example.com",
            "description": "Indexer manager",
            "category": "Arr",
            "enabled": "true",
        },
    )
    assert create_resp.status_code == 200
    assert create_resp.json() == {"ok": True}

    list_resp = client.get("/api/links")
    assert list_resp.status_code == 200
    rows = list_resp.json()
    assert len(rows) == 1
    assert rows[0]["name"] == "Prowlarr"
    link_id = rows[0]["id"]

    patch_resp = client.patch(
        f"/api/links/{link_id}",
        data={
            "name": "Prowlarr Updated",
            "url": "https://prowlarr.example.com/new",
            "description": "Updated",
            "category": "Arr",
            "enabled": "true",
        },
    )
    assert patch_resp.status_code == 200
    assert patch_resp.json() == {"ok": True}

    delete_resp = client.delete(f"/api/links/{link_id}")
    assert delete_resp.status_code == 200
    assert delete_resp.json() == {"ok": True}


def test_login_rate_limit_lockout(client: TestClient):
    client.post("/api/setup", json={"username": "admin", "password": "123456789012"})
    for _ in range(5):
        resp = client.post("/api/login", json={"username": "admin", "password": "wrong-password"})
        assert resp.status_code == 401
    locked = client.post("/api/login", json={"username": "admin", "password": "wrong-password"})
    assert locked.status_code == 429


def test_backup_restore_dry_run_and_apply(client: TestClient):
    client.post("/api/setup", json={"username": "admin", "password": "123456789012"})
    client.post("/api/login", json={"username": "admin", "password": "123456789012"})
    client.post(
        "/api/links",
        data={
            "name": "Sonarr",
            "url": "https://sonarr.example.com",
            "description": "TV",
            "category": "Arr",
            "enabled": "true",
        },
    )
    backup_resp = client.get("/api/admin/backup")
    assert backup_resp.status_code == 200
    backup = backup_resp.json()
    assert isinstance(backup.get("users"), list)
    assert isinstance(backup.get("links"), list)
    dry = client.post("/api/admin/restore?dry_run=true", json={"data": backup, "confirm": False})
    assert dry.status_code == 200
    assert dry.json()["dry_run"] is True
    apply_resp = client.post("/api/admin/restore?dry_run=false", json={"data": backup, "confirm": True})
    assert apply_resp.status_code == 200
    assert apply_resp.json()["ok"] is True