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

"""Tests for context overflow API logic paths not covered by existing tests.

Focus: pagination clamping, empty/null states, chart data token formula.
Production code: src/local_deep_research/web/routes/context_overflow_api.py

Uses Flask test client approach from test_context_overflow_api_http.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,
)

_ROUTES_MOD = "local_deep_research.web.routes.context_overflow_api"


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


def _create_test_app():
    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

    _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

    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(get_setting=Mock(return_value=8192)),
        ),
    ]

    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()


def _make_overview_and_token_rows(
    total_requests=0,
    requests_with_context=0,
    truncated_requests=0,
):
    """Create the merged overview row (counts + AVG + token-summary fields).

    The route now issues a single .first() call. The legacy two-row return
    shape is kept so existing call sites don't change — `tk` is the same
    object as `ov`, since fields are merged onto a single mock.
    """
    overview_row = Mock(
        total_requests=total_requests,
        requests_with_context=requests_with_context,
        truncated_requests=truncated_requests,
        avg_tokens_truncated=0,
        total_tokens=0,
        total_prompt_tokens=0,
        total_completion_tokens=0,
        avg_prompt_tokens=0,
        avg_completion_tokens=0,
        max_prompt_tokens=0,
    )
    return overview_row, overview_row


def _setup_multi_first(mock_query, overview_row, token_row):
    """Wire mock_query.first() to return the (merged) overview row."""
    mock_query.first.return_value = overview_row
    mock_query.scalar.return_value = 0


# ---------------------------------------------------------------------------
# Pagination logic
# ---------------------------------------------------------------------------


class TestPaginationLogic:
    """Tests for pagination parameter clamping."""

    def test_per_page_clamped_to_500(self):
        app = _create_test_app()
        with _authenticated_client(app) as (client, _, mock_query):
            ov, tk = _make_overview_and_token_rows()
            _setup_multi_first(mock_query, ov, tk)

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

    def test_per_page_zero_clamped_to_1(self):
        app = _create_test_app()
        with _authenticated_client(app) as (client, _, mock_query):
            ov, tk = _make_overview_and_token_rows()
            _setup_multi_first(mock_query, ov, tk)

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

    def test_per_page_negative_clamped_to_1(self):
        app = _create_test_app()
        with _authenticated_client(app) as (client, _, mock_query):
            ov, tk = _make_overview_and_token_rows()
            _setup_multi_first(mock_query, ov, tk)

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

    def test_page_zero_clamped_to_1(self):
        app = _create_test_app()
        with _authenticated_client(app) as (client, _, mock_query):
            ov, tk = _make_overview_and_token_rows()
            _setup_multi_first(mock_query, ov, tk)

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

    def test_page_2_with_items(self):
        """page=2, per_page=10 with 15 total → correct pagination metadata."""
        app = _create_test_app()
        with _authenticated_client(app) as (client, _, mock_query):
            ov, tk = _make_overview_and_token_rows(total_requests=15)
            _setup_multi_first(mock_query, ov, tk)
            mock_query.count.return_value = 15

            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
            assert data["pagination"]["total_count"] == 15
            assert data["pagination"]["total_pages"] == 2


# ---------------------------------------------------------------------------
# Empty/null state
# ---------------------------------------------------------------------------


class TestEmptyNullStates:
    """Tests for zero/null edge cases."""

    def test_zero_records_overview_all_zeros(self):
        app = _create_test_app()
        with _authenticated_client(app) as (client, _, mock_query):
            ov, tk = _make_overview_and_token_rows()
            _setup_multi_first(mock_query, ov, tk)

            resp = client.get("/api/context-overflow")
            assert resp.status_code == 200
            data = resp.get_json()
            assert data["overview"]["total_requests"] == 0
            assert data["overview"]["truncation_rate"] == 0
            assert data["chart_data"] == []

    def test_requests_with_context_zero_truncation_rate_zero(self):
        """requests_with_context=0 → truncation_rate=0 (not NaN)."""
        app = _create_test_app()
        with _authenticated_client(app) as (client, _, mock_query):
            ov, tk = _make_overview_and_token_rows(
                total_requests=5, requests_with_context=0, truncated_requests=0
            )
            _setup_multi_first(mock_query, ov, tk)

            resp = client.get("/api/context-overflow")
            assert resp.status_code == 200
            assert resp.get_json()["overview"]["truncation_rate"] == 0

    def test_avg_tokens_truncated_none_becomes_zero(self):
        """scalar() returns None → avg_tokens_truncated=0."""
        app = _create_test_app()
        with _authenticated_client(app) as (client, _, mock_query):
            ov, tk = _make_overview_and_token_rows()
            _setup_multi_first(mock_query, ov, tk)
            # scalar for avg_tokens_truncated returns None
            mock_query.scalar.return_value = None

            resp = client.get("/api/context-overflow")
            assert resp.status_code == 200
            assert resp.get_json()["overview"]["avg_tokens_truncated"] == 0

    def test_nonexistent_research_id_empty_response(self):
        app = _create_test_app()
        with _authenticated_client(app) as (client, _, mock_query):
            mock_query.all.return_value = []

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

    def test_all_context_truncated_false(self):
        """All records have context_truncated=False → truncated_requests=0."""
        app = _create_test_app()
        with _authenticated_client(app) as (client, _, mock_query):
            ov, tk = _make_overview_and_token_rows(
                total_requests=3, requests_with_context=3, truncated_requests=0
            )
            _setup_multi_first(mock_query, ov, tk)

            resp = client.get("/api/context-overflow")
            assert resp.status_code == 200
            data = resp.get_json()
            assert data["overview"]["truncated_requests"] == 0
            assert data["overview"]["truncation_rate"] == 0


# ---------------------------------------------------------------------------
# Chart data token formula
# ---------------------------------------------------------------------------


def _make_usage_mock(
    prompt_tokens=100,
    completion_tokens=20,
    ollama_prompt_eval_count=None,
    context_truncated=False,
    tokens_truncated=0,
    context_limit=4096,
    research_id="res-1",
    model_name="gpt-4",
    model_provider="openai",
    research_query="test",
    research_phase="search",
    total_tokens=120,
    truncation_ratio=None,
    ollama_eval_count=None,
    calling_function="search",
    response_time_ms=500,
):
    """Create a mock TokenUsage record."""
    m = Mock()
    m.prompt_tokens = prompt_tokens
    m.completion_tokens = completion_tokens
    m.ollama_prompt_eval_count = ollama_prompt_eval_count
    m.context_truncated = context_truncated
    m.tokens_truncated = tokens_truncated
    m.context_limit = context_limit
    m.research_id = research_id
    m.model_name = model_name
    m.model_provider = model_provider
    m.research_query = research_query
    m.research_phase = research_phase
    m.total_tokens = total_tokens
    m.truncation_ratio = truncation_ratio
    m.calling_function = calling_function
    m.response_time_ms = response_time_ms
    m.timestamp = datetime(2024, 1, 1, tzinfo=timezone.utc)
    return m


def _setup_chart_data_mocks(mock_query, ov, tk, usage_list):
    """Set up mocks for chart_data tests.

    The production code calls .all() 7 times in order:
    1. model_token_query 2. phase_query 3. context_limits
    4. recent_truncated 5. time_series_data 6. model_stats 7. all_requests_data
    We put usage_list at position 5 (time_series) and 7 (all_requests), empty elsewhere.
    """
    _setup_multi_first(mock_query, ov, tk)
    mock_query.all.side_effect = [
        [],  # model_token_query
        [],  # phase_query
        [],  # context_limits
        [],  # recent_truncated
        usage_list,  # time_series_data → chart_data
        [],  # model_stats
        usage_list,  # all_requests_data
    ]
    mock_query.count.return_value = len(usage_list)


class TestChartDataTokenFormula:
    """Tests for chart_data token calculation logic."""

    def test_ollama_prompt_eval_count_used_when_present(self):
        """ollama_prompt_eval_count present → used instead of prompt_tokens."""
        app = _create_test_app()
        with _authenticated_client(app) as (client, _, mock_query):
            usage = _make_usage_mock(
                prompt_tokens=100,
                ollama_prompt_eval_count=80,
                context_truncated=False,
            )
            ov, tk = _make_overview_and_token_rows(
                total_requests=1, requests_with_context=1
            )
            _setup_chart_data_mocks(mock_query, ov, tk, [usage])

            resp = client.get("/api/context-overflow")
            assert resp.status_code == 200
            chart = resp.get_json()["chart_data"]
            assert len(chart) == 1
            assert chart[0]["ollama_prompt_tokens"] == 80
            # original = 80 (ollama used), no truncation
            assert chart[0]["original_prompt_tokens"] == 80

    def test_ollama_none_falls_back_to_prompt_tokens(self):
        """ollama_prompt_eval_count=None → falls back to prompt_tokens."""
        app = _create_test_app()
        with _authenticated_client(app) as (client, _, mock_query):
            usage = _make_usage_mock(
                prompt_tokens=100,
                ollama_prompt_eval_count=None,
                context_truncated=False,
            )
            ov, tk = _make_overview_and_token_rows(total_requests=1)
            _setup_chart_data_mocks(mock_query, ov, tk, [usage])

            resp = client.get("/api/context-overflow")
            assert resp.status_code == 200
            chart = resp.get_json()["chart_data"]
            assert chart[0]["original_prompt_tokens"] == 100

    def test_truncated_adds_tokens_truncated(self):
        """context_truncated=True → original = prompt + tokens_truncated."""
        app = _create_test_app()
        with _authenticated_client(app) as (client, _, mock_query):
            usage = _make_usage_mock(
                prompt_tokens=100,
                ollama_prompt_eval_count=None,
                context_truncated=True,
                tokens_truncated=500,
            )
            ov, tk = _make_overview_and_token_rows(
                total_requests=1, requests_with_context=1, truncated_requests=1
            )
            _setup_chart_data_mocks(mock_query, ov, tk, [usage])

            resp = client.get("/api/context-overflow")
            assert resp.status_code == 200
            chart = resp.get_json()["chart_data"]
            assert chart[0]["original_prompt_tokens"] == 600  # 100 + 500

    def test_not_truncated_no_addition(self):
        """context_truncated=False → original = prompt (no addition)."""
        app = _create_test_app()
        with _authenticated_client(app) as (client, _, mock_query):
            usage = _make_usage_mock(
                prompt_tokens=200,
                context_truncated=False,
                tokens_truncated=0,
            )
            ov, tk = _make_overview_and_token_rows(total_requests=1)
            _setup_chart_data_mocks(mock_query, ov, tk, [usage])

            resp = client.get("/api/context-overflow")
            assert resp.status_code == 200
            chart = resp.get_json()["chart_data"]
            assert chart[0]["original_prompt_tokens"] == 200

    def test_all_null_token_fields_no_crash(self):
        """All null token fields → no crash, 0 values."""
        app = _create_test_app()
        with _authenticated_client(app) as (client, _, mock_query):
            usage = _make_usage_mock(
                prompt_tokens=0,
                completion_tokens=0,
                ollama_prompt_eval_count=None,
                context_truncated=False,
                tokens_truncated=None,
                total_tokens=0,
            )
            ov, tk = _make_overview_and_token_rows(total_requests=1)
            _setup_chart_data_mocks(mock_query, ov, tk, [usage])

            resp = client.get("/api/context-overflow")
            assert resp.status_code == 200
            chart = resp.get_json()["chart_data"]
            assert chart[0]["original_prompt_tokens"] == 0
            assert chart[0]["tokens_truncated"] == 0