From b3e0d1de1ba88b60ec06ed68c5319bbd8f2943b8 Mon Sep 17 00:00:00 2001 From: Paul Kronenwetter Date: Mon, 30 May 2022 16:58:41 -0400 Subject: [PATCH] Refactor tnc_states test. This will avoid cross-test interference this test has caused. --- test/test_tnc_states.py | 162 +++++++++++++++++++++++++--------------- 1 file changed, 102 insertions(+), 60 deletions(-) diff --git a/test/test_tnc_states.py b/test/test_tnc_states.py index 7a80917f..c7bc8cbc 100644 --- a/test/test_tnc_states.py +++ b/test/test_tnc_states.py @@ -2,7 +2,9 @@ Tests for the FreeDATA TNC state machine. """ +import multiprocessing import sys +import time import pytest @@ -90,66 +92,9 @@ def t_tsh_dummy(): print("In transmit_session_heartbeat") -@pytest.mark.parametrize("mycall", ["AA1AA-2", "DE2DE-0", "M4AWQ-4"]) -@pytest.mark.parametrize("dxcall", ["AA9AA-1", "DE2ED-0", "F6QWE-3"]) -def test_valid_disconnect(mycall: str, dxcall: str): - """ - Execute test to validate that receiving a session open frame sets the correct machine - state. - - :param mycall: Callsign of the near station - :type mycall: str - :param dxcall: Callsign of the far station - :type dxcall: str - :return: Bytearray of the requested frame - :rtype: bytearray - """ - # Set the SSIDs we'll use for this test. - static.SSID_LIST = [0, 1, 2, 3, 4] - - # Setup the static parameters for the connection. - mycallsign_bytes = helpers.callsign_to_bytes(mycall) - mycallsign = helpers.bytes_to_callsign(mycallsign_bytes) - static.MYCALLSIGN = mycallsign - static.MYCALLSIGN_CRC = helpers.get_crc_24(mycallsign) - - dxcallsign_bytes = helpers.callsign_to_bytes(dxcall) - dxcallsign = helpers.bytes_to_callsign(dxcallsign_bytes) - static.DXCALLSIGN = dxcallsign - static.DXCALLSIGN_CRC = helpers.get_crc_24(dxcallsign) - - # Create the TNC - tnc = data_handler.DATA() - tnc.arq_cleanup() - - # Replace the heartbeat transmit routine with our own, a No-Op. - tnc.transmit_session_heartbeat = t_tsh_dummy - - # Create packet to be 'received' by this station. - create_frame = t_create_start_session(mycall=dxcall, dxcall=mycall) - print_frame(create_frame) - tnc.received_session_opener(create_frame) - - assert static.ARQ_SESSION is True - assert static.TNC_STATE == "BUSY" - assert static.ARQ_SESSION_STATE == "connecting" - - # Create packet to be 'received' by this station. - close_frame = t_create_session_close(mycall=dxcall, dxcall=mycall) - print_frame(close_frame) - tnc.received_session_close(close_frame) - - assert helpers.callsign_to_bytes(static.MYCALLSIGN) == mycallsign_bytes - assert helpers.callsign_to_bytes(static.DXCALLSIGN) == dxcallsign_bytes - - assert static.ARQ_SESSION is False - assert static.TNC_STATE == "IDLE" - assert static.ARQ_SESSION_STATE == "disconnected" - - -@pytest.mark.parametrize("mycall", ["AA1AA-2", "DE2DE-0", "E4AWQ-4"]) -@pytest.mark.parametrize("dxcall", ["AA9AA-1", "DE2ED-0", "F6QWE-3"]) -def test_foreign_disconnect(mycall: str, dxcall: str): +# @pytest.mark.parametrize("mycall", ["AA1AA-2", "DE2DE-0", "E4AWQ-4"]) +# @pytest.mark.parametrize("dxcall", ["AA9AA-1", "DE2ED-0", "F6QWE-3"]) +def t_foreign_disconnect(mycall: str, dxcall: str): """ Execute test to validate that receiving a session open frame sets the correct machine state. @@ -223,6 +168,103 @@ def test_foreign_disconnect(mycall: str, dxcall: str): assert static.ARQ_SESSION_STATE == "connecting" +# @pytest.mark.parametrize("mycall", ["AA1AA-2", "DE2DE-0", "M4AWQ-4"]) +# @pytest.mark.parametrize("dxcall", ["AA9AA-1", "DE2ED-0", "F6QWE-3"]) +def t_valid_disconnect(mycall: str, dxcall: str): + """ + Execute test to validate that receiving a session open frame sets the correct machine + state. + + :param mycall: Callsign of the near station + :type mycall: str + :param dxcall: Callsign of the far station + :type dxcall: str + :return: Bytearray of the requested frame + :rtype: bytearray + """ + # Set the SSIDs we'll use for this test. + static.SSID_LIST = [0, 1, 2, 3, 4] + + # Setup the static parameters for the connection. + mycallsign_bytes = helpers.callsign_to_bytes(mycall) + mycallsign = helpers.bytes_to_callsign(mycallsign_bytes) + static.MYCALLSIGN = mycallsign + static.MYCALLSIGN_CRC = helpers.get_crc_24(mycallsign) + + dxcallsign_bytes = helpers.callsign_to_bytes(dxcall) + dxcallsign = helpers.bytes_to_callsign(dxcallsign_bytes) + static.DXCALLSIGN = dxcallsign + static.DXCALLSIGN_CRC = helpers.get_crc_24(dxcallsign) + + # Create the TNC + tnc = data_handler.DATA() + tnc.arq_cleanup() + + # Replace the heartbeat transmit routine with our own, a No-Op. + tnc.transmit_session_heartbeat = t_tsh_dummy + + # Create packet to be 'received' by this station. + create_frame = t_create_start_session(mycall=dxcall, dxcall=mycall) + print_frame(create_frame) + tnc.received_session_opener(create_frame) + + assert static.ARQ_SESSION is True + assert static.TNC_STATE == "BUSY" + assert static.ARQ_SESSION_STATE == "connecting" + + # Create packet to be 'received' by this station. + close_frame = t_create_session_close(mycall=dxcall, dxcall=mycall) + print_frame(close_frame) + tnc.received_session_close(close_frame) + + assert helpers.callsign_to_bytes(static.MYCALLSIGN) == mycallsign_bytes + assert helpers.callsign_to_bytes(static.DXCALLSIGN) == dxcallsign_bytes + + assert static.ARQ_SESSION is False + assert static.TNC_STATE == "IDLE" + assert static.ARQ_SESSION_STATE == "disconnected" + + +# These tests are pushed into separate processes as a workaround. These tests +# change the state of one of the static parts of the system. Unfortunately the +# specific state(s) maintained across tests in the same interpreter are not yet known. +# The other tests affected are: `test_tnc.py` and the ARQ tests. + + +@pytest.mark.parametrize("mycall", ["AA1AA-2", "DE2DE-0", "E4AWQ-4"]) +@pytest.mark.parametrize("dxcall", ["AA9AA-1", "DE2ED-0", "F6QWE-3"]) +def test_foreign_disconnect(mycall: str, dxcall: str): + proc = multiprocessing.Process(target=t_foreign_disconnect, args=(mycall, dxcall)) + # print("Starting threads.") + proc.start() + + time.sleep(0.05) + + # print("Terminating threads.") + proc.terminate() + proc.join() + + # print(f"\n{proc.exitcode=}") + assert proc.exitcode == 0 + + +@pytest.mark.parametrize("mycall", ["AA1AA-2", "DE2DE-0", "M4AWQ-4"]) +@pytest.mark.parametrize("dxcall", ["AA9AA-1", "DE2ED-0", "F6QWE-3"]) +def test_valid_disconnect(mycall: str, dxcall: str): + proc = multiprocessing.Process(target=t_valid_disconnect, args=(mycall, dxcall)) + # print("Starting threads.") + proc.start() + + time.sleep(0.05) + + # print("Terminating threads.") + proc.terminate() + proc.join() + + # print(f"\n{proc.exitcode=}") + assert proc.exitcode == 0 + + if __name__ == "__main__": # Run pytest with the current script as the filename. ecode = pytest.main(["-v", sys.argv[0]])