Infrastrukturní úklid: zamčené závislosti, verze schématu i binárky #12

Merged
david-spacil merged 8 commits from agents/claude/infrastruktura into main 2026-08-22 18:42:29 +02:00
3 changed files with 113 additions and 1 deletions
Showing only changes of commit 4ffd7d3c9d - Show all commits

No files matched your search

+6
View File
@@ -95,6 +95,12 @@ sad vedle sebe.
Zálohovat i stěhovat jde prostým zkopírováním souboru; `DICE_DB` si ho
najde kdekoli.
Schéma má verzi uloženou v `PRAGMA user_version` a `storage.migrate()` si
databázi při každém připojení dorovná. Starší soubory z verzí, kdy se verze
schématu ještě nepsala, se poznají podle nuly a orazítkují se. Databázi
založenou novější verzí počitadla appka odmítne otevřít, místo aby ji tiše
rozbila.
Počítá se s domácí sítí — appka nemá přihlašování a kdokoli na stejné WiFi
může zapisovat.
+40 -1
View File
@@ -86,6 +86,45 @@ CREATE TABLE IF NOT EXISTS turns (
CREATE INDEX IF NOT EXISTS turns_game ON turns(game_id, id);
"""
# Každý krok posouvá schéma o jedno dál; krok na indexu 0 vede z verze 1 na 2.
# Přidávat jen na konec a nikdy nesahat na ten, který už někomu proběhl —
# databáze leží lidem na discích a zpátky se nedají stáhnout.
MIGRATIONS: list[str] = []
SCHEMA_VERSION = 1 + len(MIGRATIONS)
def migrate(conn: sqlite3.Connection) -> int:
"""Dorovná databázi na aktuální schéma a vrátí její verzi.
Verze se drží v `PRAGMA user_version`, což je celé číslo, které si SQLite
veze v hlavičce souboru a jinak si ho nevšímá.
Nula znamená dvojí: buď je soubor prázdný, nebo pochází z doby, kdy se
verze ještě nezapisovala. Obojí se vyřeší stejně — `SCHEMA` je samé
`IF NOT EXISTS`, takže na hotové databázi neudělá nic a jen se orazítkuje.
"""
version = conn.execute("PRAGMA user_version").fetchone()[0]
if version > SCHEMA_VERSION:
raise RuntimeError(
f"Databáze je ve verzi {version}, tahle aplikace umí "
f"{SCHEMA_VERSION}. Nejspíš ji založila novější verze počitadla; "
f"aktualizuj, ať o data nepřijdeš.")
if version == 0:
conn.executescript(SCHEMA)
version = 1
for step in MIGRATIONS[version - 1:]:
conn.executescript(step)
version += 1
# PRAGMA neumí zástupné otazníky. Číslo sem teče z kódu, ne od uživatele.
conn.execute(f"PRAGMA user_version = {version}")
conn.commit()
return version
def connect(path: Path | str | None = None) -> sqlite3.Connection:
target = Path(path or DB_PATH)
@@ -93,7 +132,7 @@ def connect(path: Path | str | None = None) -> sqlite3.Connection:
conn = sqlite3.connect(target)
conn.row_factory = sqlite3.Row
conn.execute("PRAGMA foreign_keys = ON")
conn.executescript(SCHEMA)
migrate(conn)
return conn
+67
View File
@@ -1,3 +1,4 @@
import sqlite3
from pathlib import Path
import pytest
@@ -182,3 +183,69 @@ def test_connect_zalozi_chybejici_adresar(tmp_path):
conn.close()
assert (tmp_path / "novy" / "adresar" / "dice.db").exists()
# --- verze schématu ---------------------------------------------------------
def verze(conn) -> int:
return conn.execute("PRAGMA user_version").fetchone()[0]
def test_nova_databaze_je_orazitkovana(tmp_path):
conn = storage.connect(tmp_path / "nova.db")
try:
assert verze(conn) == storage.SCHEMA_VERSION
finally:
conn.close()
def test_databaze_z_doby_pred_verzovanim_prezije(tmp_path):
"""Přesně to, co lidem leží na disku z verzí 1.0 a 1.1.
Tabulky má, razítko ne. Nesmí se ani přijít o data, ani spadnout.
"""
path = tmp_path / "stara.db"
stara = sqlite3.connect(path)
stara.executescript(storage.SCHEMA) # bez PRAGMA user_version
stara.execute("INSERT INTO players (name, key, created_at) "
"VALUES ('Adam', 'adam', '2026-01-01T00:00:00')")
stara.commit()
stara.close()
conn = storage.connect(path)
try:
assert verze(conn) == storage.SCHEMA_VERSION
assert [p["name"] for p in storage.known_players(conn)] == ["Adam"]
finally:
conn.close()
def test_migrace_dojede_jen_ty_chybejici(tmp_path, monkeypatch):
"""Podruhé už krok proběhnout nesmí — jinak by ALTER TABLE spadl."""
monkeypatch.setattr(storage, "MIGRATIONS",
["ALTER TABLE players ADD COLUMN barva TEXT;"])
monkeypatch.setattr(storage, "SCHEMA_VERSION", 2)
path = tmp_path / "migrovana.db"
conn = storage.connect(path)
conn.close()
conn = storage.connect(path) # druhé spuštění téže verze
try:
assert verze(conn) == 2
sloupce = [r[1] for r in conn.execute("PRAGMA table_info(players)")]
assert "barva" in sloupce
finally:
conn.close()
def test_databaze_z_novejsi_verze_se_odmitne(tmp_path):
"""Radši srozumitelná hláška než tiše rozbitá data."""
path = tmp_path / "budouci.db"
conn = storage.connect(path)
conn.execute(f"PRAGMA user_version = {storage.SCHEMA_VERSION + 1}")
conn.commit()
conn.close()
with pytest.raises(RuntimeError, match="novější verze"):
storage.connect(path)