| |
| |
| |
|
|
| import asyncio |
| import json |
| import sys |
| import argparse |
| import time |
| import websockets |
| from typing import Dict, Any, List, Optional |
|
|
| class Benchmark: |
| def __init__(self, url: str, token: str): |
| self.url = url |
| self.token = token |
| self.results = [] |
| self.ws = None |
|
|
| async def connect(self): |
| self.ws = await websockets.connect(self.url) |
| await self.ws.send(json.dumps({"token": self.token})) |
| try: |
| response = await asyncio.wait_for(self.ws.recv(), timeout=10) |
| except asyncio.TimeoutError: |
| raise Exception("Timeout d'authentification – le serveur n'a pas répondu") |
| data = json.loads(response) |
| if data.get("type") != "auth_ok": |
| raise Exception(f"Authentification échouée: {data}") |
| print(f"✅ Connecté à {self.url} (authentifié)") |
|
|
| async def send_task(self, task: str, timeout: int = 120) -> Dict: |
| start = time.time() |
| await self.ws.send(json.dumps({ |
| "action": "agent", |
| "task": task, |
| "use_reasoning": False |
| })) |
| while True: |
| try: |
| raw = await asyncio.wait_for(self.ws.recv(), timeout=timeout) |
| except asyncio.TimeoutError: |
| return {"success": False, "error": "Timeout", "duration": time.time() - start} |
| except websockets.ConnectionClosed: |
| return {"success": False, "error": "Connexion fermée", "duration": time.time() - start} |
| try: |
| msg = json.loads(raw) |
| except json.JSONDecodeError: |
| continue |
| if msg.get("type") == "agent_result": |
| msg["duration"] = time.time() - start |
| return msg.get("data", {}) |
| elif msg.get("type") == "error": |
| return {"success": False, "error": msg.get("message"), "duration": time.time() - start} |
|
|
| async def run_benchmark(self): |
| tests = [ |
| { |
| "name": "1. Créer hello.py et exécuter", |
| "task": "Crée un fichier hello.py avec un print('Bonjour') et exécute-le.", |
| "check": lambda r: "Bonjour" in str(r.get("result", "")) |
| }, |
| { |
| "name": "2. Corriger division par zéro", |
| "task": "Corrige calculator.py (divide) pour retourner 'Error' si b=0, et teste 10/0.", |
| "check": lambda r: "Error" in str(r.get("result", "")) |
| }, |
| { |
| "name": "3. Créer un test unitaire avec pytest", |
| "task": "Crée test_calculator.py qui teste divide et exécute pytest.", |
| "check": lambda r: "passed" in str(r.get("result", "")).lower() |
| }, |
| { |
| "name": "4. Optimiser sum_to_n", |
| "task": "Optimise sum_to_n(n) en remplaçant la boucle par la formule n*(n+1)//2.", |
| "check": lambda r: "optimized" in str(r.get("result", "")).lower() |
| }, |
| { |
| "name": "5. Créer un projet multi-fichiers", |
| "task": "Crée utils.py (add, sub), operations.py (apply), main.py qui utilise apply.", |
| "check": lambda r: "main.py" in str(r.get("result", "")) and "utils.py" in str(r.get("result", "")) |
| }, |
| { |
| "name": "6. Gérer une dépendance manquante", |
| "task": "Crée helper.py avec help() retournant 'OK', et modifie app.py pour l'importer.", |
| "check": lambda r: "helper.py" in str(r.get("result", "")) or "OK" in str(r.get("result", "")) |
| }, |
| { |
| "name": "7. Résoudre une dépendance circulaire", |
| "task": "Crée deux fichiers a.py et b.py avec une importation circulaire, puis corrige-la.", |
| "check": lambda r: "a.py" in str(r.get("result", "")) or "b.py" in str(r.get("result", "")) |
| }, |
| { |
| "name": "8. Refactoring (extraire une fonction)", |
| "task": "Refactorise app.py en extrayant une fonction process_data dans business.py.", |
| "check": lambda r: "business.py" in str(r.get("result", "")) |
| }, |
| { |
| "name": "9. Renommer une fonction dans tout le projet", |
| "task": "Renomme la fonction 'old_name' en 'new_name' dans tous les fichiers et exécute les tests.", |
| "check": lambda r: "new_name" in str(r.get("result", "")) |
| }, |
| { |
| "name": "10. Créer un projet complexe (3 modules + tests)", |
| "task": "Crée un projet avec un module de calcul, un module de traitement, un module principal, et des tests unitaires.", |
| "check": lambda r: "test" in str(r.get("result", "")) and "calcul" in str(r.get("result", "")) |
| } |
| ] |
|
|
| for test in tests: |
| print(f"\n🧪 {test['name']}") |
| print(f" Tâche: {test['task'][:80]}...") |
| try: |
| result = await self.send_task(test["task"]) |
| passed = test["check"](result) if callable(test["check"]) else result.get("success", False) |
| status = "✅ PASS" if passed else "❌ FAIL" |
| print(f" {status}") |
| print(f" Durée: {result.get('duration', 0):.2f}s") |
| print(f" Résultat: {json.dumps(result.get('result', ''))[:150]}...") |
| self.results.append({ |
| "name": test["name"], |
| "passed": passed, |
| "duration": result.get("duration", 0), |
| "result": result |
| }) |
| except Exception as e: |
| print(f" ❌ ERREUR: {e}") |
| self.results.append({"name": test["name"], "passed": False, "error": str(e)}) |
|
|
| def report(self): |
| total = len(self.results) |
| passed = sum(1 for r in self.results if r.get("passed", False)) |
| total_time = sum(r.get("duration", 0) for r in self.results if r.get("duration")) |
| print("\n" + "=" * 70) |
| print(f"📊 RAPPORT BENCHMARK – {total} tâches") |
| for r in self.results: |
| status = "✅" if r.get("passed") else "❌" |
| dur = f"{r.get('duration', 0):.2f}s" if r.get('duration') else "N/A" |
| print(f" {status} {r['name']} ({dur})") |
| print(f"\n ✅ Réussis: {passed}/{total}") |
| print(f" ❌ Échecs: {total - passed}/{total}") |
| if total_time > 0: |
| print(f" ⏱️ Temps total: {total_time:.2f}s") |
| print("=" * 70) |
| return passed == total |
|
|
| async def main(): |
| parser = argparse.ArgumentParser(description="Benchmark pour ADN NEXUS") |
| parser.add_argument("--token", required=True, help="Token d'accès à l'IDE") |
| parser.add_argument("--url", required=True, help="URL du WebSocket (ex: wss://MON-SPACE.hf.space/ws)") |
| args = parser.parse_args() |
|
|
| bench = Benchmark(args.url, args.token) |
| await bench.connect() |
| await bench.run_benchmark() |
| all_passed = bench.report() |
| sys.exit(0 if all_passed else 1) |
|
|
| if __name__ == "__main__": |
| asyncio.run(main()) |
