Files
wehub-resource-sync 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
chore: import upstream snapshot with attribution
2026-07-13 13:08:55 +08:00

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