Spaces:
Paused
Paused
| 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() | |