learningcircuit--local-deep-research/tests/web/models/test_database.py

"""
Tests for web/models/database.py

Tests cover:
- get_db_connection (deprecated)
- calculate_duration function
- get_logs_for_research function
- get_total_logs_for_research function
"""

import pytest
from unittest.mock import Mock, patch
from datetime import datetime, timezone, timedelta


class TestGetDbConnection:
    """Tests for deprecated get_db_connection function."""

    def test_get_db_connection_raises_runtime_error(self):
        """Test that get_db_connection raises RuntimeError."""
        from local_deep_research.web.models.database import get_db_connection

        with pytest.raises(RuntimeError) as exc_info:
            get_db_connection()

        assert "Shared database access is deprecated" in str(exc_info.value)


class TestCalculateDuration:
    """Tests for calculate_duration function."""

    def test_calculate_duration_iso_format(self):
        """Test duration calculation with ISO format timestamps."""
        from local_deep_research.web.models.database import calculate_duration

        created_at = "2025-01-01T10:00:00+00:00"
        completed_at = "2025-01-01T10:05:00+00:00"

        result = calculate_duration(created_at, completed_at)

        assert result == 300  # 5 minutes = 300 seconds

    def test_calculate_duration_with_microseconds(self):
        """Test duration calculation with microseconds in ISO format."""
        from local_deep_research.web.models.database import calculate_duration

        # Use ISO format with timezone for proper parsing
        created_at = "2025-01-01T10:00:00.123456+00:00"
        completed_at = "2025-01-01T10:01:00.654321+00:00"

        result = calculate_duration(created_at, completed_at)

        # Should be approximately 60 seconds
        assert result == 60 or result == 61

    def test_calculate_duration_without_completed_at(self):
        """Test duration calculation without completed_at uses current time."""
        from local_deep_research.web.models.database import calculate_duration

        # Create a recent timestamp
        now = datetime.now(timezone.utc)
        created_at = (now - timedelta(seconds=120)).isoformat()

        result = calculate_duration(created_at)

        # Should be approximately 120 seconds (with some tolerance)
        assert result is not None
        assert 115 <= result <= 130

    def test_calculate_duration_none_created_at(self):
        """Test duration calculation with None created_at."""
        from local_deep_research.web.models.database import calculate_duration

        result = calculate_duration(None)

        assert result is None

    def test_calculate_duration_empty_created_at(self):
        """Test duration calculation with empty created_at."""
        from local_deep_research.web.models.database import calculate_duration

        result = calculate_duration("")

        assert result is None

    def test_calculate_duration_space_separated_format(self):
        """Test duration with space-separated timestamp format with timezone."""
        from local_deep_research.web.models.database import calculate_duration

        # Use format with timezone suffix for proper handling
        created_at = "2025-01-01 10:00:00+00:00"
        completed_at = "2025-01-01 10:10:00+00:00"

        result = calculate_duration(created_at, completed_at)

        assert result == 600  # 10 minutes = 600 seconds

    def test_calculate_duration_replaces_space_with_t(self):
        """Test that space is replaced with T for parsing."""
        from local_deep_research.web.models.database import calculate_duration

        # This format should fall back to replacing space with T
        created_at = "2025-01-01 10:00:00+00:00"
        completed_at = "2025-01-01 10:02:00+00:00"

        result = calculate_duration(created_at, completed_at)

        assert result == 120  # 2 minutes


