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
371 lines
12 KiB
Python
371 lines
12 KiB
Python
"""
|
|
Behavioral tests for metrics/token_counter module.
|
|
|
|
Tests TokenCountingCallback initialization, in-memory counts tracking,
|
|
and context overflow field generation (pure logic, no DB).
|
|
"""
|
|
|
|
import pytest
|
|
|
|
|
|
class TestTokenCountingCallbackInit:
|
|
"""Tests for TokenCountingCallback initialization."""
|
|
|
|
def test_default_research_id_is_none(self):
|
|
"""Default research_id is None."""
|
|
from local_deep_research.metrics.token_counter import (
|
|
TokenCountingCallback,
|
|
)
|
|
|
|
cb = TokenCountingCallback()
|
|
assert cb.research_id is None
|
|
|
|
def test_custom_research_id(self):
|
|
"""Accepts custom research_id."""
|
|
from local_deep_research.metrics.token_counter import (
|
|
TokenCountingCallback,
|
|
)
|
|
|
|
cb = TokenCountingCallback(research_id="test-123")
|
|
assert cb.research_id == "test-123"
|
|
|
|
def test_default_research_context_is_empty_dict(self):
|
|
"""Default research_context is empty dict."""
|
|
from local_deep_research.metrics.token_counter import (
|
|
TokenCountingCallback,
|
|
)
|
|
|
|
cb = TokenCountingCallback()
|
|
assert cb.research_context == {}
|
|
|
|
def test_custom_research_context(self):
|
|
"""Accepts custom research_context."""
|
|
from local_deep_research.metrics.token_counter import (
|
|
TokenCountingCallback,
|
|
)
|
|
|
|
ctx = {"query": "test", "mode": "quick"}
|
|
cb = TokenCountingCallback(research_context=ctx)
|
|
assert cb.research_context == ctx
|
|
|
|
def test_initial_model_is_none(self):
|
|
"""Initial current_model is None."""
|
|
from local_deep_research.metrics.token_counter import (
|
|
TokenCountingCallback,
|
|
)
|
|
|
|
cb = TokenCountingCallback()
|
|
assert cb.current_model is None
|
|
|
|
def test_initial_provider_is_none(self):
|
|
"""Initial current_provider is None."""
|
|
from local_deep_research.metrics.token_counter import (
|
|
TokenCountingCallback,
|
|
)
|
|
|
|
cb = TokenCountingCallback()
|
|
assert cb.current_provider is None
|
|
|
|
def test_initial_start_time_is_none(self):
|
|
"""Initial start_time is None."""
|
|
from local_deep_research.metrics.token_counter import (
|
|
TokenCountingCallback,
|
|
)
|
|
|
|
cb = TokenCountingCallback()
|
|
assert cb.start_time is None
|
|
|
|
def test_initial_success_status(self):
|
|
"""Initial success_status is 'success'."""
|
|
from local_deep_research.metrics.token_counter import (
|
|
TokenCountingCallback,
|
|
)
|
|
|
|
cb = TokenCountingCallback()
|
|
assert cb.success_status == "success"
|
|
|
|
def test_initial_error_type_is_none(self):
|
|
"""Initial error_type is None."""
|
|
from local_deep_research.metrics.token_counter import (
|
|
TokenCountingCallback,
|
|
)
|
|
|
|
cb = TokenCountingCallback()
|
|
assert cb.error_type is None
|
|
|
|
def test_context_truncated_default_false(self):
|
|
"""context_truncated defaults to False."""
|
|
from local_deep_research.metrics.token_counter import (
|
|
TokenCountingCallback,
|
|
)
|
|
|
|
cb = TokenCountingCallback()
|
|
assert cb.context_truncated is False
|
|
|
|
def test_tokens_truncated_default_zero(self):
|
|
"""tokens_truncated defaults to 0."""
|
|
from local_deep_research.metrics.token_counter import (
|
|
TokenCountingCallback,
|
|
)
|
|
|
|
cb = TokenCountingCallback()
|
|
assert cb.tokens_truncated == 0
|
|
|
|
def test_truncation_ratio_default_zero(self):
|
|
"""truncation_ratio defaults to 0.0."""
|
|
from local_deep_research.metrics.token_counter import (
|
|
TokenCountingCallback,
|
|
)
|
|
|
|
cb = TokenCountingCallback()
|
|
assert cb.truncation_ratio == 0.0
|
|
|
|
def test_ollama_metrics_empty_dict(self):
|
|
"""ollama_metrics defaults to empty dict."""
|
|
from local_deep_research.metrics.token_counter import (
|
|
TokenCountingCallback,
|
|
)
|
|
|
|
cb = TokenCountingCallback()
|
|
assert cb.ollama_metrics == {}
|
|
|
|
|
|
class TestTokenCountingCallbackCounts:
|
|
"""Tests for TokenCountingCallback in-memory counts structure."""
|
|
|
|
def test_counts_is_dict(self):
|
|
"""counts is a dictionary."""
|
|
from local_deep_research.metrics.token_counter import (
|
|
TokenCountingCallback,
|
|
)
|
|
|
|
cb = TokenCountingCallback()
|
|
assert isinstance(cb.counts, dict)
|
|
|
|
def test_counts_has_total_tokens(self):
|
|
"""counts has total_tokens key."""
|
|
from local_deep_research.metrics.token_counter import (
|
|
TokenCountingCallback,
|
|
)
|
|
|
|
cb = TokenCountingCallback()
|
|
assert "total_tokens" in cb.counts
|
|
assert cb.counts["total_tokens"] == 0
|
|
|
|
def test_counts_has_total_prompt_tokens(self):
|
|
"""counts has total_prompt_tokens key."""
|
|
from local_deep_research.metrics.token_counter import (
|
|
TokenCountingCallback,
|
|
)
|
|
|
|
cb = TokenCountingCallback()
|
|
assert "total_prompt_tokens" in cb.counts
|
|
assert cb.counts["total_prompt_tokens"] == 0
|
|
|
|
def test_counts_has_total_completion_tokens(self):
|
|
"""counts has total_completion_tokens key."""
|
|
from local_deep_research.metrics.token_counter import (
|
|
TokenCountingCallback,
|
|
)
|
|
|
|
cb = TokenCountingCallback()
|
|
assert "total_completion_tokens" in cb.counts
|
|
assert cb.counts["total_completion_tokens"] == 0
|
|
|
|
def test_counts_has_by_model(self):
|
|
"""counts has by_model key."""
|
|
from local_deep_research.metrics.token_counter import (
|
|
TokenCountingCallback,
|
|
)
|
|
|
|
cb = TokenCountingCallback()
|
|
assert "by_model" in cb.counts
|
|
assert cb.counts["by_model"] == {}
|
|
|
|
def test_get_counts_returns_counts(self):
|
|
"""get_counts returns the counts dict."""
|
|
from local_deep_research.metrics.token_counter import (
|
|
TokenCountingCallback,
|
|
)
|
|
|
|
cb = TokenCountingCallback()
|
|
result = cb.get_counts()
|
|
assert result is cb.counts
|
|
|
|
|
|
class TestContextOverflowFields:
|
|
"""Tests for _get_context_overflow_fields method."""
|
|
|
|
def test_returns_dict(self):
|
|
"""Returns a dictionary."""
|
|
from local_deep_research.metrics.token_counter import (
|
|
TokenCountingCallback,
|
|
)
|
|
|
|
cb = TokenCountingCallback()
|
|
result = cb._get_context_overflow_fields()
|
|
assert isinstance(result, dict)
|
|
|
|
def test_includes_context_limit(self):
|
|
"""Includes context_limit field."""
|
|
from local_deep_research.metrics.token_counter import (
|
|
TokenCountingCallback,
|
|
)
|
|
|
|
cb = TokenCountingCallback()
|
|
result = cb._get_context_overflow_fields()
|
|
assert "context_limit" in result
|
|
|
|
def test_includes_context_truncated(self):
|
|
"""Includes context_truncated field."""
|
|
from local_deep_research.metrics.token_counter import (
|
|
TokenCountingCallback,
|
|
)
|
|
|
|
cb = TokenCountingCallback()
|
|
result = cb._get_context_overflow_fields()
|
|
assert "context_truncated" in result
|
|
|
|
def test_default_context_limit_is_none(self):
|
|
"""Default context_limit is None."""
|
|
from local_deep_research.metrics.token_counter import (
|
|
TokenCountingCallback,
|
|
)
|
|
|
|
cb = TokenCountingCallback()
|
|
result = cb._get_context_overflow_fields()
|
|
assert result["context_limit"] is None
|
|
|
|
def test_default_context_truncated_is_false(self):
|
|
"""Default context_truncated is False."""
|
|
from local_deep_research.metrics.token_counter import (
|
|
TokenCountingCallback,
|
|
)
|
|
|
|
cb = TokenCountingCallback()
|
|
result = cb._get_context_overflow_fields()
|
|
assert result["context_truncated"] is False
|
|
|
|
def test_tokens_truncated_none_when_not_truncated(self):
|
|
"""tokens_truncated is None when not truncated."""
|
|
from local_deep_research.metrics.token_counter import (
|
|
TokenCountingCallback,
|
|
)
|
|
|
|
cb = TokenCountingCallback()
|
|
result = cb._get_context_overflow_fields()
|
|
assert result["tokens_truncated"] is None
|
|
|
|
def test_truncation_ratio_none_when_not_truncated(self):
|
|
"""truncation_ratio is None when not truncated."""
|
|
from local_deep_research.metrics.token_counter import (
|
|
TokenCountingCallback,
|
|
)
|
|
|
|
cb = TokenCountingCallback()
|
|
result = cb._get_context_overflow_fields()
|
|
assert result["truncation_ratio"] is None
|
|
|
|
def test_tokens_truncated_present_when_truncated(self):
|
|
"""tokens_truncated has value when truncated."""
|
|
from local_deep_research.metrics.token_counter import (
|
|
TokenCountingCallback,
|
|
)
|
|
|
|
cb = TokenCountingCallback()
|
|
cb.context_truncated = True
|
|
cb.tokens_truncated = 500
|
|
result = cb._get_context_overflow_fields()
|
|
assert result["tokens_truncated"] == 500
|
|
|
|
def test_truncation_ratio_present_when_truncated(self):
|
|
"""truncation_ratio has value when truncated."""
|
|
from local_deep_research.metrics.token_counter import (
|
|
TokenCountingCallback,
|
|
)
|
|
|
|
cb = TokenCountingCallback()
|
|
cb.context_truncated = True
|
|
cb.truncation_ratio = 0.25
|
|
result = cb._get_context_overflow_fields()
|
|
assert result["truncation_ratio"] == pytest.approx(0.25)
|
|
|
|
def test_includes_ollama_metrics_fields(self):
|
|
"""Includes all ollama metrics fields."""
|
|
from local_deep_research.metrics.token_counter import (
|
|
TokenCountingCallback,
|
|
)
|
|
|
|
cb = TokenCountingCallback()
|
|
result = cb._get_context_overflow_fields()
|
|
assert "ollama_prompt_eval_count" in result
|
|
assert "ollama_eval_count" in result
|
|
assert "ollama_total_duration" in result
|
|
assert "ollama_load_duration" in result
|
|
assert "ollama_prompt_eval_duration" in result
|
|
assert "ollama_eval_duration" in result
|
|
|
|
def test_ollama_metrics_populated(self):
|
|
"""Ollama metrics reflect stored values."""
|
|
from local_deep_research.metrics.token_counter import (
|
|
TokenCountingCallback,
|
|
)
|
|
|
|
cb = TokenCountingCallback()
|
|
cb.ollama_metrics = {
|
|
"prompt_eval_count": 100,
|
|
"eval_count": 50,
|
|
"total_duration": 1000000,
|
|
}
|
|
result = cb._get_context_overflow_fields()
|
|
assert result["ollama_prompt_eval_count"] == 100
|
|
assert result["ollama_eval_count"] == 50
|
|
assert result["ollama_total_duration"] == 1000000
|
|
|
|
|
|
class TestTokenCounter:
|
|
"""Tests for TokenCounter manager class."""
|
|
|
|
def test_creates_without_args(self):
|
|
"""Can be created without arguments."""
|
|
from local_deep_research.metrics.token_counter import TokenCounter
|
|
|
|
counter = TokenCounter()
|
|
assert counter is not None
|
|
|
|
def test_create_callback_returns_callback(self):
|
|
"""create_callback returns a TokenCountingCallback."""
|
|
from local_deep_research.metrics.token_counter import (
|
|
TokenCounter,
|
|
TokenCountingCallback,
|
|
)
|
|
|
|
counter = TokenCounter()
|
|
cb = counter.create_callback()
|
|
assert isinstance(cb, TokenCountingCallback)
|
|
|
|
def test_create_callback_with_research_id(self):
|
|
"""create_callback passes research_id to callback."""
|
|
from local_deep_research.metrics.token_counter import TokenCounter
|
|
|
|
counter = TokenCounter()
|
|
cb = counter.create_callback(research_id="test-456")
|
|
assert cb.research_id == "test-456"
|
|
|
|
def test_create_callback_with_context(self):
|
|
"""create_callback passes research_context to callback."""
|
|
from local_deep_research.metrics.token_counter import TokenCounter
|
|
|
|
counter = TokenCounter()
|
|
ctx = {"query": "test"}
|
|
cb = counter.create_callback(research_context=ctx)
|
|
assert cb.research_context == ctx
|
|
|
|
def test_create_callback_default_no_research_id(self):
|
|
"""create_callback defaults to no research_id."""
|
|
from local_deep_research.metrics.token_counter import TokenCounter
|
|
|
|
counter = TokenCounter()
|
|
cb = counter.create_callback()
|
|
assert cb.research_id is None
|