mirror of
https://github.com/DJ2LS/FreeDATA
synced 2024-05-14 08:04:33 +00:00
Refactor tnc_states test.
This will avoid cross-test interference this test has caused.
This commit is contained in:
Paul Kronenwetter
parent
55fc2d2263
commit
b3e0d1de1b
|
|
@ -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]])
|
||||
|
|
|
|||
Loading…
Reference in a new issue