learningcircuit--local-deep-research/tests/web/routes/test_context_overflow_api_http.py

"""
HTTP integration tests for context overflow API endpoints.

The existing 673 unit tests only test math helpers. Two "route tests" at
lines 495-524 are bogus (test wrong URLs, accept any status code). This file
provides real Flask test client tests.

Source: src/local_deep_research/web/routes/context_overflow_api.py
"""

from contextlib import contextmanager
from datetime import datetime, timezone
from unittest.mock import Mock, patch

from flask import Flask, jsonify

from local_deep_research.web.auth.routes import auth_bp
from local_deep_research.web.routes.context_overflow_api import (
    context_overflow_bp,
)


# ---------------------------------------------------------------------------
# Test Infrastructure
# ---------------------------------------------------------------------------


def _create_test_app():
    """Create Flask app with auth + context_overflow blueprints."""
    app = Flask(__name__)
    app.config["SECRET_KEY"] = "test-secret"
    app.config["WTF_CSRF_ENABLED"] = False
    app.register_blueprint(auth_bp)
    app.register_blueprint(context_overflow_bp)

    @app.errorhandler(500)
    def _handle_500(error):
        return jsonify({"error": "Internal server error"}), 500

    return app


@contextmanager
def _authenticated_client(app):
    """Provide test client with mocked auth and DB session."""
    mock_db = Mock()
    mock_db.connections = {"testuser": True}
    mock_db.has_encryption = False

    # Build a chainable mock query
    _mock_query = Mock()
    _mock_query.all.return_value = []
    _mock_query.first.return_value = None
    _mock_query.count.return_value = 0
    _mock_query.scalar.return_value = 0
    _mock_query.filter_by.return_value = _mock_query
    _mock_query.filter.return_value = _mock_query
    _mock_query.order_by.return_value = _mock_query
    _mock_query.limit.return_value = _mock_query
    _mock_query.offset.return_value = _mock_query
    _mock_query.group_by.return_value = _mock_query
    _mock_query.with_entities.return_value = _mock_query

    _mock_db_session = Mock()
    _mock_db_session.query.return_value = _mock_query

    @contextmanager
    def _fake_session(*args, **kwargs):
        yield _mock_db_session

    _routes_mod = "local_deep_research.web.routes.context_overflow_api"

    # SettingsManager(session) is invoked at the end of the route to read
    # llm.local_context_window_size; with a Mock session its lazy-init code
    # path raises. Patch the class so it returns a stub manager that yields
    # a fixed setting value.
    _mock_settings_manager = Mock()
    _mock_settings_manager.get_setting.return_value = 4096

    patches = [
        patch("local_deep_research.web.auth.decorators.db_manager", mock_db),
        patch(
            f"{_routes_mod}.get_user_db_session",
            side_effect=_fake_session,
        ),
        patch(
            f"{_routes_mod}.SettingsManager",
            return_value=_mock_settings_manager,
        ),
    ]

    try:
        for p in patches:
            p.start()
        with app.test_client() as client:
            with client.session_transaction() as sess:
                sess["username"] = "testuser"
                sess["session_id"] = "test-session-id"
            yield client, _mock_db_session, _mock_query
    finally:
        for p in reversed(patches):
            p.stop()


# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------


_TOKEN_SUMMARY_FIELDS = (
    "total_tokens",
    "total_prompt_tokens",
    "total_completion_tokens",
    "avg_prompt_tokens",
    "avg_completion_tokens",
    "max_prompt_tokens",
)


def _wire_first(mock_query, overview_row, token_row, avg_tokens_truncated=0):
    """Wire mock_query.first() for the merged single-query route.

    The route now issues one .first() returning a row that combines the old
    overview_counts + token_summary_row + AVG(tokens_truncated) scalar.
    Tests still construct two separate mocks for readability; this helper
    merges token_row's fields onto overview_row and wires the single call.
    """
    for field in _TOKEN_SUMMARY_FIELDS:
        setattr(overview_row, field, getattr(token_row, field))
    overview_row.avg_tokens_truncated = avg_tokens_truncated
    mock_query.first.return_value = overview_row


