204 lines
6.6 KiB
Python
204 lines
6.6 KiB
Python
"""
|
|
Test per verificare la connessione con il server radar.
|
|
Questo test controlla se il radar sta trasmettendo messaggi sulla porta corretta.
|
|
"""
|
|
import socket
|
|
import time
|
|
import struct
|
|
import pytest
|
|
|
|
|
|
def test_radar_sending_on_port_61553():
|
|
"""
|
|
Verifica che il radar stia inviando dati sulla porta 61553.
|
|
Questo test ascolta per 5 secondi e verifica se arrivano pacchetti UDP.
|
|
"""
|
|
port = 61553
|
|
timeout_seconds = 5
|
|
|
|
sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
|
|
sock.settimeout(1.0)
|
|
|
|
try:
|
|
sock.bind(("0.0.0.0", port))
|
|
print(f"\n✓ Socket bind riuscito su porta {port}")
|
|
except OSError as e:
|
|
pytest.skip(f"Porta {port} già in uso o non disponibile: {e}")
|
|
return
|
|
|
|
packets_received = 0
|
|
start_time = time.time()
|
|
received_data = []
|
|
|
|
print(f"Ascolto sulla porta {port} per {timeout_seconds} secondi...")
|
|
|
|
while time.time() - start_time < timeout_seconds:
|
|
try:
|
|
data, addr = sock.recvfrom(4096)
|
|
packets_received += 1
|
|
print(f" Pacchetto #{packets_received} ricevuto da {addr} ({len(data)} bytes)")
|
|
|
|
# Salva i primi 64 bytes per l'analisi
|
|
if len(received_data) < 3:
|
|
received_data.append({
|
|
'addr': addr,
|
|
'size': len(data),
|
|
'data': data[:64],
|
|
'hex': data[:64].hex().upper()
|
|
})
|
|
|
|
except socket.timeout:
|
|
continue
|
|
|
|
sock.close()
|
|
|
|
# Stampa dettagli dei pacchetti ricevuti
|
|
if packets_received > 0:
|
|
print(f"\n✓ SUCCESSO: Ricevuti {packets_received} pacchetti dal radar")
|
|
print("\nDettagli primi pacchetti:")
|
|
for i, pkt in enumerate(received_data, 1):
|
|
print(f"\n Pacchetto #{i}:")
|
|
print(f" Da: {pkt['addr']}")
|
|
print(f" Dimensione: {pkt['size']} bytes")
|
|
print(f" Hex (primi 64 bytes): {pkt['hex']}")
|
|
|
|
# Verifica marker 1553 (0x1553 o 0x5315 in little-endian)
|
|
if len(pkt['data']) >= 2:
|
|
marker = struct.unpack('<H', pkt['data'][:2])[0]
|
|
if marker == 0x1553:
|
|
print(f" ✓ Marker 1553 valido trovato: 0x{marker:04X}")
|
|
else:
|
|
print(f" ⚠ Marker non riconosciuto: 0x{marker:04X}")
|
|
else:
|
|
print(f"\n✗ FALLITO: Nessun pacchetto ricevuto in {timeout_seconds} secondi")
|
|
print("\nPossibili cause:")
|
|
print(" 1. Il radar non sta trasmettendo")
|
|
print(" 2. Il radar sta trasmettendo su una porta diversa")
|
|
print(" 3. Firewall Windows sta bloccando il traffico")
|
|
print(" 4. Il radar sta trasmettendo a un IP diverso")
|
|
print("\nSuggerimenti:")
|
|
print(f" - Verifica che il radar sia in esecuzione")
|
|
print(f" - Esegui: python tools\\scan_udp_ports.py")
|
|
print(f" - Esegui: python tools\\sniff_udp_traffic.py")
|
|
|
|
# Il test passa se ha ricevuto almeno un pacchetto
|
|
assert packets_received > 0, (
|
|
f"Nessun pacchetto ricevuto sulla porta {port} in {timeout_seconds} secondi. "
|
|
"Verifica che il radar sia in esecuzione e stia trasmettendo."
|
|
)
|
|
|
|
|
|
def test_radar_connection_with_multiple_ports():
|
|
"""
|
|
Scansiona più porte contemporaneamente per trovare dove sta trasmettendo il radar.
|
|
"""
|
|
ports_to_check = [51553, 61553, 1553, 15530, 5000, 6000]
|
|
timeout_seconds = 3
|
|
|
|
import threading
|
|
import queue
|
|
|
|
results = queue.Queue()
|
|
|
|
def listen_on_port(port):
|
|
sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
|
|
sock.settimeout(0.5)
|
|
|
|
try:
|
|
sock.bind(("0.0.0.0", port))
|
|
except OSError:
|
|
results.put(('bind_fail', port, None))
|
|
return
|
|
|
|
start = time.time()
|
|
packets = 0
|
|
|
|
while time.time() - start < timeout_seconds:
|
|
try:
|
|
data, addr = sock.recvfrom(4096)
|
|
packets += 1
|
|
if packets == 1: # Solo il primo pacchetto
|
|
results.put(('packet', port, addr, len(data), data[:32]))
|
|
except socket.timeout:
|
|
continue
|
|
|
|
if packets == 0:
|
|
results.put(('no_data', port, None))
|
|
|
|
sock.close()
|
|
|
|
# Avvia thread per ogni porta
|
|
threads = []
|
|
for port in ports_to_check:
|
|
t = threading.Thread(target=listen_on_port, args=(port,), daemon=True)
|
|
t.start()
|
|
threads.append(t)
|
|
time.sleep(0.01)
|
|
|
|
# Aspetta che tutti i thread finiscano
|
|
for t in threads:
|
|
t.join(timeout=timeout_seconds + 1)
|
|
|
|
# Analizza risultati
|
|
ports_with_data = []
|
|
ports_no_data = []
|
|
ports_bind_fail = []
|
|
|
|
while not results.empty():
|
|
result = results.get()
|
|
if result[0] == 'packet':
|
|
ports_with_data.append(result)
|
|
elif result[0] == 'no_data':
|
|
ports_no_data.append(result[1])
|
|
elif result[0] == 'bind_fail':
|
|
ports_bind_fail.append(result[1])
|
|
|
|
print(f"\n=== Scansione Multi-Porta ===")
|
|
print(f"Porte testate: {ports_to_check}")
|
|
|
|
if ports_with_data:
|
|
print(f"\n✓ Porte con traffico ricevuto:")
|
|
for _, port, addr, size, data in ports_with_data:
|
|
print(f" - Porta {port}: Pacchetto da {addr} ({size} bytes)")
|
|
print(f" Hex: {data.hex().upper()}")
|
|
|
|
if ports_no_data:
|
|
print(f"\n○ Porte senza traffico: {ports_no_data}")
|
|
|
|
if ports_bind_fail:
|
|
print(f"\n✗ Porte già in uso: {ports_bind_fail}")
|
|
|
|
# Il test passa se almeno una porta ha ricevuto dati
|
|
if not ports_with_data:
|
|
pytest.skip(
|
|
f"Nessuna porta ha ricevuto traffico. "
|
|
f"Verifica che il radar sia in esecuzione."
|
|
)
|
|
|
|
assert len(ports_with_data) > 0, "Dovrebbe esserci traffico su almeno una porta"
|
|
|
|
|
|
if __name__ == "__main__":
|
|
"""
|
|
Esegui direttamente questo script per un test rapido senza pytest.
|
|
"""
|
|
print("=== Test Connessione Radar ===\n")
|
|
|
|
try:
|
|
print("Test 1: Verifica porta 61553...")
|
|
test_radar_sending_on_port_61553()
|
|
print("\n✓ Test 1 PASSATO\n")
|
|
except AssertionError as e:
|
|
print(f"\n✗ Test 1 FALLITO: {e}\n")
|
|
except Exception as e:
|
|
print(f"\n✗ Test 1 ERRORE: {e}\n")
|
|
|
|
print("="*60)
|
|
|
|
try:
|
|
print("\nTest 2: Scansione multi-porta...")
|
|
test_radar_connection_with_multiple_ports()
|
|
print("\n✓ Test 2 PASSATO\n")
|
|
except Exception as e:
|
|
print(f"\n⚠ Test 2: {e}\n")
|