""" Behavioral tests for benchmarks/metrics/calculation module. Tests pure logic functions: calculate_resource_metrics, calculate_combined_score, and calculate_metrics with temp file. """ import json import os import tempfile import pytest class TestCalculateResourceMetrics: """Tests for calculate_resource_metrics() function.""" def test_default_config(self): from local_deep_research.benchmarks.metrics.calculation import ( calculate_resource_metrics, ) result = calculate_resource_metrics({}) assert "resource_score" in result assert "estimated_complexity" in result def test_resource_score_between_0_and_1(self): from local_deep_research.benchmarks.metrics.calculation import ( calculate_resource_metrics, ) result = calculate_resource_metrics( {"iterations": 5, "questions_per_iteration": 5} ) assert 0.0 <= result["resource_score"] <= 1.0 def test_higher_iterations_lower_score(self): from local_deep_research.benchmarks.metrics.calculation import ( calculate_resource_metrics, ) low = calculate_resource_metrics( {"iterations": 1, "questions_per_iteration": 1} ) high = calculate_resource_metrics( {"iterations": 10, "questions_per_iteration": 10} ) assert low["resource_score"] > high["resource_score"] def test_complexity_increases_with_iterations(self): from local_deep_research.benchmarks.metrics.calculation import ( calculate_resource_metrics, ) low = calculate_resource_metrics({"iterations": 1}) high = calculate_resource_metrics({"iterations": 5}) assert high["estimated_complexity"] > low["estimated_complexity"] def test_max_results_affects_complexity(self): from local_deep_research.benchmarks.metrics.calculation import ( calculate_resource_metrics, ) low = calculate_resource_metrics({"max_results": 10}) high = calculate_resource_metrics({"max_results": 200}) assert high["estimated_complexity"] > low["estimated_complexity"] def test_default_values_used_when_not_specified(self): from local_deep_research.benchmarks.metrics.calculation import ( calculate_resource_metrics, ) result = calculate_resource_metrics({}) # Default: iterations=2, questions=2, max_results=50 # complexity = 2 * 2 * (50/50) = 4.0 assert result["estimated_complexity"] == pytest.approx(4.0) class TestCalculateCombinedScore: """Tests for calculate_combined_score() function.""" def test_default_weights(self): from local_deep_research.benchmarks.metrics.calculation import ( calculate_combined_score, ) metrics = { "quality": {"quality_score": 1.0}, "speed": {"speed_score": 1.0}, "resource": {"resource_score": 1.0}, } score = calculate_combined_score(metrics) assert score == pytest.approx(1.0) def test_zero_metrics(self): from local_deep_research.benchmarks.metrics.calculation import ( calculate_combined_score, ) metrics = { "quality": {"quality_score": 0.0}, "speed": {"speed_score": 0.0}, "resource": {"resource_score": 0.0}, } score = calculate_combined_score(metrics) assert score == 0.0 def test_quality_weighted_most(self): """Quality has default weight 0.6, so it dominates the score.""" from local_deep_research.benchmarks.metrics.calculation import ( calculate_combined_score, ) quality_only = { "quality": {"quality_score": 1.0}, "speed": {"speed_score": 0.0}, "resource": {"resource_score": 0.0}, } speed_only = { "quality": {"quality_score": 0.0}, "speed": {"speed_score": 1.0}, "resource": {"resource_score": 0.0}, } q_score = calculate_combined_score(quality_only) s_score = calculate_combined_score(speed_only) assert q_score > s_score def test_custom_weights(self): from local_deep_research.benchmarks.metrics.calculation import ( calculate_combined_score, ) metrics = { "quality": {"quality_score": 0.5}, "speed": {"speed_score": 0.8}, } score = calculate_combined_score( metrics, weights={"quality": 0.5, "speed": 0.5} ) # (0.5 * 0.5 + 0.8 * 0.5) / 1.0 = 0.65 assert score == pytest.approx(0.65) def test_zero_total_weight_returns_zero(self): from local_deep_research.benchmarks.metrics.calculation import ( calculate_combined_score, ) score = calculate_combined_score({}, weights={}) assert score == 0.0 def test_empty_metrics(self): from local_deep_research.benchmarks.metrics.calculation import ( calculate_combined_score, ) score = calculate_combined_score({}) assert score == 0.0 def test_partial_metrics(self): from local_deep_research.benchmarks.metrics.calculation import ( calculate_combined_score, ) metrics = {"quality": {"quality_score": 0.8}} score = calculate_combined_score(metrics) # Default weights: quality=0.6, speed=0.3, resource=0.1 # Only quality contributes: 0.8 * 0.6 = 0.48 assert score == pytest.approx(0.48) class TestCalculateMetrics: """Tests for calculate_metrics() function with temp files.""" def test_empty_file_returns_error(self): from local_deep_research.benchmarks.metrics.calculation import ( calculate_metrics, ) with tempfile.NamedTemporaryFile( mode="w", suffix=".jsonl", delete=False ) as f: f.write("") f.flush() result = calculate_metrics(f.name) os.unlink(f.name) assert "error" in result def test_single_correct_result(self): from local_deep_research.benchmarks.metrics.calculation import ( calculate_metrics, ) with tempfile.NamedTemporaryFile( mode="w", suffix=".jsonl", delete=False ) as f: f.write( json.dumps({"is_correct": True, "processing_time": 5.0}) + "\n" ) f.flush() result = calculate_metrics(f.name) os.unlink(f.name) assert result["accuracy"] == 1.0 assert result["total_examples"] == 1 assert result["correct"] == 1 def test_mixed_results_accuracy(self): from local_deep_research.benchmarks.metrics.calculation import ( calculate_metrics, ) with tempfile.NamedTemporaryFile( mode="w", suffix=".jsonl", delete=False ) as f: f.write(json.dumps({"is_correct": True}) + "\n") f.write(json.dumps({"is_correct": False}) + "\n") f.write(json.dumps({"is_correct": True}) + "\n") f.flush() result = calculate_metrics(f.name) os.unlink(f.name) assert result["accuracy"] == pytest.approx(2 / 3) def test_processing_time_average(self): from local_deep_research.benchmarks.metrics.calculation import ( calculate_metrics, ) with tempfile.NamedTemporaryFile( mode="w", suffix=".jsonl", delete=False ) as f: f.write(json.dumps({"processing_time": 10.0}) + "\n") f.write(json.dumps({"processing_time": 20.0}) + "\n") f.flush() result = calculate_metrics(f.name) os.unlink(f.name) assert result["average_processing_time"] == pytest.approx(15.0) def test_error_count(self): from local_deep_research.benchmarks.metrics.calculation import ( calculate_metrics, ) with tempfile.NamedTemporaryFile( mode="w", suffix=".jsonl", delete=False ) as f: f.write(json.dumps({"error": "something failed"}) + "\n") f.write(json.dumps({"is_correct": True}) + "\n") f.flush() result = calculate_metrics(f.name) os.unlink(f.name) assert result["error_count"] == 1 assert result["error_rate"] == pytest.approx(0.5) def test_category_metrics(self): from local_deep_research.benchmarks.metrics.calculation import ( calculate_metrics, ) with tempfile.NamedTemporaryFile( mode="w", suffix=".jsonl", delete=False ) as f: f.write( json.dumps({"is_correct": True, "category": "science"}) + "\n" ) f.write( json.dumps({"is_correct": False, "category": "science"}) + "\n" ) f.write( json.dumps({"is_correct": True, "category": "history"}) + "\n" ) f.flush() result = calculate_metrics(f.name) os.unlink(f.name) assert "categories" in result assert result["categories"]["science"]["accuracy"] == pytest.approx(0.5) assert result["categories"]["history"]["accuracy"] == 1.0 def test_nonexistent_file_returns_error(self): from local_deep_research.benchmarks.metrics.calculation import ( calculate_metrics, ) result = calculate_metrics("/nonexistent/path/file.jsonl") assert "error" in result def test_confidence_average(self): from local_deep_research.benchmarks.metrics.calculation import ( calculate_metrics, ) with tempfile.NamedTemporaryFile( mode="w", suffix=".jsonl", delete=False ) as f: f.write(json.dumps({"confidence": 80}) + "\n") f.write(json.dumps({"confidence": 60}) + "\n") f.flush() result = calculate_metrics(f.name) os.unlink(f.name) assert result["average_confidence"] == pytest.approx(70.0) def test_result_has_timestamp(self): from local_deep_research.benchmarks.metrics.calculation import ( calculate_metrics, ) with tempfile.NamedTemporaryFile( mode="w", suffix=".jsonl", delete=False ) as f: f.write(json.dumps({"is_correct": True}) + "\n") f.flush() result = calculate_metrics(f.name) os.unlink(f.name) assert "timestamp" in result