benchmark.py

#!/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())