Bertrand Benjamin
9f5f46a93a
Les comptes rendus n'identifient un immeuble que par son code de gestion (« 33689020 »), illisible partout où il s'affiche. La dénomination (« Servient ») le remplace à l'écran sans toucher au code, qui reste la clé venue des PDF. La colonne s'ajoute à une table déjà installée : elle passe donc par le rattrapage de schéma, qui accepte désormais l'absence de valeur de rattrapage. Déduire un nom d'usage du code en inventerait un. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
103 lines
3.7 KiB
Python
103 lines
3.7 KiB
Python
"""Rattrapage de schema sur une base creee avant l'ajout d'une colonne.
|
|
|
|
Le projet n'a pas d'outil de migration : `create_all` laisse intactes les tables
|
|
deja presentes. Sans le rattrapage d'`init_db`, une base installee cesserait de
|
|
fonctionner des qu'une colonne est ajoutee au modele.
|
|
"""
|
|
|
|
from datetime import datetime, timezone
|
|
|
|
from sqlalchemy import text
|
|
|
|
from plesna_gerance.database import connection
|
|
|
|
|
|
def test_ajoute_extracted_at_a_une_base_existante(tmp_path, monkeypatch):
|
|
db_path = tmp_path / "ancienne.sqlite"
|
|
monkeypatch.setenv("PLESNA_DB_PATH", str(db_path))
|
|
monkeypatch.setenv("PLESNA_STORAGE_PATH", str(tmp_path / "documents"))
|
|
|
|
# Base telle qu'elle existait avant le champ, avec un document dedans.
|
|
connection.reset_connection()
|
|
connection.init_db(db_path)
|
|
engine = connection.get_engine(db_path)
|
|
importe_le = datetime(2026, 3, 1, 10, 0, tzinfo=timezone.utc)
|
|
with engine.begin() as conn:
|
|
conn.execute(text("ALTER TABLE documents DROP COLUMN extracted_at"))
|
|
conn.execute(text("INSERT INTO immeubles (code) VALUES ('IMM1');"))
|
|
conn.execute(
|
|
text(
|
|
"INSERT INTO documents (reference, date, immeuble_id, json_data,"
|
|
" created_at) VALUES ('REF', '2026-03-01', 1, '{}', :created)"
|
|
),
|
|
{"created": importe_le},
|
|
)
|
|
connection.reset_connection()
|
|
|
|
connection.init_db(db_path)
|
|
|
|
with connection.get_engine(db_path).begin() as conn:
|
|
colonnes = {
|
|
row[1] for row in conn.execute(text("PRAGMA table_info(documents)"))
|
|
}
|
|
assert "extracted_at" in colonnes
|
|
|
|
# Un document deja en base a ete extrait lors de son import.
|
|
extracted_at, created_at = conn.execute(
|
|
text("SELECT extracted_at, created_at FROM documents")
|
|
).one()
|
|
assert extracted_at == created_at
|
|
|
|
connection.reset_connection()
|
|
|
|
|
|
def test_ajoute_denomination_a_une_base_existante(tmp_path, monkeypatch):
|
|
"""La colonne arrive vide : deduire un nom d'usage du code l'inventerait."""
|
|
db_path = tmp_path / "ancienne.sqlite"
|
|
monkeypatch.setenv("PLESNA_DB_PATH", str(db_path))
|
|
monkeypatch.setenv("PLESNA_STORAGE_PATH", str(tmp_path / "documents"))
|
|
|
|
connection.reset_connection()
|
|
connection.init_db(db_path)
|
|
engine = connection.get_engine(db_path)
|
|
with engine.begin() as conn:
|
|
conn.execute(text("ALTER TABLE immeubles DROP COLUMN denomination"))
|
|
conn.execute(text("INSERT INTO immeubles (code) VALUES ('33689020')"))
|
|
connection.reset_connection()
|
|
|
|
connection.init_db(db_path)
|
|
|
|
with connection.get_engine(db_path).begin() as conn:
|
|
colonnes = {
|
|
row[1] for row in conn.execute(text("PRAGMA table_info(immeubles)"))
|
|
}
|
|
assert "denomination" in colonnes
|
|
|
|
denomination, code = conn.execute(
|
|
text("SELECT denomination, code FROM immeubles")
|
|
).one()
|
|
assert denomination is None
|
|
assert code == "33689020"
|
|
|
|
connection.reset_connection()
|
|
|
|
|
|
def test_rattrapage_idempotent(tmp_path, monkeypatch):
|
|
"""Relancer init_db sur une base a jour ne doit rien casser."""
|
|
db_path = tmp_path / "a_jour.sqlite"
|
|
monkeypatch.setenv("PLESNA_DB_PATH", str(db_path))
|
|
monkeypatch.setenv("PLESNA_STORAGE_PATH", str(tmp_path / "documents"))
|
|
|
|
connection.reset_connection()
|
|
connection.init_db(db_path)
|
|
connection.reset_connection()
|
|
connection.init_db(db_path)
|
|
|
|
with connection.get_engine(db_path).begin() as conn:
|
|
colonnes = [
|
|
row[1] for row in conn.execute(text("PRAGMA table_info(documents)"))
|
|
]
|
|
assert colonnes.count("extracted_at") == 1
|
|
|
|
connection.reset_connection()
|