# ---------------------------------------------------------------------------
# GET /api/context-overflow
# ---------------------------------------------------------------------------


class TestGetContextOverflowMetrics:
    """Tests for GET /api/context-overflow."""

    def test_unauthenticated_returns_401(self):
        """Unauthenticated request returns 401."""
        app = _create_test_app()
        mock_db = Mock()
        mock_db.connections = {}
        mock_db.has_encryption = False

        with patch(
            "local_deep_research.web.auth.decorators.db_manager", mock_db
        ):
            with app.test_client() as client:
                resp = client.get("/api/context-overflow")
                assert resp.status_code == 401

    def test_default_period_success(self):
        """Default period (30d) returns success with overview."""
        app = _create_test_app()
        with _authenticated_client(app) as (client, mock_session, mock_query):
            # Mock overview_counts result
            overview_row = Mock()
            overview_row.total_requests = 5
            overview_row.requests_with_context = 3
            overview_row.truncated_requests = 1

            # Mock token_summary_row
            token_row = Mock()
            token_row.total_requests = 5
            token_row.total_tokens = 1000
            token_row.total_prompt_tokens = 700
            token_row.total_completion_tokens = 300
            token_row.avg_prompt_tokens = 140.0
            token_row.avg_completion_tokens = 60.0
            token_row.max_prompt_tokens = 200

            _wire_first(mock_query, overview_row, token_row)

            resp = client.get("/api/context-overflow")
            assert resp.status_code == 200
            data = resp.get_json()
            assert data["status"] == "success"
            assert "overview" in data

    def test_period_7d_accepted(self):
        """period=7d is a valid period."""
        app = _create_test_app()
        with _authenticated_client(app) as (client, mock_session, mock_query):
            overview_row = Mock(
                total_requests=0, requests_with_context=0, truncated_requests=0
            )
            token_row = Mock(
                total_requests=0,
                total_tokens=0,
                total_prompt_tokens=0,
                total_completion_tokens=0,
                avg_prompt_tokens=0,
                avg_completion_tokens=0,
                max_prompt_tokens=0,
            )

            _wire_first(mock_query, overview_row, token_row)

            resp = client.get("/api/context-overflow?period=7d")
            assert resp.status_code == 200

    def test_period_all_accepted(self):
        """period=all returns success (no date filter)."""
        app = _create_test_app()
        with _authenticated_client(app) as (client, mock_session, mock_query):
            overview_row = Mock(
                total_requests=0, requests_with_context=0, truncated_requests=0
            )
            token_row = Mock(
                total_requests=0,
                total_tokens=0,
                total_prompt_tokens=0,
                total_completion_tokens=0,
                avg_prompt_tokens=0,
                avg_completion_tokens=0,
                max_prompt_tokens=0,
            )
            _wire_first(mock_query, overview_row, token_row)

            resp = client.get("/api/context-overflow?period=all")
            assert resp.status_code == 200

    def test_invalid_period_defaults_to_30d(self):
        """Invalid period value defaults to 30d."""
        app = _create_test_app()
        with _authenticated_client(app) as (client, mock_session, mock_query):
            overview_row = Mock(
                total_requests=0, requests_with_context=0, truncated_requests=0
            )
            token_row = Mock(
                total_requests=0,
                total_tokens=0,
                total_prompt_tokens=0,
                total_completion_tokens=0,
                avg_prompt_tokens=0,
                avg_completion_tokens=0,
                max_prompt_tokens=0,
            )
            _wire_first(mock_query, overview_row, token_row)

            resp = client.get("/api/context-overflow?period=invalid")
            assert resp.status_code == 200

    def test_pagination_params(self):
        """page and per_page params are accepted."""
        app = _create_test_app()
        with _authenticated_client(app) as (client, mock_session, mock_query):
            overview_row = Mock(
                total_requests=0, requests_with_context=0, truncated_requests=0
            )
            token_row = Mock(
                total_requests=0,
                total_tokens=0,
                total_prompt_tokens=0,
                total_completion_tokens=0,
                avg_prompt_tokens=0,
                avg_completion_tokens=0,
                max_prompt_tokens=0,
            )
            _wire_first(mock_query, overview_row, token_row)

            resp = client.get("/api/context-overflow?page=2&per_page=10")
            assert resp.status_code == 200
            data = resp.get_json()
            assert data["pagination"]["page"] == 2
            assert data["pagination"]["per_page"] == 10

    def test_per_page_clamped_to_500(self):
        """per_page > 500 is clamped to 500."""
        app = _create_test_app()
        with _authenticated_client(app) as (client, mock_session, mock_query):
            overview_row = Mock(
                total_requests=0, requests_with_context=0, truncated_requests=0
            )
            token_row = Mock(
                total_requests=0,
                total_tokens=0,
                total_prompt_tokens=0,
                total_completion_tokens=0,
                avg_prompt_tokens=0,
                avg_completion_tokens=0,
                max_prompt_tokens=0,
            )
            _wire_first(mock_query, overview_row, token_row)

            resp = client.get("/api/context-overflow?per_page=999")
            assert resp.status_code == 200
            data = resp.get_json()
            assert data["pagination"]["per_page"] == 500

    def test_per_page_min_is_1(self):
        """per_page < 1 is clamped to 1."""
        app = _create_test_app()
        with _authenticated_client(app) as (client, mock_session, mock_query):
            overview_row = Mock(
                total_requests=0, requests_with_context=0, truncated_requests=0
            )
            token_row = Mock(
                total_requests=0,
                total_tokens=0,
                total_prompt_tokens=0,
                total_completion_tokens=0,
                avg_prompt_tokens=0,
                avg_completion_tokens=0,
                max_prompt_tokens=0,
            )
            _wire_first(mock_query, overview_row, token_row)

            resp = client.get("/api/context-overflow?per_page=-5")
            assert resp.status_code == 200
            data = resp.get_json()
            assert data["pagination"]["per_page"] == 1

    def test_db_exception_returns_500(self):
        """DB exception returns 500 with error message."""
        app = _create_test_app()
        mock_db = Mock()
        mock_db.connections = {"testuser": True}
        mock_db.has_encryption = False

        @contextmanager
        def _exploding_session(*args, **kwargs):
            raise RuntimeError("DB connection failed")
            yield  # pragma: no cover

        _routes_mod = "local_deep_research.web.routes.context_overflow_api"
        with (
            patch(
                "local_deep_research.web.auth.decorators.db_manager", mock_db
            ),
            patch(
                f"{_routes_mod}.get_user_db_session",
                side_effect=_exploding_session,
            ),
        ):
            with app.test_client() as client:
                with client.session_transaction() as sess:
                    sess["username"] = "testuser"
                    sess["session_id"] = "test-session-id"
                resp = client.get("/api/context-overflow")
                assert resp.status_code == 500
                data = resp.get_json()
                assert data["status"] == "error"


