4NK_node/tests/external/test_dev3_connectivity.py

#!/usr/bin/env python3
"""
Script de test de connectivité WSS avec dev3.4nkweb.com
Teste la connectivité WebSocket Secure et la compatibilité avec l'ancienne version
"""

import asyncio
import websockets
import json
import ssl
import time
from datetime import datetime

# Configuration du relais dev3
DEV3_URI = "ws://dev3.4nkweb.com:8090"

class Dev3Tester:
    def __init__(self):
        self.connection_success = False
        self.handshake_success = False
        self.messages_received = []
        self.test_results = {}

    async def test_connection(self):
        """Test de connexion WSS basique"""
        print(f"🔌 Test de connexion WSS à {DEV3_URI}")

        try:
            # Tentative de connexion WebSocket simple
            websocket = await asyncio.wait_for(
                websockets.connect(DEV3_URI),
                timeout=10
            )

            self.connection_success = True
            print("✅ Connexion WSS établie")

            # Test de handshake simple (ancienne version)
            await self.test_old_version_handshake(websocket)

            # Test de handshake nouvelle version
            await self.test_new_version_handshake(websocket)

            await websocket.close()

        except asyncio.TimeoutError:
            print("❌ Timeout lors de la connexion WSS")
        except Exception as e:
            print(f"❌ Erreur de connexion WSS: {e}")

    async def test_old_version_handshake(self, websocket):
        """Test de handshake compatible ancienne version"""
        print("🧪 Test handshake ancienne version...")

        try:
            # Message de handshake simple (ancienne version)
            old_handshake = {
                "type": "handshake",
                "client_id": "test-client-old",
                "version": "0.9.0"
            }

            await websocket.send(json.dumps(old_handshake))

            # Attendre une réponse
            try:
                response = await asyncio.wait_for(websocket.recv(), timeout=5)
                data = json.loads(response)
                print(f"📨 Réponse ancienne version: {data}")

                if data.get("type") == "handshake_response":
                    self.handshake_success = True
                    print("✅ Handshake ancienne version réussi")
                    self.messages_received.append({
                        "type": "old_handshake",
                        "response": data,
                        "timestamp": datetime.now()
                    })
                else:
                    print("⚠️ Réponse inattendue pour ancienne version")

            except asyncio.TimeoutError:
                print("⚠️ Pas de réponse pour handshake ancienne version")

        except Exception as e:
            print(f"❌ Erreur handshake ancienne version: {e}")

    async def test_new_version_handshake(self, websocket):
        """Test de handshake nouvelle version avec synchronisation"""
        print("🧪 Test handshake nouvelle version...")

        try:
            # Message de handshake nouvelle version avec sync
            new_handshake = {
                "flag": "Sync",
                "content": {
                    "type": "RelaySync",
                    "relay_id": "test-client-new",
                    "timestamp": int(time.time()),
                    "sequence": 1,
                    "payload": {
                        "discovery": True,
                        "relay_info": {
                            "id": "test-client-new",
                            "capabilities": ["sync", "mesh"]
                        }
                    }
                }
            }

            await websocket.send(json.dumps(new_handshake))

            # Attendre une réponse
            try:
                response = await asyncio.wait_for(websocket.recv(), timeout=5)
                data = json.loads(response)
                print(f"📨 Réponse nouvelle version: {data}")

                self.messages_received.append({
                    "type": "new_handshake",
                    "response": data,
                    "timestamp": datetime.now()
                })

                if data.get("flag") == "Sync":
                    print("✅ Handshake nouvelle version réussi")
                else:
                    print("⚠️ Réponse inattendue pour nouvelle version")

            except asyncio.TimeoutError:
                print("⚠️ Pas de réponse pour handshake nouvelle version")

        except Exception as e:
            print(f"❌ Erreur handshake nouvelle version: {e}")

    async def test_websocket_protocol(self, websocket):
        """Test du protocole WebSocket"""
        print("🧪 Test du protocole WebSocket...")

        try:
            # Test de ping/pong
            pong_waiter = await websocket.ping()
            await pong_waiter
            print("✅ Ping/Pong fonctionne")

            # Test de fermeture propre
            await websocket.close()
            print("✅ Fermeture propre réussie")

        except Exception as e:
            print(f"❌ Erreur protocole WebSocket: {e}")

    def generate_report(self):
        """Génère un rapport de test"""
        print("\n" + "="*50)
        print("📊 RAPPORT DE TEST - dev3.4nkweb.com")
        print("="*50)

        print(f"🔌 Connexion WSS: {'✅' if self.connection_success else '❌'}")
        print(f"🤝 Handshake: {'✅' if self.handshake_success else '❌'}")
        print(f"📨 Messages reçus: {len(self.messages_received)}")

        if self.messages_received:
            print("\n📋 Messages reçus:")
            for msg in self.messages_received:
                print(f"  - {msg['type']}: {msg['response'].get('type', 'N/A')}")

        print("\n🎯 Recommandations:")
        if self.connection_success:
            print("  ✅ Le relais dev3.4nkweb.com est accessible en WSS")
            if self.handshake_success:
                print("  ✅ Compatible avec l'ancienne version")
            else:
                print("  ⚠️ Peut nécessiter une mise à jour pour la synchronisation")
        else:
            print("  ❌ Le relais n'est pas accessible")

        print("="*50)

async def main():
    """Fonction principale"""
    print("🚀 Test de connectivité WSS avec dev3.4nkweb.com")
    print("="*50)

    tester = Dev3Tester()

    # Test de connexion
    await tester.test_connection()

    # Générer le rapport
    tester.generate_report()

if __name__ == "__main__":
    asyncio.run(main())