Spaces:
Running
Running
File size: 3,991 Bytes
c8365f5 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 | import asyncio
import json
import tempfile
import unittest
from pathlib import Path
from unittest.mock import AsyncMock, patch
from api import benchmark_handler as handler
class _Process:
returncode = 0
async def communicate(self):
return b"", b""
class ExtendedBenchmarkHandlerTests(unittest.IsolatedAsyncioTestCase):
async def test_extended_runner_uses_full_mode_and_reports_20_categories(self):
categories = [
"bug_fix", "refactor", "feature", "devops", "security", "performance",
"autonomy", "code_correct", "sql", "context_window", "adversarial", "mmlu",
"reasoning", "data_analysis", "technical_writing", "research_synthesis",
"orchestration", "memory_context", "recovery", "robustness",
]
report = {
"timestamp": "2026-08-15T12:00:00Z",
"version": "extended-v5",
"summary": {"avgScore": 70, "avgReplit": 60, "avgCursor": 65, "avgDevin": 70, "avgManus": 75, "gapCount": 0, "verdict": "PARI_REPLIT"},
"tasks": [{"cat": category, "score": 70} for category in categories],
"gapCards": [],
}
replies = AsyncMock()
with tempfile.TemporaryDirectory() as directory:
script = Path(directory) / "benchmark-extended.mjs"
output = Path(directory) / "benchmark-v5-latest.json"
script.write_text("// runner")
output.write_text(json.dumps(report))
with patch.object(handler, "_BENCH_SCRIPT", str(script)), \
patch.object(handler, "_REPORT_V7", str(output)), \
patch("asyncio.create_subprocess_exec", AsyncMock(return_value=_Process())) as create_process:
await handler.run_benchmark_task(123, replies)
command = create_process.await_args.args
self.assertEqual(command[:2], ("node", str(script)))
self.assertIn("--full", command)
self.assertIn("--json", command)
self.assertIn("--gap-analysis", command)
final_text = replies.await_args_list[-1].args[1]
self.assertIn("20/20 categorie", final_text)
async def test_targeted_weak_runner_uses_only_low_score_categories_and_separate_report(self):
categories = list(handler._WEAK_CATEGORIES)
report = {
"timestamp": "2026-08-15T12:00:00Z",
"version": "extended-v5",
"summary": {"avgScore": 31, "avgReplit": 56, "avgCursor": 62, "avgDevin": 68, "avgManus": 71, "gapCount": 8, "verdict": "SOTTO_REPLIT"},
"tasks": [{"cat": category, "score": 31} for category in categories],
"gapCards": [],
}
replies = AsyncMock()
with tempfile.TemporaryDirectory() as directory:
script = Path(directory) / "benchmark-extended.mjs"
full_output = Path(directory) / "benchmark-v5-latest.json"
weak_output = Path(directory) / "benchmark-v5-weak-latest.json"
script.write_text("// runner")
full_output.write_text(json.dumps({"tasks": []}))
weak_output.write_text(json.dumps(report))
with patch.object(handler, "_BENCH_SCRIPT", str(script)), \
patch.object(handler, "_REPORT_V7", str(full_output)), \
patch.object(handler, "_REPORT_V7_WEAK", str(weak_output)), \
patch("asyncio.create_subprocess_exec", AsyncMock(return_value=_Process())) as create_process:
await handler.run_benchmark_task(123, replies, mode="weak")
command = create_process.await_args.args
self.assertNotIn("--full", command)
self.assertIn("--categories=" + ",".join(handler._WEAK_CATEGORIES), command)
self.assertIn("--output=" + str(weak_output), command)
final_text = replies.await_args_list[-1].args[1]
self.assertIn("10/10 categorie", final_text)
self.assertIn("categorie deboli", final_text)
if __name__ == "__main__":
unittest.main()
|