fix: A31 — transaction par migration + helper columns() (v7.6.0)
FlowDeck CI / lint (push) Canceled after 0s
FlowDeck CI / test (push) Canceled after 0s
FlowDeck CI / docker (push) Canceled after 0s

- `_apply_one()` : BEGIN explicite → `fn(conn)` → marque `schema_version` →
  commit ; rollback complet à l'échec. Avant le DDL sortait en autocommit
  (isolation_level legacy) : un échec au milieu laissait un schéma partiel
  commité SANS ligne de version, et la reprise rejouait un DDL déjà appliqué.
  Si une transaction englobante subsiste (init_db commit juste avant), on la
  vide d'abord plutôt que de l'englober.
- Helper unique `columns(conn, table)` (valide l'identifiant, ValueError sinon)
  : 25 copies de `{r[1] for r in conn.execute("PRAGMA table_info(...)")}`
  éliminées dans migrations.py (21 littéraux + 3 f-string + 1 variante row).
  `table_exists`/`column_exists` préconisés par l'audit NON livrés : aucune
  migration n'interroge sqlite_master, un contrôle unitaire se lit dans le set.
- Smoke : DB fraîche → 28 migrations → version 29, ré-apply idempotent.

tests : test_migration_transaction_rolls_back (DDL partiel annulé + zéro marque
de version), test_columns_helper_validates_table_name

suite **1036/1036** · `ruff check app tests` OK · OpenAPI 511 chemins / 7.6.0
docs (ROADMAP/CHANGELOG/WORKLOAD/VERSION) à jour
This commit is contained in:
2026-10-01 10:13:23 -04:00
parent 7be96f0618
commit f706424f90
8 changed files with 140 additions and 34 deletions
+52
View File
@@ -163,6 +163,58 @@ def test_gitea_cache_evicts_expired():
assert "k" not in c._cache and c._cache["k2"][1] == "v2"
def test_migration_transaction_rolls_back():
"""A31 : un échec au milieu d'une migration ne laisse ni DDL partiel, ni
ligne dans schema_version → la reprise rejoue proprement."""
import sqlite3 as _sqlite3
import pytest as _pytest
from app.migrations import _apply_one, _ensure_table
conn = _sqlite3.connect(":memory:")
_ensure_table(conn)
def boom(c):
c.execute("CREATE TABLE partial_x (id INTEGER)")
raise RuntimeError("boom")
with _pytest.raises(RuntimeError, match="boom"):
_apply_one(conn, 9999, "boom", boom)
assert (
conn.execute("SELECT name FROM sqlite_master WHERE name='partial_x'").fetchone()
is None
), "DDL partiel non annulé"
assert (
conn.execute("SELECT COUNT(*) FROM schema_version WHERE version=9999").fetchone()[0]
== 0
)
# chemin nominal : DDL + marque de version dans la même transaction
_apply_one(conn, 9998, "ok", lambda c: c.execute("CREATE TABLE ok_x (id INTEGER)"))
assert (
conn.execute("SELECT COUNT(*) FROM schema_version WHERE version=9998").fetchone()[0]
== 1
)
conn.close()
def test_columns_helper_validates_table_name():
"""A31 : `columns()` remplace les 24 copies de PRAGMA table_info + valide l'identifiant."""
import sqlite3 as _sqlite3
from app.migrations import columns
conn = _sqlite3.connect(":memory:")
conn.execute("CREATE TABLE t1 (id INTEGER, nom TEXT)")
assert columns(conn, "t1") == {"id", "nom"}
try:
columns(conn, "t1; DROP TABLE users")
raise AssertionError("identifiant non validé")
except ValueError:
pass
conn.close()
def test_no_duplicate_routes():
"""A24 : deux routes même méthode+chemin → l'une écrase silencieusement l'autre."""
from app.main import app