Files
odysseus/tests/test_sqlite_foreign_keys.py

43 lines
1.2 KiB
Python
Raw Normal View History

DB: enable SQLite foreign key cascades * fix(db): enable SQLite foreign keys so ondelete cascades actually fire core/database.py declares DB-level FK actions throughout (ondelete="CASCADE" / "SET NULL"), but SQLite disables foreign-key enforcement per connection by default and the engine had no connect-event listener turning it on. So every one of those ondelete actions was dead. Concrete impact: cleanup_old_sessions() in src/cleanup_service.py removes old sessions with a bulk `query(Session).delete()`, which bypasses the ORM-level relationship cascade and relies solely on the DB-level ondelete="CASCADE" on ChatMessage.session_id. With foreign keys off, the messages are never deleted — they pile up as orphaned rows on every cleanup cycle. Add the standard SQLAlchemy connect listener issuing `PRAGMA foreign_keys=ON`, guarded by `isinstance(conn, sqlite3.Connection)` so it only affects SQLite and leaves other backends untouched. tests/test_sqlite_foreign_keys.py inserts a Session + ChatMessage, deletes the Session via bulk `query().delete()`, and asserts the ChatMessage is cascade-deleted. Fails before this change (orphan remains). * docs(db): clarify FK pragma scope per review; trim test comments Address review feedback on the foreign_keys PRAGMA change: - Note that the class-level connect listener fires for every Engine in the process and is a no-op on non-SQLite backends (isinstance guard). - Warn near init_db() that FK enforcement is now global, so a migration that temporarily violates FK constraints must disable foreign_keys around that work. - Drop the step-by-step narration comments from the regression test. No behavior change.
2026-06-02 18:36:13 +07:00
import pytest
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
2026-06-04 01:27:29 +02:00
from tests.helpers.import_state import clear_fake_database_modules
2026-06-04 01:27:29 +02:00
clear_fake_database_modules()
2026-06-04 01:27:29 +02:00
DB: enable SQLite foreign key cascades * fix(db): enable SQLite foreign keys so ondelete cascades actually fire core/database.py declares DB-level FK actions throughout (ondelete="CASCADE" / "SET NULL"), but SQLite disables foreign-key enforcement per connection by default and the engine had no connect-event listener turning it on. So every one of those ondelete actions was dead. Concrete impact: cleanup_old_sessions() in src/cleanup_service.py removes old sessions with a bulk `query(Session).delete()`, which bypasses the ORM-level relationship cascade and relies solely on the DB-level ondelete="CASCADE" on ChatMessage.session_id. With foreign keys off, the messages are never deleted — they pile up as orphaned rows on every cleanup cycle. Add the standard SQLAlchemy connect listener issuing `PRAGMA foreign_keys=ON`, guarded by `isinstance(conn, sqlite3.Connection)` so it only affects SQLite and leaves other backends untouched. tests/test_sqlite_foreign_keys.py inserts a Session + ChatMessage, deletes the Session via bulk `query().delete()`, and asserts the ChatMessage is cascade-deleted. Fails before this change (orphan remains). * docs(db): clarify FK pragma scope per review; trim test comments Address review feedback on the foreign_keys PRAGMA change: - Note that the class-level connect listener fires for every Engine in the process and is a no-op on non-SQLite backends (isinstance guard). - Warn near init_db() that FK enforcement is now global, so a migration that temporarily violates FK constraints must disable foreign_keys around that work. - Drop the step-by-step narration comments from the regression test. No behavior change.
2026-06-02 18:36:13 +07:00
from core.database import Base, Session, ChatMessage
from datetime import datetime
def test_sqlite_foreign_keys_cascade():
engine = create_engine("sqlite:///:memory:", connect_args={"check_same_thread": False})
Base.metadata.create_all(bind=engine)
TestSessionLocal = sessionmaker(bind=engine)
db = TestSessionLocal()
session_id = "test-session-123"
s = Session(
id=session_id,
name="Test Session",
endpoint_url="http://localhost:8000",
model="gpt-4",
created_at=datetime.utcnow(),
updated_at=datetime.utcnow()
)
m = ChatMessage(id="test-msg-123", session_id=session_id, role="user", content="test message")
db.add(s)
db.add(m)
db.commit()
assert db.query(Session).count() == 1
assert db.query(ChatMessage).count() == 1
db.query(Session).filter(Session.id == session_id).delete()
db.commit()
assert db.query(ChatMessage).count() == 0
db.close()