From cfd0da34cd39b786a60d4ddb70eeaef154ffbb02 Mon Sep 17 00:00:00 2001
From: Colin Maudry <colin@maudry.com>
Date: Wed, 15 Apr 2026 16:27:41 +0200
Subject: [PATCH] =?UTF-8?q?test(db):=20s=C3=A9rialisation=20des=20builds?=
 =?UTF-8?q?=20concurrents=20par=20fcntl.flock=20(#71)?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
---
 tests/test_db.py | 61 ++++++++++++++++++++++++++++++++++++++++++++++++
 1 file changed, 61 insertions(+)

diff --git a/tests/test_db.py b/tests/test_db.py
index d99ded0..6ce740a 100644
--- a/tests/test_db.py
+++ b/tests/test_db.py
@@ -192,3 +192,64 @@ def test_query_marches_returns_polars_frame(built_db, monkeypatch):
     assert isinstance(frame, pl.DataFrame)
     assert frame.height == 2
     assert set(frame["uid"].to_list()) == {"1", "2"}
+
+
+def test_concurrent_build_serialized(tmp_path):
+    """Multiple threads calling _ensure_database must serialize via flock.
+
+    Only one should actually build; others wait, see the fresh DB, and skip.
+    No tmp file should leak. No exceptions should occur.
+    """
+    import fcntl
+    import threading
+
+    import src.db as db
+
+    # Set up source parquet
+    parquet_path = tmp_path / "src.parquet"
+    df = pl.DataFrame(
+        {
+            "uid": ["A"],
+            "donneesActuelles": [True],
+            "dateNotification": ["2024-01-01"],
+            "objet": ["Test"],
+            "acheteur_id": ["a1"],
+            "acheteur_nom": ["A1"],
+            "titulaire_id": ["t1"],
+            "titulaire_nom": ["T1"],
+            "acheteur_departement_code": ["75"],
+            "titulaire_departement_code": ["75"],
+            "montant": [1000.0],
+            "dureeMois": [12],
+        }
+    )
+    df.write_parquet(parquet_path)
+
+    db_path = tmp_path / "decp.duckdb"
+    lock_path = db_path.with_suffix(".duckdb.lock")
+    tmp_path_artifact = db_path.with_suffix(".duckdb.tmp")
+
+    errors: list[BaseException] = []
+
+    def worker():
+        try:
+            # Mirror the locking logic in _ensure_database
+            with open(lock_path, "w") as lf:
+                fcntl.flock(lf.fileno(), fcntl.LOCK_EX)
+                try:
+                    if db.should_rebuild(db_path, parquet_path):
+                        db.build_database(db_path, parquet_path)
+                finally:
+                    fcntl.flock(lf.fileno(), fcntl.LOCK_UN)
+        except BaseException as exc:
+            errors.append(exc)
+
+    threads = [threading.Thread(target=worker) for _ in range(3)]
+    for t in threads:
+        t.start()
+    for t in threads:
+        t.join()
+
+    assert errors == []
+    assert db_path.exists()
+    assert not tmp_path_artifact.exists()