From 6090b98ab0a7e01d76476574f7244a72c4d17db6 Mon Sep 17 00:00:00 2001
From: Tianyu Liu <liu.tianyu93@hotmail.com>
Date: Mon, 22 Jun 2026 20:45:47 +0200
Subject: [PATCH] M5: enable SQLite foreign_keys pragma for DB-level FK
 enforcement

---
 app/db.py                   |  1 +
 tests/test_modbus_models.py | 59 +++++++++++++++++++++++++++++++++++++
 2 files changed, 60 insertions(+)

diff --git a/app/db.py b/app/db.py
index ed8068d..de4c9aa 100644
--- a/app/db.py
+++ b/app/db.py
@@ -28,6 +28,7 @@ def _get_engine(database_url: str) -> Engine:
         def _enable_sqlite_wal(dbapi_connection, _connection_record):
             cursor = dbapi_connection.cursor()
             cursor.execute("PRAGMA journal_mode=WAL")
+            cursor.execute("PRAGMA foreign_keys=ON")
             cursor.close()
 
     return engine
diff --git a/tests/test_modbus_models.py b/tests/test_modbus_models.py
index 9674dfd..0daa091 100644
--- a/tests/test_modbus_models.py
+++ b/tests/test_modbus_models.py
@@ -403,3 +403,62 @@ def test_modbus_reading_restrict_prevents_device_deletion(tmp_path: Path):
             session.commit()
 
     engine.dispose()
+
+
+def test_modbus_reading_restrict_enforced_via_app_engine(tmp_path: Path):
+    """FK RESTRICT is enforced by the *default* app engine (db.py) without any manual pragma.
+
+    After enabling ``PRAGMA foreign_keys=ON`` in the db.py connect event, the app
+    engine must enforce ON DELETE RESTRICT on modbus_reading.device_id without
+    callers having to set the pragma themselves.  This test uses the real app engine
+    obtained via ``app.db.get_engine`` (routed through the lru_cache, just like the
+    running app) so it exercises the exact production code path.
+    """
+    import sqlalchemy.exc
+
+    from app.db import _get_engine, reset_db_caches
+
+    db_path = tmp_path / "app_engine_restrict_test.db"
+    db_url = f"sqlite:///{db_path}"
+
+    alembic_cfg = _make_app_alembic_config(db_url)
+    command.upgrade(alembic_cfg, "head")
+
+    # Obtain engine through the same lru_cache path that production uses.
+    # Clear caches first so we get a fresh engine pointing at our tmp DB.
+    reset_db_caches()
+    engine = _get_engine(db_url)
+
+    try:
+        now = datetime.now(tz=timezone.utc)
+        with Session(engine) as session:
+            device = ModbusDevice(
+                friendly_name="App-Engine RESTRICT Test",
+                host="10.0.0.2",
+                profile="sdm120",
+                created_at=now,
+                updated_at=now,
+            )
+            session.add(device)
+            session.flush()
+
+            reading = ModbusReading(
+                device_id=device.id,
+                recorded_at=now,
+                payload={"voltage": 230.0},
+            )
+            session.add(reading)
+            session.commit()
+            device_id = device.id
+
+        # Without manual pragma, the app engine (db.py) must still enforce RESTRICT.
+        with pytest.raises(sqlalchemy.exc.IntegrityError):
+            with Session(engine) as session:
+                session.execute(
+                    text("DELETE FROM modbus_device WHERE id = :did"),
+                    {"did": device_id},
+                )
+                session.commit()
+    finally:
+        reset_db_caches()
+        engine.dispose()