@@ -300,188 +300,74 @@ def run_test_scenario(self, test_server, user_conn_type, engineer_conn_type,
300300
301301
302302class TestUDPConnections (BaseConnectionTest ):
303- """Test suite for UDP connection functionality using PyMAVLink
304- with authentication."""
303+ """UDP/UDP scenarios. One method per signing case so xdist can run
304+ them in parallel — each one pays the proxy's 10s conn1-idle wait,
305+ so collapsing them into a single sequential method made wall-clock
306+ time roughly the sum."""
305307
306- def test_basic_udp_message_reception (self , test_server ):
307- """Test UDP message reception with different authentication scenarios."""
308-
309- # Test scenario 1: Engineer without signed connection
310- print ("\n === TEST 1: Engineer without signed connection ===" )
311- self ._test_unsigned_engineer (test_server )
312-
313- # Test scenario 2: Engineer with bad signing key
314- print ("\n === TEST 2: Engineer with bad signing key ===" )
315- self ._test_bad_signing_key (test_server )
316-
317- # Test scenario 3: Engineer with correct signing key
318- print ("\n === TEST 3: Engineer with correct signing key ===" )
319- self ._test_good_signing_key (test_server )
320-
321- def _test_unsigned_engineer (self , test_server ):
322- """Test engineer connection without signing - should only get HEARTBEAT."""
308+ def test_unsigned_engineer (self , test_server ):
309+ """Engineer without signing - should only get HEARTBEAT."""
323310 heartbeat_count , system_time_count = self .run_test_scenario (
324311 test_server , 'udp' , 'udp' , engineer_signing_key = None , test_duration = 3
325312 )
326-
327313 assert heartbeat_count > 0 , \
328314 "Engineer should receive HEARTBEAT messages even without signing"
329315 assert system_time_count == 0 , \
330316 "Engineer should NOT receive SYSTEM_TIME without proper signing"
331- print (f"SUCCESS: Unsigned engineer received { heartbeat_count } HEARTBEAT, "
332- f"{ system_time_count } SYSTEM_TIME (expected)" )
333-
334- expected_messages = ["Need to use support signing key" ]
335- found_messages = self .check_udpproxy_output (
336- test_server , expected_messages )
337317
338- if "Need to use support signing key" in found_messages :
339- print ("✅ CONFIRMED: UDPProxy logged 'Need to use support signing key'" )
340- else :
341- print ("⚠️ UDPProxy output doesn't contain expected signing message" )
342-
343- def _test_bad_signing_key (self , test_server ):
344- """Test engineer connection with bad signing key."""
345- bad_passphrase = "wrong_auth" # Different from TEST_PASSPHRASE
346- bad_key = passphrase_to_key (bad_passphrase )
347-
348- heartbeat_count , system_time_count = self .run_test_scenario (
349- test_server , 'udp' , 'udp' , engineer_signing_key = bad_key , test_duration = 6
318+ def test_bad_signing_key (self , test_server ):
319+ """Engineer with the wrong passphrase - nothing through."""
320+ bad_key = passphrase_to_key ("wrong_auth" )
321+ self .run_test_scenario (
322+ test_server , 'udp' , 'udp' , engineer_signing_key = bad_key , test_duration = 4
350323 )
351-
352- print (f"SUCCESS: Bad key engineer received { heartbeat_count } HEARTBEAT, "
353- f"{ system_time_count } SYSTEM_TIME (sent 6 seconds of signed messages)" )
354-
355324 expected_messages = ["Bad support signing key" ]
356- found_messages = self .check_udpproxy_output (
357- test_server , expected_messages )
358-
359- if found_messages :
360- print (f"✅ FOUND bad signing messages: { found_messages } " )
361- else :
362- print ("⚠️ UDPProxy output doesn't contain expected bad key message" )
325+ self .check_udpproxy_output (test_server , expected_messages )
363326
364- def _test_good_signing_key (self , test_server ):
365- """Test engineer connection with correct signing key - should get both
366- HEARTBEAT and SYSTEM_TIME."""
327+ def test_good_signing_key (self , test_server ):
328+ """Engineer with the correct key - should get HEARTBEAT and SYSTEM_TIME."""
367329 correct_key = passphrase_to_key (TEST_PASSPHRASE )
368-
369330 heartbeat_count , system_time_count = self .run_test_scenario (
370331 test_server , 'udp' , 'udp' , engineer_signing_key = correct_key , test_duration = 4
371332 )
372-
373333 assert heartbeat_count > 0 , "Engineer should receive HEARTBEAT messages"
374334 assert system_time_count > 0 , \
375335 "Engineer should receive SYSTEM_TIME messages with correct signing"
376- print (f"SUCCESS: Correct key engineer received { heartbeat_count } HEARTBEAT, "
377- f"{ system_time_count } SYSTEM_TIME" )
378-
379- expected_messages = ["Got good signature" ]
380- found_messages = self .check_udpproxy_output (
381- test_server , expected_messages )
382-
383- if found_messages :
384- print (f"✅ FOUND authentication messages: { found_messages } " )
385- else :
386- print ("⚠️ No authentication messages found in UDPProxy output" )
387336
388337
389338class TestTCPConnections (BaseConnectionTest ):
390- """Test suite for TCP connection functionality using PyMAVLink
391- with authentication."""
392-
393- def test_basic_tcp_message_reception (self , test_server ):
394- """Test TCP message reception with different authentication scenarios."""
395-
396- # Test scenario 1: Engineer without signed connection
397- print ("\n === TCP TEST 1: Engineer without signed connection ===" )
398- self ._test_unsigned_engineer_tcp (test_server )
339+ """TCP/TCP scenarios. Same parallel-friendly split as TestUDPConnections."""
399340
400- # Test scenario 2: Engineer with bad signing key
401- print ("\n === TCP TEST 2: Engineer with bad signing key ===" )
402- self ._test_bad_signing_key_tcp (test_server )
403-
404- # Test scenario 3: Engineer with correct signing key
405- print ("\n === TCP TEST 3: Engineer with correct signing key ===" )
406- self ._test_good_signing_key_tcp (test_server )
407-
408- def _test_unsigned_engineer_tcp (self , test_server ):
409- """Test engineer TCP connection without signing - should only get HEARTBEAT."""
341+ def test_unsigned_engineer (self , test_server ):
342+ """Engineer TCP without signing - should only get HEARTBEAT."""
410343 heartbeat_count , system_time_count = self .run_test_scenario (
411344 test_server , 'tcp' , 'tcp' , engineer_signing_key = None ,
412345 test_duration = 3
413346 )
414-
415347 assert heartbeat_count > 0 , \
416348 "TCP Engineer should receive HEARTBEAT messages even without signing"
417349 assert system_time_count == 0 , \
418350 "TCP Engineer should NOT receive SYSTEM_TIME without proper signing"
419- print (f"SUCCESS: TCP Unsigned engineer received { heartbeat_count } "
420- f"HEARTBEAT, { system_time_count } SYSTEM_TIME (expected)" )
421-
422- expected_messages = ["Need to use support signing key" ]
423- found_messages = self .check_udpproxy_output (
424- test_server , expected_messages )
425-
426- if "Need to use support signing key" in found_messages :
427- print ("✅ CONFIRMED: UDPProxy logged 'Need to use support signing key' "
428- "for TCP" )
429- else :
430- print ("⚠️ UDPProxy output doesn't contain expected signing message "
431- "for TCP" )
432351
433- def _test_bad_signing_key_tcp (self , test_server ):
434- """Test engineer TCP connection with bad signing key."""
435- bad_passphrase = "wrong_auth" # Different from TEST_PASSPHRASE
436- bad_key = passphrase_to_key (bad_passphrase )
437-
438- heartbeat_count , system_time_count = self .run_test_scenario (
352+ def test_bad_signing_key (self , test_server ):
353+ """Engineer TCP with wrong passphrase - nothing through."""
354+ bad_key = passphrase_to_key ("wrong_auth" )
355+ self .run_test_scenario (
439356 test_server , 'tcp' , 'tcp' , engineer_signing_key = bad_key ,
440- test_duration = 8
357+ test_duration = 4
441358 )
442359
443- print (f"SUCCESS: TCP Bad key engineer received { heartbeat_count } "
444- f"HEARTBEAT, { system_time_count } SYSTEM_TIME "
445- f"(sent 8 seconds of signed messages)" )
446-
447- expected_messages = ["Bad support signing key" ,
448- "Need to use support signing key" ]
449- found_messages = self .check_udpproxy_output (
450- test_server , expected_messages )
451-
452- if found_messages :
453- print (f"✅ FOUND TCP bad signing messages: { found_messages } " )
454- else :
455- print ("⚠️ UDPProxy output doesn't contain expected bad key message "
456- "for TCP" )
457-
458- def _test_good_signing_key_tcp (self , test_server ):
459- """Test engineer TCP connection with correct signing key - should get
460- both HEARTBEAT and SYSTEM_TIME."""
360+ def test_good_signing_key (self , test_server ):
361+ """Engineer TCP with correct key - HEARTBEAT and SYSTEM_TIME flow."""
461362 correct_key = passphrase_to_key (TEST_PASSPHRASE )
462-
463363 heartbeat_count , system_time_count = self .run_test_scenario (
464364 test_server , 'tcp' , 'tcp' , engineer_signing_key = correct_key ,
465365 test_duration = 4
466366 )
467-
468367 assert heartbeat_count > 0 , "TCP Engineer should receive HEARTBEAT messages"
469-
470368 assert system_time_count > 0 , \
471369 "TCP Engineer should receive SYSTEM_TIME messages with correct signing"
472370
473- print (f"SUCCESS: TCP Correct key engineer received { heartbeat_count } "
474- f"HEARTBEAT, { system_time_count } SYSTEM_TIME" )
475-
476- expected_messages = ["Got good signature" ]
477- found_messages = self .check_udpproxy_output (
478- test_server , expected_messages )
479-
480- if found_messages :
481- print (f"✅ FOUND TCP authentication messages: { found_messages } " )
482- else :
483- print ("⚠️ No TCP authentication messages found in UDPProxy output" )
484-
485371
486372class TestMixedConnections (BaseConnectionTest ):
487373 """Test suite for mixed UDP/TCP connection scenarios."""
0 commit comments