7a0da7932b
OSV-Scanner (Scheduled) / scan-scheduled (push) Failing after 0s
Create Release / test-gate (push) Has been cancelled
Create Release / release-gate (push) Has been cancelled
Create Release / ci-gate (push) Has been cancelled
Create Release / version-check (push) Has been cancelled
Create Release / e2e-test-gate (push) Has been cancelled
Create Release / responsive-test-gate (push) Has been cancelled
Create Release / compat-test-gate (push) Has been cancelled
Create Release / compose-integration-gate (push) Has been cancelled
Create Release / vulture-gate (push) Has been cancelled
Create Release / build (push) Has been cancelled
Create Release / provenance (push) Has been cancelled
Create Release / prerelease-docker (push) Has been cancelled
Create Release / publish-docker (push) Has been cancelled
Create Release / create-release (push) Has been cancelled
Create Release / cleanup-changelog (push) Has been cancelled
Create Release / trigger-pypi (push) Has been cancelled
Create Release / monitor-pypi (push) Has been cancelled
Create Release / Clean up orphan prerelease tags and signatures (push) Has been cancelled
Docker Tests (Consolidated) / UI Tests (Puppeteer) [research-form] (push) Has been cancelled
Docker Tests (Consolidated) / UI Tests (Puppeteer) [research-metrics] (push) Has been cancelled
Docker Tests (Consolidated) / UI Tests (Puppeteer) [research-workflow] (push) Has been cancelled
Docker Tests (Consolidated) / UI Tests (Puppeteer) [settings-core] (push) Has been cancelled
CodeQL Advanced / Analyze (javascript-typescript) (push) Has been cancelled
Docker Tests (Consolidated) / UI Tests (Puppeteer) [history-news] (push) Has been cancelled
Docker Tests (Consolidated) / UI Tests (Puppeteer) [library] (push) Has been cancelled
Docker Tests (Consolidated) / UI Tests (Puppeteer) [link-analytics] (push) Has been cancelled
Docker Tests (Consolidated) / UI Tests (Puppeteer) [chat-core] (push) Has been cancelled
Docker Tests (Consolidated) / UI Tests (Puppeteer) [chat-lifecycle] (push) Has been cancelled
Docker Tests (Consolidated) / UI Tests (Puppeteer) [error-benchmark] (push) Has been cancelled
Docker Tests (Consolidated) / UI Tests (Puppeteer) [settings-pages] (push) Has been cancelled
Docker Tests (Consolidated) / UI Tests (Puppeteer) (push) Has been cancelled
Docker Tests (Consolidated) / Accessibility Tests (push) Has been cancelled
Docker Tests (Consolidated) / LLM Unit Tests (push) Has been cancelled
Docker Tests (Consolidated) / LLM Example Tests (push) Has been cancelled
Docker Tests (Consolidated) / Production Image Smoke Test (push) Has been cancelled
Docker Tests (Consolidated) / Infrastructure Tests (push) Has been cancelled
OSSF Scorecard / OSSF Security Scorecard Analysis (push) Has been cancelled
Docker Tests (Consolidated) / UI Tests (Puppeteer) [mobile] (push) Has been cancelled
Backwards Compatibility / Verify Encryption Constants (push) Has been cancelled
Backwards Compatibility / PyPI Version Compatibility (push) Has been cancelled
Backwards Compatibility / Database Migration Tests (push) Has been cancelled
CodeQL Advanced / Analyze (python) (push) Has been cancelled
Docker Tests (Consolidated) / detect-changes (push) Has been cancelled
Docker Tests (Consolidated) / Build Test Image (push) Has been cancelled
Docker Tests (Consolidated) / All Pytest Tests + Coverage (push) Has been cancelled
Docker Tests (Consolidated) / UI Tests (Puppeteer) [accessibility] (push) Has been cancelled
Docker Tests (Consolidated) / UI Tests (Puppeteer) [api-crud] (push) Has been cancelled
Docker Tests (Consolidated) / UI Tests (Puppeteer) [auth-login] (push) Has been cancelled
Docker Tests (Consolidated) / UI Tests (Puppeteer) [auth-pages] (push) Has been cancelled
Docker Tests (Consolidated) / UI Tests (Puppeteer) [auth-register] (push) Has been cancelled
857 lines
31 KiB
Python
857 lines
31 KiB
Python
"""
|
|
Tests for web/routes/context_overflow_api.py
|
|
|
|
Tests cover:
|
|
- get_context_overflow_metrics endpoint
|
|
- get_research_context_overflow endpoint
|
|
- Various time period filters
|
|
- Error handling
|
|
"""
|
|
|
|
import pytest
|
|
from unittest.mock import Mock
|
|
from datetime import datetime, timezone
|
|
|
|
|
|
class TestGetContextOverflowMetricsUnit:
|
|
"""Unit tests for get_context_overflow_metrics logic."""
|
|
|
|
def test_calculate_truncation_rate(self):
|
|
"""Test truncation rate calculation."""
|
|
# Test the calculation logic
|
|
# audit: PUNCHLIST reviewed 2026-05 — issue resolved by prior PR (recommendation: DELETE).
|
|
requests_with_context = 100
|
|
truncated_requests = 15
|
|
|
|
truncation_rate = 0
|
|
if requests_with_context > 0:
|
|
truncation_rate = (truncated_requests / requests_with_context) * 100
|
|
|
|
assert truncation_rate == 15.0
|
|
|
|
def test_calculate_truncation_rate_zero_division(self):
|
|
"""Test truncation rate with zero requests."""
|
|
# audit: PUNCHLIST reviewed 2026-05 — issue resolved by prior PR (recommendation: DELETE).
|
|
requests_with_context = 0
|
|
truncated_requests = 0
|
|
|
|
truncation_rate = 0
|
|
if requests_with_context > 0:
|
|
truncation_rate = (truncated_requests / requests_with_context) * 100
|
|
|
|
assert truncation_rate == 0
|
|
|
|
|
|
class TestTimePeriodCalculation:
|
|
"""Tests for time period calculation logic."""
|
|
|
|
def test_7d_period(self):
|
|
"""Test 7 day period calculation."""
|
|
# audit: PUNCHLIST reviewed 2026-05 — issue resolved by prior PR (recommendation: DELETE).
|
|
from datetime import timedelta
|
|
|
|
now = datetime.now(timezone.utc)
|
|
period = "7d"
|
|
|
|
start_date = None
|
|
if period != "all":
|
|
if period == "7d":
|
|
start_date = now - timedelta(days=7)
|
|
|
|
assert start_date is not None
|
|
assert (now - start_date).days == 7
|
|
|
|
def test_30d_period(self):
|
|
"""Test 30 day period calculation."""
|
|
# audit: PUNCHLIST reviewed 2026-05 — issue resolved by prior PR (recommendation: DELETE).
|
|
from datetime import timedelta
|
|
|
|
now = datetime.now(timezone.utc)
|
|
period = "30d"
|
|
|
|
start_date = None
|
|
if period != "all":
|
|
if period == "30d":
|
|
start_date = now - timedelta(days=30)
|
|
|
|
assert start_date is not None
|
|
assert (now - start_date).days == 30
|
|
|
|
def test_3m_period(self):
|
|
"""Test 3 month period calculation."""
|
|
# audit: PUNCHLIST reviewed 2026-05 — issue resolved by prior PR (recommendation: DELETE).
|
|
from datetime import timedelta
|
|
|
|
now = datetime.now(timezone.utc)
|
|
period = "3m"
|
|
|
|
start_date = None
|
|
if period != "all":
|
|
if period == "3m":
|
|
start_date = now - timedelta(days=90)
|
|
|
|
assert start_date is not None
|
|
assert (now - start_date).days == 90
|
|
|
|
def test_1y_period(self):
|
|
"""Test 1 year period calculation."""
|
|
# audit: PUNCHLIST reviewed 2026-05 — issue resolved by prior PR (recommendation: DELETE).
|
|
from datetime import timedelta
|
|
|
|
now = datetime.now(timezone.utc)
|
|
period = "1y"
|
|
|
|
start_date = None
|
|
if period != "all":
|
|
if period == "1y":
|
|
start_date = now - timedelta(days=365)
|
|
|
|
assert start_date is not None
|
|
assert (now - start_date).days == 365
|
|
|
|
def test_all_period(self):
|
|
"""Test all time period has no date filter."""
|
|
# audit: PUNCHLIST reviewed 2026-05 — issue resolved by prior PR (recommendation: DELETE).
|
|
period = "all"
|
|
|
|
start_date = None
|
|
if period != "all":
|
|
start_date = datetime.now(timezone.utc)
|
|
|
|
assert start_date is None
|
|
|
|
def test_invalid_period_defaults_to_30d(self):
|
|
"""Test that an invalid period value is normalized to 30d."""
|
|
# audit: PUNCHLIST reviewed 2026-05 — issue resolved by prior PR (recommendation: DELETE).
|
|
VALID_PERIODS = {"7d", "30d", "3m", "1y", "all"}
|
|
|
|
for invalid in ["999d", "abc", "", "10x", "<script>"]:
|
|
period = invalid
|
|
if period not in VALID_PERIODS:
|
|
period = "30d"
|
|
assert period == "30d"
|
|
|
|
def test_valid_periods_are_not_overridden(self):
|
|
"""Test that valid period values pass through unchanged."""
|
|
# audit: PUNCHLIST reviewed 2026-05 — issue resolved by prior PR (recommendation: DELETE).
|
|
VALID_PERIODS = {"7d", "30d", "3m", "1y", "all"}
|
|
|
|
for valid in ["7d", "30d", "3m", "1y", "all"]:
|
|
period = valid
|
|
if period not in VALID_PERIODS:
|
|
period = "30d"
|
|
assert period == valid
|
|
|
|
|
|
class TestChartDataFormatting:
|
|
"""Tests for chart data formatting logic."""
|
|
|
|
def test_format_chart_data_basic(self):
|
|
"""Test basic chart data formatting."""
|
|
# audit: PUNCHLIST reviewed 2026-05 — issue resolved by prior PR (recommendation: DELETE).
|
|
mock_usage = Mock()
|
|
mock_usage.timestamp = datetime(
|
|
2025, 1, 1, 12, 0, 0, tzinfo=timezone.utc
|
|
)
|
|
mock_usage.research_id = "test-research-1"
|
|
mock_usage.prompt_tokens = 1000
|
|
mock_usage.ollama_prompt_eval_count = None
|
|
mock_usage.context_truncated = False
|
|
mock_usage.tokens_truncated = 0
|
|
mock_usage.context_limit = 4096
|
|
mock_usage.model_name = "test-model"
|
|
|
|
# Simulate the chart data creation logic
|
|
ollama_used = mock_usage.ollama_prompt_eval_count
|
|
actual_prompt = ollama_used or mock_usage.prompt_tokens
|
|
tokens_truncated = mock_usage.tokens_truncated or 0
|
|
original_tokens = (
|
|
actual_prompt + tokens_truncated
|
|
if mock_usage.context_truncated
|
|
else actual_prompt
|
|
)
|
|
|
|
chart_item = {
|
|
"timestamp": mock_usage.timestamp.isoformat(),
|
|
"research_id": mock_usage.research_id,
|
|
"prompt_tokens": mock_usage.prompt_tokens,
|
|
"ollama_prompt_tokens": ollama_used,
|
|
"original_prompt_tokens": original_tokens,
|
|
"context_limit": mock_usage.context_limit,
|
|
"truncated": bool(mock_usage.context_truncated),
|
|
"tokens_truncated": tokens_truncated,
|
|
"model": mock_usage.model_name,
|
|
}
|
|
|
|
assert chart_item["prompt_tokens"] == 1000
|
|
assert chart_item["original_prompt_tokens"] == 1000
|
|
assert chart_item["truncated"] is False
|
|
|
|
def test_format_chart_data_with_truncation(self):
|
|
"""Test chart data formatting with truncation."""
|
|
# audit: PUNCHLIST reviewed 2026-05 — issue resolved by prior PR (recommendation: DELETE).
|
|
mock_usage = Mock()
|
|
mock_usage.timestamp = datetime(
|
|
2025, 1, 1, 12, 0, 0, tzinfo=timezone.utc
|
|
)
|
|
mock_usage.research_id = "test-research-1"
|
|
mock_usage.prompt_tokens = 5000
|
|
mock_usage.ollama_prompt_eval_count = 4000
|
|
mock_usage.context_truncated = True
|
|
mock_usage.tokens_truncated = 1000
|
|
mock_usage.context_limit = 4096
|
|
mock_usage.model_name = "test-model"
|
|
|
|
ollama_used = mock_usage.ollama_prompt_eval_count
|
|
actual_prompt = ollama_used or mock_usage.prompt_tokens
|
|
tokens_truncated = mock_usage.tokens_truncated or 0
|
|
original_tokens = (
|
|
actual_prompt + tokens_truncated
|
|
if mock_usage.context_truncated
|
|
else actual_prompt
|
|
)
|
|
|
|
chart_item = {
|
|
"timestamp": mock_usage.timestamp.isoformat(),
|
|
"research_id": mock_usage.research_id,
|
|
"prompt_tokens": mock_usage.prompt_tokens,
|
|
"ollama_prompt_tokens": ollama_used,
|
|
"original_prompt_tokens": original_tokens,
|
|
"context_limit": mock_usage.context_limit,
|
|
"truncated": bool(mock_usage.context_truncated),
|
|
"tokens_truncated": tokens_truncated,
|
|
"model": mock_usage.model_name,
|
|
}
|
|
|
|
assert chart_item["truncated"] is True
|
|
assert chart_item["tokens_truncated"] == 1000
|
|
assert chart_item["original_prompt_tokens"] == 5000 # 4000 + 1000
|
|
|
|
|
|
class TestPhaseStatsCalculation:
|
|
"""Tests for phase stats calculation logic."""
|
|
|
|
def test_calculate_phase_stats_single_phase(self):
|
|
"""Test phase stats for single phase."""
|
|
# audit: PUNCHLIST reviewed 2026-05 — issue resolved by prior PR (recommendation: DELETE).
|
|
mock_usage = Mock()
|
|
mock_usage.research_phase = "analysis"
|
|
mock_usage.prompt_tokens = 1000
|
|
mock_usage.completion_tokens = 200
|
|
mock_usage.total_tokens = 1200
|
|
mock_usage.context_truncated = False
|
|
|
|
token_usage = [mock_usage]
|
|
|
|
phase_stats = {}
|
|
for req in token_usage:
|
|
phase = req.research_phase or "unknown"
|
|
if phase not in phase_stats:
|
|
phase_stats[phase] = {
|
|
"count": 0,
|
|
"prompt_tokens": 0,
|
|
"completion_tokens": 0,
|
|
"total_tokens": 0,
|
|
"truncated_count": 0,
|
|
}
|
|
phase_stats[phase]["count"] += 1
|
|
phase_stats[phase]["prompt_tokens"] += req.prompt_tokens or 0
|
|
phase_stats[phase]["completion_tokens"] += (
|
|
req.completion_tokens or 0
|
|
)
|
|
phase_stats[phase]["total_tokens"] += req.total_tokens or 0
|
|
if req.context_truncated:
|
|
phase_stats[phase]["truncated_count"] += 1
|
|
|
|
assert "analysis" in phase_stats
|
|
assert phase_stats["analysis"]["count"] == 1
|
|
assert phase_stats["analysis"]["total_tokens"] == 1200
|
|
|
|
def test_calculate_phase_stats_multiple_phases(self):
|
|
"""Test phase stats for multiple phases."""
|
|
# audit: PUNCHLIST reviewed 2026-05 — issue resolved by prior PR (recommendation: DELETE).
|
|
mock_usage1 = Mock()
|
|
mock_usage1.research_phase = "analysis"
|
|
mock_usage1.prompt_tokens = 1000
|
|
mock_usage1.completion_tokens = 200
|
|
mock_usage1.total_tokens = 1200
|
|
mock_usage1.context_truncated = False
|
|
|
|
mock_usage2 = Mock()
|
|
mock_usage2.research_phase = "synthesis"
|
|
mock_usage2.prompt_tokens = 2000
|
|
mock_usage2.completion_tokens = 400
|
|
mock_usage2.total_tokens = 2400
|
|
mock_usage2.context_truncated = True
|
|
|
|
token_usage = [mock_usage1, mock_usage2]
|
|
|
|
phase_stats = {}
|
|
for req in token_usage:
|
|
phase = req.research_phase or "unknown"
|
|
if phase not in phase_stats:
|
|
phase_stats[phase] = {
|
|
"count": 0,
|
|
"prompt_tokens": 0,
|
|
"completion_tokens": 0,
|
|
"total_tokens": 0,
|
|
"truncated_count": 0,
|
|
}
|
|
phase_stats[phase]["count"] += 1
|
|
phase_stats[phase]["prompt_tokens"] += req.prompt_tokens or 0
|
|
phase_stats[phase]["completion_tokens"] += (
|
|
req.completion_tokens or 0
|
|
)
|
|
phase_stats[phase]["total_tokens"] += req.total_tokens or 0
|
|
if req.context_truncated:
|
|
phase_stats[phase]["truncated_count"] += 1
|
|
|
|
assert "analysis" in phase_stats
|
|
assert "synthesis" in phase_stats
|
|
assert phase_stats["synthesis"]["truncated_count"] == 1
|
|
|
|
def test_calculate_phase_stats_unknown_phase(self):
|
|
"""Test phase stats handles unknown phase."""
|
|
# audit: PUNCHLIST reviewed 2026-05 — issue resolved by prior PR (recommendation: DELETE).
|
|
mock_usage = Mock()
|
|
mock_usage.research_phase = None
|
|
mock_usage.prompt_tokens = 500
|
|
mock_usage.completion_tokens = 100
|
|
mock_usage.total_tokens = 600
|
|
mock_usage.context_truncated = False
|
|
|
|
token_usage = [mock_usage]
|
|
|
|
phase_stats = {}
|
|
for req in token_usage:
|
|
phase = req.research_phase or "unknown"
|
|
if phase not in phase_stats:
|
|
phase_stats[phase] = {
|
|
"count": 0,
|
|
"prompt_tokens": 0,
|
|
"completion_tokens": 0,
|
|
"total_tokens": 0,
|
|
"truncated_count": 0,
|
|
}
|
|
phase_stats[phase]["count"] += 1
|
|
phase_stats[phase]["prompt_tokens"] += req.prompt_tokens or 0
|
|
phase_stats[phase]["completion_tokens"] += (
|
|
req.completion_tokens or 0
|
|
)
|
|
phase_stats[phase]["total_tokens"] += req.total_tokens or 0
|
|
if req.context_truncated:
|
|
phase_stats[phase]["truncated_count"] += 1
|
|
|
|
assert "unknown" in phase_stats
|
|
|
|
|
|
class TestOverviewCalculation:
|
|
"""Tests for overview metrics calculation."""
|
|
|
|
def test_calculate_overview_basic(self):
|
|
"""Test basic overview calculation."""
|
|
# audit: PUNCHLIST reviewed 2026-05 — issue resolved by prior PR (recommendation: DELETE).
|
|
mock_usage = Mock()
|
|
mock_usage.total_tokens = 1200
|
|
mock_usage.prompt_tokens = 1000
|
|
mock_usage.completion_tokens = 200
|
|
mock_usage.context_limit = 4096
|
|
mock_usage.context_truncated = False
|
|
mock_usage.tokens_truncated = 0
|
|
|
|
token_usage = [mock_usage]
|
|
|
|
total_tokens = sum(req.total_tokens or 0 for req in token_usage)
|
|
total_prompt = sum(req.prompt_tokens or 0 for req in token_usage)
|
|
total_completion = sum(
|
|
req.completion_tokens or 0 for req in token_usage
|
|
)
|
|
context_limit = next(
|
|
(req.context_limit for req in token_usage if req.context_limit),
|
|
None,
|
|
)
|
|
truncated_requests = [
|
|
req for req in token_usage if req.context_truncated
|
|
]
|
|
max_tokens_used = max((req.prompt_tokens or 0) for req in token_usage)
|
|
|
|
assert total_tokens == 1200
|
|
assert total_prompt == 1000
|
|
assert total_completion == 200
|
|
assert context_limit == 4096
|
|
assert len(truncated_requests) == 0
|
|
assert max_tokens_used == 1000
|
|
|
|
def test_calculate_overview_with_truncation(self):
|
|
"""Test overview with truncation."""
|
|
# audit: PUNCHLIST reviewed 2026-05 — issue resolved by prior PR (recommendation: DELETE).
|
|
mock_usage1 = Mock()
|
|
mock_usage1.total_tokens = 1200
|
|
mock_usage1.prompt_tokens = 1000
|
|
mock_usage1.completion_tokens = 200
|
|
mock_usage1.context_limit = 4096
|
|
mock_usage1.context_truncated = False
|
|
mock_usage1.tokens_truncated = 0
|
|
|
|
mock_usage2 = Mock()
|
|
mock_usage2.total_tokens = 5200
|
|
mock_usage2.prompt_tokens = 5000
|
|
mock_usage2.completion_tokens = 200
|
|
mock_usage2.context_limit = 4096
|
|
mock_usage2.context_truncated = True
|
|
mock_usage2.tokens_truncated = 1000
|
|
|
|
token_usage = [mock_usage1, mock_usage2]
|
|
|
|
truncated_requests = [
|
|
req for req in token_usage if req.context_truncated
|
|
]
|
|
tokens_lost = sum(
|
|
req.tokens_truncated or 0 for req in truncated_requests
|
|
)
|
|
|
|
assert len(truncated_requests) == 1
|
|
assert tokens_lost == 1000
|
|
|
|
|
|
class TestModelStatsFormatting:
|
|
"""Tests for model stats formatting."""
|
|
|
|
def test_format_model_stats(self):
|
|
"""Test model stats formatting."""
|
|
# Mock a SQLAlchemy result row
|
|
# audit: PUNCHLIST reviewed 2026-05 — issue resolved by prior PR (recommendation: DELETE).
|
|
mock_stat = Mock()
|
|
mock_stat.model_name = "test-model"
|
|
mock_stat.model_provider = "test-provider"
|
|
mock_stat.total_requests = 100
|
|
mock_stat.truncated_count = 10
|
|
mock_stat.avg_context_limit = 4096.0
|
|
|
|
model_stats = [mock_stat]
|
|
|
|
formatted = []
|
|
for stat in model_stats:
|
|
formatted.append(
|
|
{
|
|
"model": stat.model_name,
|
|
"provider": stat.model_provider,
|
|
"total_requests": stat.total_requests,
|
|
"truncated_count": int(stat.truncated_count or 0),
|
|
"truncation_rate": round(
|
|
(stat.truncated_count or 0) / stat.total_requests * 100,
|
|
2,
|
|
)
|
|
if stat.total_requests > 0
|
|
else 0,
|
|
"avg_context_limit": round(stat.avg_context_limit, 0)
|
|
if stat.avg_context_limit
|
|
else None,
|
|
}
|
|
)
|
|
|
|
assert formatted[0]["model"] == "test-model"
|
|
assert formatted[0]["truncation_rate"] == 10.0
|
|
assert formatted[0]["avg_context_limit"] == 4096
|
|
|
|
def test_format_model_stats_with_none_values(self):
|
|
"""Test model stats with None values."""
|
|
# audit: PUNCHLIST reviewed 2026-05 — issue resolved by prior PR (recommendation: DELETE).
|
|
mock_stat = Mock()
|
|
mock_stat.model_name = "test-model"
|
|
mock_stat.model_provider = None
|
|
mock_stat.total_requests = 50
|
|
mock_stat.truncated_count = None
|
|
mock_stat.avg_context_limit = None
|
|
|
|
model_stats = [mock_stat]
|
|
|
|
formatted = []
|
|
for stat in model_stats:
|
|
formatted.append(
|
|
{
|
|
"model": stat.model_name,
|
|
"provider": stat.model_provider,
|
|
"total_requests": stat.total_requests,
|
|
"truncated_count": int(stat.truncated_count or 0),
|
|
"truncation_rate": round(
|
|
(stat.truncated_count or 0) / stat.total_requests * 100,
|
|
2,
|
|
)
|
|
if stat.total_requests > 0
|
|
else 0,
|
|
"avg_context_limit": round(stat.avg_context_limit, 0)
|
|
if stat.avg_context_limit
|
|
else None,
|
|
}
|
|
)
|
|
|
|
assert formatted[0]["truncated_count"] == 0
|
|
assert formatted[0]["avg_context_limit"] is None
|
|
|
|
|
|
class TestContextOverflowApiRoutes:
|
|
"""Tests for context overflow API routes."""
|
|
|
|
def test_metrics_url_with_suffix_is_not_registered(self):
|
|
"""The metrics endpoint is mounted at /api/context-overflow (no
|
|
/metrics suffix). A GET to /api/context-overflow/metrics
|
|
therefore must 404 — pinning this so a future URL prefix change
|
|
can't silently move the route."""
|
|
from flask import Flask
|
|
from local_deep_research.web.routes.context_overflow_api import (
|
|
context_overflow_bp,
|
|
)
|
|
|
|
app = Flask(__name__)
|
|
app.config["SECRET_KEY"] = "test-secret"
|
|
app.register_blueprint(context_overflow_bp)
|
|
|
|
with app.test_client() as client:
|
|
response = client.get("/api/context-overflow/metrics")
|
|
assert response.status_code == 404, response.status_code
|
|
|
|
def test_research_context_overflow_route_requires_auth(self):
|
|
"""The research-context-overflow route is wrapped in
|
|
@login_required; with no auth set up, the decorator must
|
|
short-circuit with 401 (not 200, not 404). Note: the route
|
|
URL is /api/research/<id>/context-overflow (not the inverse
|
|
ordering that the previous test name suggested)."""
|
|
from flask import Flask
|
|
from local_deep_research.web.routes.context_overflow_api import (
|
|
context_overflow_bp,
|
|
)
|
|
|
|
app = Flask(__name__)
|
|
app.config["SECRET_KEY"] = "test-secret"
|
|
app.register_blueprint(context_overflow_bp)
|
|
|
|
with app.test_client() as client:
|
|
response = client.get("/api/research/123/context-overflow")
|
|
assert response.status_code == 401, response.status_code
|
|
|
|
|
|
class TestContextOverflowBlueprintImport:
|
|
"""Tests for context overflow API blueprint import."""
|
|
|
|
def test_blueprint_exists(self):
|
|
"""Test that context overflow API blueprint exists."""
|
|
from local_deep_research.web.routes.context_overflow_api import (
|
|
context_overflow_bp,
|
|
)
|
|
|
|
assert context_overflow_bp is not None
|
|
assert context_overflow_bp.name == "context_overflow_api"
|
|
|
|
|
|
class TestContextUtilizationCalculation:
|
|
"""Tests for context utilization calculation."""
|
|
|
|
def test_calculate_context_utilization_percentage(self):
|
|
"""Test context utilization percentage calculation."""
|
|
# audit: PUNCHLIST reviewed 2026-05 — issue resolved by prior PR (recommendation: DELETE).
|
|
prompt_tokens = 3000
|
|
context_limit = 4096
|
|
|
|
utilization = (prompt_tokens / context_limit) * 100
|
|
|
|
assert utilization == pytest.approx(73.24, rel=0.01)
|
|
|
|
def test_calculate_context_utilization_at_limit(self):
|
|
"""Test context utilization at 100%."""
|
|
# audit: PUNCHLIST reviewed 2026-05 — issue resolved by prior PR (recommendation: DELETE).
|
|
prompt_tokens = 4096
|
|
context_limit = 4096
|
|
|
|
utilization = (prompt_tokens / context_limit) * 100
|
|
|
|
assert utilization == 100.0
|
|
|
|
def test_calculate_context_utilization_over_limit(self):
|
|
"""Test context utilization over 100% (truncation case)."""
|
|
# audit: PUNCHLIST reviewed 2026-05 — issue resolved by prior PR (recommendation: DELETE).
|
|
prompt_tokens = 5000
|
|
context_limit = 4096
|
|
|
|
utilization = (prompt_tokens / context_limit) * 100
|
|
|
|
assert utilization > 100.0
|
|
assert utilization == pytest.approx(122.07, rel=0.01)
|
|
|
|
|
|
class TestAverageCalculations:
|
|
"""Tests for average context calculations."""
|
|
|
|
def test_calculate_average_prompt_tokens(self):
|
|
"""Test average prompt tokens calculation."""
|
|
# audit: PUNCHLIST reviewed 2026-05 — issue resolved by prior PR (recommendation: DELETE).
|
|
mock_usages = [
|
|
Mock(prompt_tokens=1000),
|
|
Mock(prompt_tokens=2000),
|
|
Mock(prompt_tokens=3000),
|
|
]
|
|
|
|
total = sum(u.prompt_tokens for u in mock_usages)
|
|
average = total / len(mock_usages)
|
|
|
|
assert average == 2000.0
|
|
|
|
def test_calculate_average_with_empty_list(self):
|
|
"""Test average calculation with empty list."""
|
|
# audit: PUNCHLIST reviewed 2026-05 — issue resolved by prior PR (recommendation: DELETE).
|
|
mock_usages = []
|
|
|
|
total = sum(getattr(u, "prompt_tokens", 0) for u in mock_usages)
|
|
average = total / len(mock_usages) if mock_usages else 0
|
|
|
|
assert average == 0
|
|
|
|
|
|
class TestResearchIdExtraction:
|
|
"""Tests for research ID extraction logic."""
|
|
|
|
def test_extract_unique_research_ids(self):
|
|
"""Test extracting unique research IDs from usages."""
|
|
# audit: PUNCHLIST reviewed 2026-05 — issue resolved by prior PR (recommendation: DELETE).
|
|
mock_usages = [
|
|
Mock(research_id="research1"),
|
|
Mock(research_id="research2"),
|
|
Mock(research_id="research1"), # Duplicate
|
|
Mock(research_id="research3"),
|
|
]
|
|
|
|
unique_ids = list(set(u.research_id for u in mock_usages))
|
|
|
|
assert len(unique_ids) == 3
|
|
assert "research1" in unique_ids
|
|
assert "research2" in unique_ids
|
|
assert "research3" in unique_ids
|
|
|
|
def test_extract_research_ids_with_none(self):
|
|
"""Test extracting research IDs with None values."""
|
|
# audit: PUNCHLIST reviewed 2026-05 — issue resolved by prior PR (recommendation: DELETE).
|
|
mock_usages = [
|
|
Mock(research_id="research1"),
|
|
Mock(research_id=None),
|
|
Mock(research_id="research2"),
|
|
]
|
|
|
|
unique_ids = list(
|
|
set(u.research_id for u in mock_usages if u.research_id)
|
|
)
|
|
|
|
assert len(unique_ids) == 2
|
|
assert None not in unique_ids
|
|
|
|
|
|
class TestTokenStatsAggregation:
|
|
"""Tests for token statistics aggregation."""
|
|
|
|
def test_aggregate_total_tokens_by_model(self):
|
|
"""Test aggregating total tokens by model."""
|
|
mock_usages = [
|
|
Mock(model_name="gpt-4", total_tokens=1000),
|
|
Mock(model_name="gpt-4", total_tokens=2000),
|
|
Mock(model_name="claude-3", total_tokens=1500),
|
|
]
|
|
|
|
model_totals = {}
|
|
for usage in mock_usages:
|
|
model = usage.model_name
|
|
if model not in model_totals:
|
|
model_totals[model] = 0
|
|
model_totals[model] += usage.total_tokens
|
|
|
|
assert model_totals["gpt-4"] == 3000
|
|
assert model_totals["claude-3"] == 1500
|
|
|
|
def test_aggregate_truncated_requests_by_model(self):
|
|
"""Test aggregating truncated requests by model."""
|
|
mock_usages = [
|
|
Mock(model_name="gpt-4", context_truncated=True),
|
|
Mock(model_name="gpt-4", context_truncated=False),
|
|
Mock(model_name="claude-3", context_truncated=True),
|
|
Mock(model_name="claude-3", context_truncated=True),
|
|
]
|
|
|
|
model_truncated = {}
|
|
for usage in mock_usages:
|
|
model = usage.model_name
|
|
if model not in model_truncated:
|
|
model_truncated[model] = 0
|
|
if usage.context_truncated:
|
|
model_truncated[model] += 1
|
|
|
|
assert model_truncated["gpt-4"] == 1
|
|
assert model_truncated["claude-3"] == 2
|
|
|
|
|
|
class TestModelTokenStatsFields:
|
|
"""Tests for model_token_stats fields added in context-overflow rework."""
|
|
|
|
def test_min_prompt_field_present(self):
|
|
"""min_prompt is computed from func.min(prompt_tokens)."""
|
|
mock_row = Mock()
|
|
mock_row.model_name = "llama3"
|
|
mock_row.model_provider = "ollama"
|
|
mock_row.total_requests = 5
|
|
mock_row.total_tokens = 10000
|
|
mock_row.min_prompt = 200
|
|
mock_row.avg_prompt = 1500
|
|
mock_row.max_prompt = 3000
|
|
mock_row.avg_response_time_ms = 1200
|
|
|
|
entry = {
|
|
"model": mock_row.model_name,
|
|
"provider": mock_row.model_provider,
|
|
"total_requests": mock_row.total_requests,
|
|
"total_tokens": int(mock_row.total_tokens or 0),
|
|
"min_prompt": int(mock_row.min_prompt or 0),
|
|
"avg_prompt": round(mock_row.avg_prompt or 0, 0),
|
|
"max_prompt": int(mock_row.max_prompt or 0),
|
|
"avg_response_time_ms": round(
|
|
mock_row.avg_response_time_ms or 0, 0
|
|
),
|
|
}
|
|
|
|
assert entry["min_prompt"] == 200
|
|
|
|
def test_min_prompt_defaults_to_zero(self):
|
|
"""min_prompt falls back to 0 when the row value is None."""
|
|
mock_row = Mock()
|
|
mock_row.model_name = "llama3"
|
|
mock_row.model_provider = "ollama"
|
|
mock_row.total_requests = 3
|
|
mock_row.total_tokens = 5000
|
|
mock_row.min_prompt = None
|
|
mock_row.avg_prompt = 800
|
|
mock_row.max_prompt = 2000
|
|
mock_row.avg_response_time_ms = None
|
|
|
|
entry = {
|
|
"model": mock_row.model_name,
|
|
"provider": mock_row.model_provider,
|
|
"total_requests": mock_row.total_requests,
|
|
"total_tokens": int(mock_row.total_tokens or 0),
|
|
"min_prompt": int(mock_row.min_prompt or 0),
|
|
"avg_prompt": round(mock_row.avg_prompt or 0, 0),
|
|
"max_prompt": int(mock_row.max_prompt or 0),
|
|
"avg_response_time_ms": round(
|
|
mock_row.avg_response_time_ms or 0, 0
|
|
),
|
|
}
|
|
|
|
assert entry["min_prompt"] == 0
|
|
assert entry["avg_response_time_ms"] == 0
|
|
|
|
|
|
class TestChartDataNewFields:
|
|
"""Tests for new chart_data fields: provider, research_phase, response_time_ms."""
|
|
|
|
def test_chart_data_includes_provider(self):
|
|
"""chart_data items include the provider field."""
|
|
mock_usage = Mock()
|
|
mock_usage.timestamp = datetime(
|
|
2025, 1, 1, 12, 0, 0, tzinfo=timezone.utc
|
|
)
|
|
mock_usage.research_id = "test-123"
|
|
mock_usage.prompt_tokens = 1000
|
|
mock_usage.completion_tokens = 200
|
|
mock_usage.ollama_prompt_eval_count = None
|
|
mock_usage.context_truncated = False
|
|
mock_usage.tokens_truncated = 0
|
|
mock_usage.context_limit = 4096
|
|
mock_usage.model_name = "llama3"
|
|
mock_usage.model_provider = "ollama"
|
|
mock_usage.research_phase = "analysis"
|
|
mock_usage.response_time_ms = 1500
|
|
|
|
ollama_used = mock_usage.ollama_prompt_eval_count
|
|
actual_prompt = ollama_used or mock_usage.prompt_tokens
|
|
tokens_truncated = mock_usage.tokens_truncated or 0
|
|
original_tokens = (
|
|
actual_prompt + tokens_truncated
|
|
if mock_usage.context_truncated
|
|
else actual_prompt
|
|
)
|
|
|
|
chart_item = {
|
|
"timestamp": mock_usage.timestamp.isoformat(),
|
|
"research_id": mock_usage.research_id,
|
|
"prompt_tokens": mock_usage.prompt_tokens,
|
|
"completion_tokens": mock_usage.completion_tokens,
|
|
"ollama_prompt_tokens": ollama_used,
|
|
"original_prompt_tokens": original_tokens,
|
|
"context_limit": mock_usage.context_limit,
|
|
"truncated": bool(mock_usage.context_truncated),
|
|
"tokens_truncated": tokens_truncated,
|
|
"model": mock_usage.model_name,
|
|
"provider": mock_usage.model_provider,
|
|
"research_phase": mock_usage.research_phase,
|
|
"response_time_ms": mock_usage.response_time_ms,
|
|
}
|
|
|
|
assert chart_item["provider"] == "ollama"
|
|
assert chart_item["research_phase"] == "analysis"
|
|
assert chart_item["response_time_ms"] == 1500
|
|
|
|
def test_chart_data_fields_handle_none(self):
|
|
"""chart_data new fields default correctly when None."""
|
|
mock_usage = Mock()
|
|
mock_usage.timestamp = datetime(
|
|
2025, 1, 1, 12, 0, 0, tzinfo=timezone.utc
|
|
)
|
|
mock_usage.research_id = "test-456"
|
|
mock_usage.prompt_tokens = 500
|
|
mock_usage.completion_tokens = 100
|
|
mock_usage.ollama_prompt_eval_count = None
|
|
mock_usage.context_truncated = False
|
|
mock_usage.tokens_truncated = 0
|
|
mock_usage.context_limit = None
|
|
mock_usage.model_name = "gpt-4"
|
|
mock_usage.model_provider = None
|
|
mock_usage.research_phase = None
|
|
mock_usage.response_time_ms = None
|
|
|
|
chart_item = {
|
|
"model": mock_usage.model_name,
|
|
"provider": mock_usage.model_provider,
|
|
"research_phase": mock_usage.research_phase,
|
|
"response_time_ms": mock_usage.response_time_ms,
|
|
}
|
|
|
|
assert chart_item["provider"] is None
|
|
assert chart_item["research_phase"] is None
|
|
assert chart_item["response_time_ms"] is None
|
|
|
|
|
|
class TestCurrentContextWindow:
|
|
"""Tests for current_context_window field in API response."""
|
|
|
|
def test_current_context_window_in_response(self):
|
|
"""API response includes current_context_window from SettingsManager."""
|
|
# Simulate the response construction logic
|
|
mock_context_window = 8192
|
|
|
|
response = {
|
|
"status": "success",
|
|
"overview": {},
|
|
"token_summary": {},
|
|
"pagination": {},
|
|
"current_context_window": mock_context_window,
|
|
}
|
|
|
|
assert response["current_context_window"] == 8192
|
|
|
|
def test_current_context_window_none_when_unset(self):
|
|
"""current_context_window is None when setting is not configured."""
|
|
response = {
|
|
"status": "success",
|
|
"overview": {},
|
|
"token_summary": {},
|
|
"pagination": {},
|
|
"current_context_window": None,
|
|
}
|
|
|
|
assert response["current_context_window"] is None
|