Bertrand Benjamin
92cf21309f
La liste des documents montrait la date d'import, figee au premier enregistrement : c'est la date de la derniere extraction ayant produit les donnees affichees qui renseigne, surtout apres un balayage de re-extraction. La colonne `extracted_at` la porte, mise a jour a chaque enregistrement, et la date d'import reste consultable en infobulle. Le projet n'ayant pas d'outil de migration (`create_all` laisse intactes les tables existantes), `init_db` rattrape les colonnes ajoutees apres coup : sans cela une base deja installee cesserait de fonctionner. Les documents deja presents recoivent leur date d'import, qui est bien celle de leur extraction. Une fois la selection appliquee sans echec, le balayage renvoie vers la liste des documents ; en cas d'echec on reste sur place, les lignes concernees portant leur message. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
74 lines
2.6 KiB
Python
74 lines
2.6 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_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()
|