class TestGetLogsForResearch:
    """Tests for get_logs_for_research function."""

    def test_get_logs_success(self):
        """Test successful log retrieval."""
        mock_session = Mock()
        mock_log = Mock()
        mock_log.timestamp = "2025-01-01T10:00:00"
        mock_log.message = "Test log message"
        mock_log.level = "INFO"
        mock_log.module = "test_module"
        mock_log.line_no = 42

        mock_query = Mock()
        mock_query.filter.return_value = mock_query
        mock_query.order_by.return_value = mock_query
        mock_query.all.return_value = [mock_log]
        mock_session.query.return_value = mock_query

        with patch(
            "local_deep_research.web.models.database.get_user_db_session"
        ) as mock_get_session:
            mock_get_session.return_value.__enter__ = Mock(
                return_value=mock_session
            )
            mock_get_session.return_value.__exit__ = Mock(return_value=False)

            from local_deep_research.web.models.database import (
                get_logs_for_research,
            )

            result = get_logs_for_research("test-research-id")

            assert len(result) == 1
            assert result[0]["message"] == "Test log message"
            assert result[0]["type"] == "INFO"
            assert result[0]["module"] == "test_module"
            assert result[0]["line_no"] == 42

    def test_get_logs_empty(self):
        """Test log retrieval when no logs exist."""
        mock_session = Mock()
        mock_query = Mock()
        mock_query.filter.return_value = mock_query
        mock_query.order_by.return_value = mock_query
        mock_query.all.return_value = []
        mock_session.query.return_value = mock_query

        with patch(
            "local_deep_research.web.models.database.get_user_db_session"
        ) as mock_get_session:
            mock_get_session.return_value.__enter__ = Mock(
                return_value=mock_session
            )
            mock_get_session.return_value.__exit__ = Mock(return_value=False)

            from local_deep_research.web.models.database import (
                get_logs_for_research,
            )

            result = get_logs_for_research("nonexistent-id")

            assert result == []

    def test_get_logs_error_returns_empty(self):
        """Test that errors return empty list."""
        with patch(
            "local_deep_research.web.models.database.get_user_db_session"
        ) as mock_get_session:
            mock_get_session.return_value.__enter__ = Mock(
                side_effect=Exception("Database error")
            )

            from local_deep_research.web.models.database import (
                get_logs_for_research,
            )

            result = get_logs_for_research("test-id")

            assert result == []

    def test_get_logs_with_limit_uses_desc_order_and_reverses(self):
        """When limit is set, the query takes the newest N rows (desc) and
        the result is reversed so the caller still sees oldest-first.
        """
        # Three rows ordered newest-first as the DB would return them.
        rows = []
        for i, msg in enumerate(["newest", "middle", "oldest"]):
            row = Mock()
            row.timestamp = f"2025-01-01T10:0{2 - i}:00"
            row.message = msg
            row.level = "INFO"
            row.module = "m"
            row.line_no = i
            rows.append(row)

        mock_session = Mock()
        mock_query = Mock()
        mock_query.filter.return_value = mock_query
        mock_query.order_by.return_value = mock_query
        mock_query.limit.return_value = mock_query
        mock_query.all.return_value = rows
        mock_session.query.return_value = mock_query

        with patch(
            "local_deep_research.web.models.database.get_user_db_session"
        ) as mock_get_session:
            mock_get_session.return_value.__enter__ = Mock(
                return_value=mock_session
            )
            mock_get_session.return_value.__exit__ = Mock(return_value=False)

            from local_deep_research.web.models.database import (
                get_logs_for_research,
            )

            result = get_logs_for_research("rid", limit=3)

        # limit() was called with the requested cap.
        mock_query.limit.assert_called_once_with(3)
        # Caller sees oldest-first ordering.
        assert [r["message"] for r in result] == [
            "oldest",
            "middle",
            "newest",
        ]

    def test_get_logs_with_limit_returns_newest_n_in_ascending_order_real_sqlite(
        self,
    ):
        """End-to-end SQL check against a real in-memory SQLite session.

        The mock-based ordering test above only proves Python's ``reversed()``
        is called — the mock returns a fixed list regardless of which
        ``order_by``/``limit`` was invoked. This test inserts 10 actual rows,
        runs the function through SQLAlchemy, and asserts that the SQL
        produces newest-first DESC + LIMIT and the caller sees the 3 newest
        rows in oldest-first order. Catches any regression where ``desc()``
        is dropped, ``limit()`` is moved, or ``reversed()`` is removed.
        """
        from datetime import datetime, timedelta, timezone

        from sqlalchemy import create_engine
        from sqlalchemy.orm import sessionmaker

        from local_deep_research.database.models import Base, ResearchLog
        from local_deep_research.web.models.database import (
            get_logs_for_research,
        )

        engine = create_engine("sqlite:///:memory:")
        Base.metadata.create_all(engine)
        Session = sessionmaker(bind=engine)
        session = Session()

        # 10 orphan log rows at 1-minute intervals. No parent ResearchHistory
        # row is created — research_id is nullable and SQLite does not enforce
        # FK constraints without PRAGMA foreign_keys=ON, so the insert is
        # valid and isolates this test from the parent-table schema.
        base_time = datetime(2025, 1, 1, 0, 0, 0, tzinfo=timezone.utc)
        for i in range(10):
            session.add(
                ResearchLog(
                    research_id="test-rid",
                    timestamp=base_time + timedelta(minutes=i),
                    message=f"Log {i}",
                    module="test",
                    function="test",
                    line_no=i,
                    level="INFO",
                )
            )
        session.commit()

        try:
            with patch(
                "local_deep_research.web.models.database.get_user_db_session"
            ) as mock_get_session:
                mock_get_session.return_value.__enter__ = Mock(
                    return_value=session
                )
                mock_get_session.return_value.__exit__ = Mock(
                    return_value=False
                )

                result = get_logs_for_research("test-rid", limit=3)

            assert [r["message"] for r in result] == [
                "Log 7",
                "Log 8",
                "Log 9",
            ]
        finally:
            session.close()
            engine.dispose()

    def test_get_logs_with_limit_tie_breaks_equal_timestamps_by_id(self):
        """When rows share an identical timestamp, ``id`` is the tie-break so
        the newest ``limit`` selection is deterministic.

        ``timestamp`` is not unique, so ``order_by(timestamp.desc()).limit(N)``
        alone leaves the rows surviving the boundary SQL-undefined. All 10 rows
        here share one timestamp; Log i has id i+1, so the newest 3 by id are
        Log 7/8/9, returned oldest-first.
        """
        from datetime import datetime, timezone

        from sqlalchemy import create_engine
        from sqlalchemy.orm import sessionmaker

        from local_deep_research.database.models import Base, ResearchLog
        from local_deep_research.web.models.database import (
            get_logs_for_research,
        )

        engine = create_engine("sqlite:///:memory:")
        Base.metadata.create_all(engine)
        session = sessionmaker(bind=engine)()

        shared_time = datetime(2025, 1, 1, 0, 0, 0, tzinfo=timezone.utc)
        for i in range(10):
            session.add(
                ResearchLog(
                    research_id="test-rid",
                    timestamp=shared_time,
                    message=f"Log {i}",
                    module="test",
                    function="test",
                    line_no=i,
                    level="INFO",
                )
            )
        session.commit()

        try:
            with patch(
                "local_deep_research.web.models.database.get_user_db_session"
            ) as mock_get_session:
                mock_get_session.return_value.__enter__ = Mock(
                    return_value=session
                )
                mock_get_session.return_value.__exit__ = Mock(
                    return_value=False
                )

                result = get_logs_for_research("test-rid", limit=3)

            assert [r["message"] for r in result] == [
                "Log 7",
                "Log 8",
                "Log 9",
            ]
        finally:
            session.close()
            engine.dispose()

    def test_get_logs_without_limit_does_not_call_limit(self):
        """When limit is None, the function preserves legacy behaviour and
        does not invoke .limit() on the query (returns every row, asc)."""
        mock_session = Mock()
        mock_query = Mock()
        mock_query.filter.return_value = mock_query
        mock_query.order_by.return_value = mock_query
        mock_query.all.return_value = []
        # Track whether limit() is invoked — it should NOT be.
        mock_query.limit = Mock(side_effect=AssertionError("limit() called"))
        mock_session.query.return_value = mock_query

        with patch(
            "local_deep_research.web.models.database.get_user_db_session"
        ) as mock_get_session:
            mock_get_session.return_value.__enter__ = Mock(
                return_value=mock_session
            )
            mock_get_session.return_value.__exit__ = Mock(return_value=False)

            from local_deep_research.web.models.database import (
                get_logs_for_research,
            )

            # Should complete without triggering the AssertionError side-effect.
            result = get_logs_for_research("rid")
            assert result == []


