"""Tests for chat database models.""" import pytest from sqlalchemy import create_engine, event, text from sqlalchemy.exc import IntegrityError class TestChatSessionModel: """Tests for the ChatSession database model.""" def test_chat_session_table_exists(self, setup_database_for_all_tests): """Test that the chat_sessions table exists in the database.""" from src.local_deep_research.database.models import Base # Get all table names from metadata table_names = set(Base.metadata.tables.keys()) assert "chat_sessions" in table_names, ( "chat_sessions table is missing from database schema. " "Ensure the ChatSession model is properly defined and imported." ) def test_chat_session_has_required_columns( self, setup_database_for_all_tests ): """Test that ChatSession has all required columns.""" from src.local_deep_research.database.models.chat import ChatSession required_columns = { "id", "title", "status", "accumulated_context", "created_at", "message_count", } actual_columns = set(ChatSession.__table__.columns.keys()) missing = required_columns - actual_columns assert not missing, ( f"ChatSession is missing required columns: {missing}\n" "This will break chat session storage." ) def test_chat_session_json_fields_serialize( self, setup_database_for_all_tests ): """Test that accumulated_context JSON field serializes correctly.""" from datetime import datetime, UTC from src.local_deep_research.database.models.chat import ChatSession SessionLocal = setup_database_for_all_tests session = SessionLocal() try: test_context = { "key_entities": ["test1", "test2"], "topics": ["topic1"], "summary": "", } chat_session = ChatSession( id="test-session-json-1", title="Test Session", status="active", accumulated_context=test_context, created_at=datetime.now(UTC), ) session.add(chat_session) session.commit() retrieved = ( session.query(ChatSession) .filter_by(id="test-session-json-1") .first() ) assert retrieved is not None assert retrieved.accumulated_context == test_context finally: session.rollback() session.close() def test_chat_session_status_default_is_active( self, setup_database_for_all_tests ): """Test that default status is 'active'.""" from datetime import datetime, UTC from src.local_deep_research.database.models.chat import ChatSession SessionLocal = setup_database_for_all_tests session = SessionLocal() try: chat_session = ChatSession( id="test-session-default-1", title="Test Session", created_at=datetime.now(UTC), ) session.add(chat_session) session.commit() retrieved = ( session.query(ChatSession) .filter_by(id="test-session-default-1") .first() ) assert retrieved is not None assert retrieved.status == "active" finally: session.rollback() session.close() class TestChatMessageModel: """Tests for the ChatMessage database model.""" def test_chat_message_table_exists(self, setup_database_for_all_tests): """Test that the chat_messages table exists in the database.""" from src.local_deep_research.database.models import Base table_names = set(Base.metadata.tables.keys()) assert "chat_messages" in table_names, ( "chat_messages table is missing from database schema. " "Ensure the ChatMessage model is properly defined and imported." ) def test_chat_message_has_required_columns( self, setup_database_for_all_tests ): """Test that ChatMessage has all required columns.""" from src.local_deep_research.database.models.chat import ChatMessage required_columns = { "id", "session_id", "role", "content", "message_type", "research_id", "sequence_number", "created_at", } actual_columns = set(ChatMessage.__table__.columns.keys()) missing = required_columns - actual_columns assert not missing, ( f"ChatMessage is missing required columns: {missing}\n" "This will break chat message storage." ) def test_chat_message_foreign_key_to_session( self, setup_database_for_all_tests ): """Test that ChatMessage has foreign key relationship to ChatSession.""" from src.local_deep_research.database.models.chat import ChatMessage # Check foreign keys foreign_keys = [ fk.target_fullname for fk in ChatMessage.__table__.foreign_keys ] assert any("chat_sessions" in fk for fk in foreign_keys), ( "ChatMessage should have a foreign key to chat_sessions table" ) def test_chat_message_research_id_nullable( self, setup_database_for_all_tests ): """Test that research_id column is nullable.""" from datetime import datetime, UTC from src.local_deep_research.database.models.chat import ( ChatMessage, ChatSession, ) SessionLocal = setup_database_for_all_tests session = SessionLocal() try: now = datetime.now(UTC) # Create session first chat_session = ChatSession( id="test-session-nullable-1", title="Test Session", created_at=now, ) session.add(chat_session) session.commit() # Create message without research_id message = ChatMessage( id="test-msg-nullable-1", session_id="test-session-nullable-1", role="user", content="Test question", message_type="query", sequence_number=1, created_at=now, # research_id intentionally omitted ) session.add(message) session.commit() # Retrieve and verify retrieved = ( session.query(ChatMessage) .filter_by(id="test-msg-nullable-1") .first() ) assert retrieved is not None assert retrieved.research_id is None finally: session.rollback() session.close() class TestChatModelCascade: """Tests for cascade delete behavior between ChatSession and ChatMessage.""" def test_delete_session_cascades_to_messages( self, setup_database_for_all_tests ): """Test that deleting a session also deletes its messages.""" from datetime import datetime, UTC from src.local_deep_research.database.models.chat import ( ChatSession, ChatMessage, ) SessionLocal = setup_database_for_all_tests session = SessionLocal() try: now = datetime.now(UTC) # Create session chat_session = ChatSession( id="test-session-cascade-1", title="Test Session", created_at=now, ) session.add(chat_session) session.commit() # Create messages for i in range(3): message = ChatMessage( id=f"test-msg-cascade-{i}", session_id="test-session-cascade-1", role="user" if i % 2 == 0 else "assistant", content=f"Message {i}", message_type="query" if i % 2 == 0 else "response", sequence_number=i + 1, created_at=now, ) session.add(message) session.commit() # Verify messages exist messages_before = ( session.query(ChatMessage) .filter_by(session_id="test-session-cascade-1") .all() ) assert len(messages_before) == 3 # Delete session session.delete(chat_session) session.commit() # Verify messages were also deleted messages_after = ( session.query(ChatMessage) .filter_by(session_id="test-session-cascade-1") .all() ) assert len(messages_after) == 0 finally: session.rollback() session.close() def test_messages_ordered_by_sequence_number( self, setup_database_for_all_tests ): """Test that messages are properly ordered by sequence number.""" from datetime import datetime, UTC from src.local_deep_research.database.models.chat import ( ChatSession, ChatMessage, ) SessionLocal = setup_database_for_all_tests session = SessionLocal() try: now = datetime.now(UTC) # Create session chat_session = ChatSession( id="test-session-order-1", title="Test Session", created_at=now, ) session.add(chat_session) session.commit() # Create messages in random order sequence_numbers = [3, 1, 4, 2, 5] for seq in sequence_numbers: message = ChatMessage( id=f"test-msg-order-{seq}", session_id="test-session-order-1", role="user", content=f"Message {seq}", message_type="query", sequence_number=seq, created_at=now, ) session.add(message) session.commit() # Query with order_by messages = ( session.query(ChatMessage) .filter_by(session_id="test-session-order-1") .order_by(ChatMessage.sequence_number) .all() ) # Verify order for i, msg in enumerate(messages): assert msg.sequence_number == i + 1 finally: session.rollback() session.close() class TestChatSessionStatus: """Tests for chat session status management.""" def test_session_can_be_archived(self, setup_database_for_all_tests): """Test that a session status can be changed to 'archived'.""" from datetime import datetime, UTC from src.local_deep_research.database.models.chat import ChatSession SessionLocal = setup_database_for_all_tests session = SessionLocal() try: chat_session = ChatSession( id="test-session-archive-1", title="Test Session", status="active", created_at=datetime.now(UTC), ) session.add(chat_session) session.commit() # Update status chat_session.status = "archived" session.commit() retrieved = ( session.query(ChatSession) .filter_by(id="test-session-archive-1") .first() ) assert retrieved.status == "archived" finally: session.rollback() session.close() class TestChatMessageContentNotNull: """Schema invariant: chat_messages.content is NOT NULL. Negative tests ensuring the schema-level constraint is in place and the application-layer validator agrees with it. """ def test_db_rejects_null_content(self, setup_database_for_all_tests): """Inserting a chat_messages row with content=NULL must raise IntegrityError at commit time. This pins the migration's ``content NOT NULL`` declaration. """ import pytest from datetime import datetime, UTC from sqlalchemy.exc import IntegrityError from src.local_deep_research.database.models.chat import ( ChatMessage, ChatSession, ) SessionLocal = setup_database_for_all_tests session = SessionLocal() try: # Parent session row parent = ChatSession( id="test-not-null-session", created_at=datetime.now(UTC), ) session.add(parent) session.commit() # Insert with content=None must fail at commit. msg = ChatMessage( id="test-not-null-msg", session_id="test-not-null-session", role="assistant", message_type="response", content=None, sequence_number=1, created_at=datetime.now(UTC), ) session.add(msg) with pytest.raises(IntegrityError): session.commit() finally: session.rollback() session.close() def test_service_rejects_none_content_via_value_error(self): """ChatService.add_message validates content!=None before any DB write, raising ValueError so the route layer can surface a 400.""" import pytest from src.local_deep_research.chat.service import ChatService service = ChatService(username="not-real-for-validator-test") with pytest.raises(ValueError, match="content is required"): service.add_message( session_id="any", role="assistant", content=None, message_type="response", research_id="any", ) # --------------------------------------------------------------------------- # DB-level cascade / FK / uniqueness tests # # The shared `setup_database_for_all_tests` fixture in tests/conftest.py does # NOT enable PRAGMA foreign_keys=ON — SQLite defaults FKs OFF, so DB-level # `ondelete` rules silently no-op there. The tests below need real FK # enforcement, so they use a private `fk_enforced_engine` fixture that mirrors # the pattern at tests/database/test_research_strategy_fk_regression.py. # --------------------------------------------------------------------------- @pytest.fixture def fk_enforced_engine(tmp_path): """Fresh-install SQLite engine with FK enforcement on for every connection.""" from src.local_deep_research.database.models import Base db_path = tmp_path / "fk_test.db" engine = create_engine(f"sqlite:///{db_path}") @event.listens_for(engine, "connect") def _enable_fk(dbapi_connection, _): dbapi_connection.execute("PRAGMA foreign_keys = ON") Base.metadata.create_all(engine) yield engine engine.dispose() def _seed_session(conn, sid="s1"): conn.execute( text( "INSERT INTO chat_sessions " "(id, status, message_count, created_at) " "VALUES (:id, 'active', 0, '2026-01-01T00:00:00')" ), {"id": sid}, ) def _seed_research(conn, rid="r1"): conn.execute( text( "INSERT INTO research_history " "(id, query, mode, status, created_at) " "VALUES (:id, 'q', 'quick', 'completed', " "'2026-01-01T00:00:00')" ), {"id": rid}, ) def _insert_chat_message( conn, *, mid, sid, rid=None, seq=1, role="user", mtype="query" ): conn.execute( text( "INSERT INTO chat_messages " "(id, session_id, research_id, role, message_type, " " content, sequence_number, created_at) " "VALUES (:mid, :sid, :rid, :role, :mtype, 'x', :seq, " "'2026-01-01T00:00:00')" ), { "mid": mid, "sid": sid, "rid": rid, "role": role, "mtype": mtype, "seq": seq, }, ) def _insert_progress_step(conn, *, pid, rid, sid, seq=1): conn.execute( text( "INSERT INTO chat_progress_steps " "(id, research_id, session_id, content, " " sequence_number, created_at) " "VALUES (:pid, :rid, :sid, 'step content', :seq, " "'2026-01-01T00:00:00')" ), {"pid": pid, "rid": rid, "sid": sid, "seq": seq}, ) class TestChatModelDBLevelCascade: """Cascade rules enforced by the database (FK pragma on, raw SQL DELETE).""" def test_research_delete_sets_chat_messages_research_id_to_null( self, fk_enforced_engine ): with fk_enforced_engine.begin() as conn: _seed_session(conn, sid="s1") _seed_research(conn, rid="r1") _insert_chat_message(conn, mid="m1", sid="s1", rid="r1") with fk_enforced_engine.begin() as conn: conn.execute(text("DELETE FROM research_history WHERE id='r1'")) with fk_enforced_engine.connect() as conn: row = conn.execute( text("SELECT research_id FROM chat_messages WHERE id='m1'") ).first() assert row is not None assert row.research_id is None def test_session_delete_cascades_chat_progress_steps( self, fk_enforced_engine ): with fk_enforced_engine.begin() as conn: _seed_session(conn, sid="s1") _seed_research(conn, rid="r1") _insert_progress_step(conn, pid="p1", rid="r1", sid="s1") with fk_enforced_engine.begin() as conn: conn.execute(text("DELETE FROM chat_sessions WHERE id='s1'")) with fk_enforced_engine.connect() as conn: count = conn.execute( text("SELECT COUNT(*) FROM chat_progress_steps") ).scalar() assert count == 0 def test_session_delete_cascades_chat_messages(self, fk_enforced_engine): """Migration 0009 declares ondelete=CASCADE on chat_messages.session_id. The ORM-level cascade is exercised by TestChatModelCascade::test_delete_session_cascades_to_messages, but until this test was added the DB-level CASCADE behaviour was only structurally defined, not verified — a migration regression that silently dropped the ondelete clause would not have failed any test. Drive the DELETE through raw SQL with the FK pragma on so we exercise the migration's CASCADE, not SQLAlchemy's. """ with fk_enforced_engine.begin() as conn: _seed_session(conn, sid="s1") _insert_chat_message(conn, mid="m1", sid="s1", seq=1) _insert_chat_message(conn, mid="m2", sid="s1", seq=2) with fk_enforced_engine.begin() as conn: conn.execute(text("DELETE FROM chat_sessions WHERE id='s1'")) with fk_enforced_engine.connect() as conn: count = conn.execute( text("SELECT COUNT(*) FROM chat_messages") ).scalar() assert count == 0 def test_session_delete_sets_research_history_chat_session_id_null( self, fk_enforced_engine ): """Migration 0009 declares ondelete=SET NULL on research_history.chat_session_id so research artefacts survive a chat-session delete (just unlinked). Until this test was added, this behaviour was only structurally defined, not verified at the DB level — only the symmetric chat_messages.research_id SET NULL was covered. Seed a research row with a chat_session_id, delete the session, and assert the FK column was nulled rather than the row being cascade-deleted. """ with fk_enforced_engine.begin() as conn: _seed_session(conn, sid="s1") # Seed a research row with chat_session_id pointing at s1. # The standard _seed_research helper does not set the FK, # so do it inline here. conn.execute( text( "INSERT INTO research_history " "(id, query, mode, status, created_at, " " chat_session_id) " "VALUES (:id, 'q', 'quick', 'completed', " "'2026-01-01T00:00:00', :sid)" ), {"id": "r1", "sid": "s1"}, ) with fk_enforced_engine.begin() as conn: conn.execute(text("DELETE FROM chat_sessions WHERE id='s1'")) with fk_enforced_engine.connect() as conn: row = conn.execute( text( "SELECT id, chat_session_id " "FROM research_history WHERE id='r1'" ) ).first() # Row must still exist — SET NULL preserves research artefacts. assert row is not None assert row.chat_session_id is None class TestChatModelDBLevelFKEnforcement: """Negative tests against DB-level FK + unique constraints.""" def test_chat_messages_fk_session_id_enforced(self, fk_enforced_engine): # FK fires immediately at execute() under PRAGMA foreign_keys=ON, so # pytest.raises must wrap the engine.begin() block (otherwise its # __exit__ tries to commit an aborted DBAPI transaction). with pytest.raises(IntegrityError): with fk_enforced_engine.begin() as conn: _insert_chat_message(conn, mid="m1", sid="nonexistent-session") def test_chat_messages_unique_session_seq(self, fk_enforced_engine): with fk_enforced_engine.begin() as conn: _seed_session(conn, sid="s1") _insert_chat_message(conn, mid="m1", sid="s1", seq=1) with pytest.raises(IntegrityError): with fk_enforced_engine.begin() as conn: _insert_chat_message(conn, mid="m2", sid="s1", seq=1) # The chat Enum columns deliberately omit a DB-level CHECK constraint # (matching the migration). Enum-value enforcement is # at the ORM/service layer (ChatRole(value) raises ValueError before # any INSERT). The tests below pin THAT contract — both the fact that # the raw INSERT succeeds (no CHECK) and that the service-layer guard # catches invalid values. def test_chat_messages_role_no_db_check_constraint( self, fk_enforced_engine ): """Raw INSERT of an invalid `role` succeeds because the model has no `create_constraint=True` — same as the migration. Schema must agree between create_all (fresh installs) and the upgrade path.""" with fk_enforced_engine.begin() as conn: _seed_session(conn, sid="s1") with fk_enforced_engine.begin() as conn: conn.execute( text( "INSERT INTO chat_messages " "(id, session_id, role, message_type, content, " " sequence_number, created_at) " "VALUES ('m1', 's1', 'BOGUS', 'query', 'x', 1, " "'2026-01-01T00:00:00')" ) ) def test_chat_messages_message_type_no_db_check_constraint( self, fk_enforced_engine ): """Same as above for `message_type`.""" with fk_enforced_engine.begin() as conn: _seed_session(conn, sid="s1") with fk_enforced_engine.begin() as conn: conn.execute( text( "INSERT INTO chat_messages " "(id, session_id, role, message_type, content, " " sequence_number, created_at) " "VALUES ('m1', 's1', 'user', 'BOGUS', 'x', 1, " "'2026-01-01T00:00:00')" ) ) def test_chat_sessions_status_no_db_check_constraint( self, fk_enforced_engine ): """Same as above for `status`.""" with fk_enforced_engine.begin() as conn: conn.execute( text( "INSERT INTO chat_sessions " "(id, status, message_count, created_at) " "VALUES ('s1', 'BOGUS', 0, '2026-01-01T00:00:00')" ) )