108 lines
3.4 KiB
Python
108 lines
3.4 KiB
Python
"""Tests für Datenbank-Initialisierung und die SQLite-Hilfsfunktionen."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import sqlite3
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
from flask import Flask
|
|
|
|
from app import create_app
|
|
from app.models import (
|
|
SCHEMA_VERSION,
|
|
connect,
|
|
consume_run_token,
|
|
init_db,
|
|
insert_score,
|
|
rank_of,
|
|
register_submission_attempt,
|
|
top_scores,
|
|
)
|
|
|
|
|
|
def _tables(path: Path) -> set[str]:
|
|
with sqlite3.connect(path) as conn:
|
|
rows = conn.execute("SELECT name FROM sqlite_master WHERE type = 'table'").fetchall()
|
|
return {row[0] for row in rows}
|
|
|
|
|
|
def test_create_app_initializes_database(app: Flask, db_path: Path) -> None:
|
|
assert db_path.exists()
|
|
assert {"highscores", "submission_log", "used_run_tokens"} <= _tables(db_path)
|
|
|
|
|
|
def test_init_db_creates_missing_parent_directories(tmp_path: Path) -> None:
|
|
target = tmp_path / "tief" / "verschachtelt" / "scores.sqlite3"
|
|
init_db(target)
|
|
assert target.exists()
|
|
with sqlite3.connect(target) as conn:
|
|
assert conn.execute("PRAGMA user_version").fetchone()[0] == SCHEMA_VERSION
|
|
|
|
|
|
def test_init_db_is_idempotent_and_keeps_data(tmp_path: Path) -> None:
|
|
target = tmp_path / "scores.sqlite3"
|
|
init_db(target)
|
|
conn = connect(target)
|
|
insert_score(conn, "Bleibt", 42, wave=1, duration_s=10)
|
|
conn.close()
|
|
|
|
init_db(target)
|
|
init_db(target)
|
|
conn = connect(target)
|
|
assert [row["name"] for row in top_scores(conn, 10)] == ["Bleibt"]
|
|
conn.close()
|
|
|
|
|
|
def test_db_path_from_environment(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
|
|
target = tmp_path / "env" / "glory.sqlite3"
|
|
monkeypatch.setenv("GLORY_DB_PATH", str(target))
|
|
monkeypatch.setenv("SECRET_KEY", "env-secret")
|
|
app = create_app()
|
|
assert Path(app.config["DATABASE_PATH"]) == target
|
|
assert target.exists()
|
|
|
|
|
|
def test_init_db_cli_command(app: Flask, db_path: Path) -> None:
|
|
db_path.unlink()
|
|
result = app.test_cli_runner().invoke(args=["init-db"])
|
|
assert result.exit_code == 0
|
|
assert "Datenbank bereit" in result.output
|
|
assert db_path.exists()
|
|
|
|
|
|
def test_rank_of_matches_top_scores_order(tmp_path: Path) -> None:
|
|
target = tmp_path / "rank.sqlite3"
|
|
init_db(target)
|
|
conn = connect(target)
|
|
ids = [
|
|
insert_score(conn, name, score, 1, 1)
|
|
for name, score in [("A", 300), ("B", 500), ("C", 300), ("D", 100)]
|
|
]
|
|
ranks = [rank_of(conn, score_id) for score_id in ids]
|
|
assert ranks == [2, 1, 3, 4]
|
|
conn.close()
|
|
|
|
|
|
def test_register_submission_attempt_limits_and_recovers(tmp_path: Path) -> None:
|
|
target = tmp_path / "limit.sqlite3"
|
|
init_db(target)
|
|
conn = connect(target)
|
|
now = 5_000.0
|
|
results = [register_submission_attempt(conn, "client", now + i, 60, 3) for i in range(4)]
|
|
assert [allowed for allowed, _ in results] == [True, True, True, False]
|
|
# Ältester Versuch bei t=5000 fällt bei t=5060 aus dem Fenster; Anfrage kam bei t=5003.
|
|
assert results[3][1] == 57
|
|
assert register_submission_attempt(conn, "client", now + 61, 60, 3)[0] is True
|
|
conn.close()
|
|
|
|
|
|
def test_consume_run_token_only_once(tmp_path: Path) -> None:
|
|
target = tmp_path / "tokens.sqlite3"
|
|
init_db(target)
|
|
conn = connect(target)
|
|
assert consume_run_token(conn, "run-1", 100.0, 3600) is True
|
|
assert consume_run_token(conn, "run-1", 101.0, 3600) is False
|
|
assert consume_run_token(conn, "run-2", 101.0, 3600) is True
|
|
conn.close()
|