#!/usr/bin/env python3 # benchmark.py – Test complet pour ADN NEXUS V10.3.2 # Usage: python benchmark.py --token VOTRE_TOKEN --url wss://MON-SPACE.hf.space/ws 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())