# ---------------------------------------------------------------------------
# GET /api/research/<id>/context-overflow
# ---------------------------------------------------------------------------


class TestGetResearchContextOverflow:
    """Tests for GET /api/research/<id>/context-overflow."""

    def test_no_token_usage_returns_empty(self):
        """No token usage for research returns empty data."""
        app = _create_test_app()
        with _authenticated_client(app) as (client, mock_session, mock_query):
            mock_query.all.return_value = []

            resp = client.get("/api/research/res-1/context-overflow")
            assert resp.status_code == 200
            data = resp.get_json()
            assert data["status"] == "success"
            assert data["data"]["overview"]["total_requests"] == 0

    def test_with_data_returns_overview_and_requests(self):
        """With token usage data, returns overview + phase_stats + requests."""
        app = _create_test_app()
        with _authenticated_client(app) as (client, mock_session, mock_query):
            usage1 = Mock()
            usage1.total_tokens = 100
            usage1.prompt_tokens = 80
            usage1.completion_tokens = 20
            usage1.context_limit = 4096
            usage1.context_truncated = False
            usage1.tokens_truncated = 0
            usage1.research_phase = "search"
            usage1.timestamp = datetime(2024, 1, 1, tzinfo=timezone.utc)
            usage1.ollama_prompt_eval_count = None
            usage1.calling_function = "search"
            usage1.response_time_ms = 500
            usage1.model_name = "gpt-4"
            usage1.model_provider = "openai"
            usage1.research_id = "res-1"

            usage2 = Mock()
            usage2.total_tokens = 200
            usage2.prompt_tokens = 150
            usage2.completion_tokens = 50
            usage2.context_limit = 4096
            usage2.context_truncated = True
            usage2.tokens_truncated = 30
            usage2.research_phase = "analysis"
            usage2.timestamp = datetime(2024, 1, 2, tzinfo=timezone.utc)
            usage2.ollama_prompt_eval_count = None
            usage2.calling_function = "analyze"
            usage2.response_time_ms = 800
            usage2.model_name = "gpt-4"
            usage2.model_provider = "openai"
            usage2.research_id = "res-1"

            mock_query.all.return_value = [usage1, usage2]

            resp = client.get("/api/research/res-1/context-overflow")
            assert resp.status_code == 200
            data = resp.get_json()
            assert data["status"] == "success"
            assert data["data"]["overview"]["total_requests"] == 2
            assert data["data"]["overview"]["total_tokens"] == 300
            assert data["data"]["overview"]["truncation_occurred"] is True
            assert "search" in data["data"]["phase_stats"]
            assert "analysis" in data["data"]["phase_stats"]
            assert len(data["data"]["requests"]) == 2

    def test_phase_stats_grouped_correctly(self):
        """Phase stats are grouped by research_phase."""
        app = _create_test_app()
        with _authenticated_client(app) as (client, mock_session, mock_query):
            # Two entries in same phase
            usage1 = Mock(
                total_tokens=100,
                prompt_tokens=80,
                completion_tokens=20,
                context_limit=4096,
                context_truncated=False,
                tokens_truncated=0,
                research_phase="search",
                model_name="gpt-4",
                model_provider="openai",
                timestamp=datetime(2024, 1, 1, tzinfo=timezone.utc),
                ollama_prompt_eval_count=None,
                calling_function="s",
                response_time_ms=100,
                research_id="res-1",
            )
            usage2 = Mock(
                total_tokens=100,
                prompt_tokens=80,
                completion_tokens=20,
                context_limit=4096,
                context_truncated=True,
                tokens_truncated=10,
                research_phase="search",
                model_name="gpt-4",
                model_provider="openai",
                timestamp=datetime(2024, 1, 2, tzinfo=timezone.utc),
                ollama_prompt_eval_count=None,
                calling_function="s",
                response_time_ms=100,
                research_id="res-1",
            )
            mock_query.all.return_value = [usage1, usage2]

            resp = client.get("/api/research/res-1/context-overflow")
            data = resp.get_json()

            search_stats = data["data"]["phase_stats"]["search"]
            assert search_stats["count"] == 2
            assert search_stats["truncated_count"] == 1
            assert search_stats["total_tokens"] == 200

    def test_db_exception_returns_500(self):
        """DB exception returns 500."""
        app = _create_test_app()
        mock_db = Mock()
        mock_db.connections = {"testuser": True}
        mock_db.has_encryption = False

        @contextmanager
        def _exploding_session(*args, **kwargs):
            raise RuntimeError("DB error")
            yield  # pragma: no cover

        _routes_mod = "local_deep_research.web.routes.context_overflow_api"
        with (
            patch(
                "local_deep_research.web.auth.decorators.db_manager", mock_db
            ),
            patch(
                f"{_routes_mod}.get_user_db_session",
                side_effect=_exploding_session,
            ),
        ):
            with app.test_client() as client:
                with client.session_transaction() as sess:
                    sess["username"] = "testuser"
                    sess["session_id"] = "test-session-id"
                resp = client.get("/api/research/res-1/context-overflow")
                assert resp.status_code == 500