learningcircuit--local-deep-research/tests/database/test_chat_models.py

"""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')"
                )
            )