class TestGetTotalLogsForResearch:
    """Tests for get_total_logs_for_research function."""

    def test_get_total_logs_success(self):
        """Test successful total log count."""
        mock_session = Mock()
        mock_query = Mock()
        mock_query.filter.return_value = mock_query
        mock_query.count.return_value = 42
        mock_session.query.return_value = mock_query

        with patch(
            "local_deep_research.web.models.database.get_user_db_session"
        ) as mock_get_session:
            mock_get_session.return_value.__enter__ = Mock(
                return_value=mock_session
            )
            mock_get_session.return_value.__exit__ = Mock(return_value=False)

            from local_deep_research.web.models.database import (
                get_total_logs_for_research,
            )

            result = get_total_logs_for_research("test-research-id")

            assert result == 42

    def test_get_total_logs_zero(self):
        """Test total log count when no logs exist."""
        mock_session = Mock()
        mock_query = Mock()
        mock_query.filter.return_value = mock_query
        mock_query.count.return_value = 0
        mock_session.query.return_value = mock_query

        with patch(
            "local_deep_research.web.models.database.get_user_db_session"
        ) as mock_get_session:
            mock_get_session.return_value.__enter__ = Mock(
                return_value=mock_session
            )
            mock_get_session.return_value.__exit__ = Mock(return_value=False)

            from local_deep_research.web.models.database import (
                get_total_logs_for_research,
            )

            result = get_total_logs_for_research("nonexistent-id")

            assert result == 0


class TestDurationEdgeCases:
    """Edge case tests for calculate_duration."""

    def test_negative_duration(self):
        """Test when completed_at is before created_at."""
        from local_deep_research.web.models.database import calculate_duration

        created_at = "2025-01-01T10:10:00+00:00"
        completed_at = "2025-01-01T10:00:00+00:00"

        result = calculate_duration(created_at, completed_at)

        # Should return negative value
        assert result == -600  # -10 minutes

    def test_very_long_duration(self):
        """Test with very long duration."""
        from local_deep_research.web.models.database import calculate_duration

        created_at = "2025-01-01T00:00:00+00:00"
        completed_at = "2025-01-02T00:00:00+00:00"

        result = calculate_duration(created_at, completed_at)

        assert result == 86400  # 24 hours = 86400 seconds

    def test_zero_duration(self):
        """Test with same timestamps."""
        from local_deep_research.web.models.database import calculate_duration

        created_at = "2025-01-01T10:00:00+00:00"
        completed_at = "2025-01-01T10:00:00+00:00"

        result = calculate_duration(created_at, completed_at)

        assert result == 0