Crash_Override/FreeDATA
1
0
Fork 0
mirror of https://github.com/DJ2LS/FreeDATA synced 2024-05-14 08:04:33 +00:00
Code Issues Packages Projects Releases Wiki Activity

Work toward a usable test.

Browse source
This commit is contained in:
Paul Kronenwetter 2022-05-31 19:39:12 -04:00
parent b3e0d1de1b
commit 59dbf4174d
3 changed files with 69 additions and 25 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

7
test/wip_test_arq_file.py → test/test_arq_file.py
View file

@ -20,9 +20,12 @@ sys.path.insert(0, "..")
sys.path.insert(0, "../tnc")
sys.path.insert(0, "test")
import helpers
import log_handler
import util_arq_chat_file_1 as util1
import util_arq_chat_file_2 as util2
log_handler.setup_logging("/tmp/test")
STATIONS = ["AA2BB", "ZZ9YY"]
bytes_out = b'{"dt":"f","fn":"zeit.txt","ft":"text\\/plain","d":"data:text\\/plain;base64,MyBtb2Rlcywgb2huZSBjbGFzcwowLjAwMDk2OTQ4MTE4MDk5MTg0MTcKCjIgbW9kZXMsIG9obmUgY2xhc3MKMC4wMDA5NjY1NDUxODkxMjI1Mzk0CgoxIG1vZGUsIG9obmUgY2xhc3MKMC4wMDA5NjY5NzY1NTU4Nzc4MjA5CgMyBtb2Rlcywgb2huZSBjbGFzcwowLjAwMDk2OTQ4MTE4MDk5MTg0MTcKCjIgbW9kZXMsIG9obmUgY2xhc3MKMC4wMDA5NjY1NDUxODkxMjI1Mzk0CgoxIG1vZGUsIG9obmUgY2xhc3MKMC4wMDA5NjY5NzY1NTU4Nzc4MjA5Cg=MyBtb2Rlcywgb2huZSBjbGFzcwowLjAwMDk2OTQ4MTE4MDk5MTg0MTcKCjIgbW9kZXMsIG9obmUgY2xhc3MKMC4wMDA5NjY1NDUxODkxMjI1Mzk0CgoxIG1vZGUsIG9obmUgY2xhc3MKMC4wMDA5NjY5NzY1NTU4Nzc4MjA5CgMyBtb2Rlcywgb2huZSBjbGFzcwowLjAwMDk2OTQ4MTE4MDk5MTg0MTcKCjIgbW9kZXMsIG9obmUgY2xhc3MKMC4wMDA5NjY1NDUxODkxMjI1Mzk0CgoxIG1vZGUsIG9obmUgY2xhc3MKMC4wMDA5NjY5NzY1NTU4Nzc4MjA5CgMyBtb2Rlcywgb2huZSBjbGFzcwowLjAwMDk2OTQ4MTE4MDk5MTg0MTcKCjIgbW9kZXMsIG9obmUgY2xhc3MKMC4wMDA5NjY1NDUxODkxMjI1Mzk0CgoxIG1vZGUsIG9obmUgY2xhc3MKMC4wMDA5NjY5NzY1NTU4Nzc4MjA5Cg=","crc":"123123123"}'
@ -35,7 +38,7 @@ def analyze_results(station1: list, station2: list):
"""Examine the information retrieved from the sub-processes."""
# Data in these lists is either a series of bytes of received data,
# or a bytearray of transmitted data from the station.
log = structlog.get_logger(__name__)
log = structlog.get_logger("analyze_results")
for s1, s2, text in [(station1, station2, "S1"), (station2, station1, "S2")]:
for item in s1:
@ -67,7 +70,7 @@ def analyze_results(station1: list, station2: list):
@pytest.mark.parametrize("n_frames_per_burst", [2])
@pytest.mark.parametrize("lowbwmode", [False, True])
def test_arq_short(freedv_mode: str, n_frames_per_burst: int, lowbwmode: bool):
log = structlog.get_logger(__name__)
log = structlog.get_logger("test_arq_short")
s1_data = []
s2_data = []

48
test/util_arq_chat_file_1.py
View file

@ -34,7 +34,7 @@ def t_highsnr_arq_short_station1(
message: str,
lowbwmode: bool,
):
log = structlog.get_logger(__name__)
log = structlog.get_logger("station1")
orig_tx_func: object
orig_rx_func: object
@ -44,9 +44,10 @@ def t_highsnr_arq_short_station1(
t_frames = frames
parent_pipe.send(t_frames)
log.debug("S1 TX: ", frames=t_frames)
# frametype = int.from_bytes(t_frames[:1], "big")
# log.debug("S1 RX: ", frametype=frametype)
# log.info("S1 TX: ", frames=t_frames)
for item in t_frames:
frametype = int.from_bytes(item[:1], "big")
log.info("S1 TX: ", frametype=static.FRAME_TYPE(frametype).name)
# Apologies for the Python "magic." "orig_func" is a pointer to the
# original function captured before this one was put in place.
@ -63,7 +64,7 @@ def t_highsnr_arq_short_station1(
bytes_per_frame=bytes_per_frame,
)
frametype = int.from_bytes(t_bytes_out[:1], "big")
log.debug("S1 RX: ", frametype=frametype)
log.info("S1 RX: ", frametype=static.FRAME_TYPE(frametype).name)
# Apologies for the Python "magic." "orig_func" is a pointer to the
# original function captured before this one was put in place.
@ -82,7 +83,7 @@ def t_highsnr_arq_short_station1(
mycallsign = helpers.callsign_to_bytes(mycall)
mycallsign = helpers.bytes_to_callsign(mycallsign)
static.MYCALLSIGN = helpers.bytes_to_callsign(mycallsign)
static.MYCALLSIGN = mycallsign
static.MYCALLSIGN_CRC = helpers.get_crc_24(static.MYCALLSIGN)
dxcallsign = helpers.callsign_to_bytes(dxcall)
@ -94,11 +95,13 @@ def t_highsnr_arq_short_station1(
tnc = data_handler.DATA()
orig_rx_func = data_handler.DATA.process_data
data_handler.DATA.process_data = t_process_data
tnc.log = structlog.get_logger("station1_DATA")
# Create the modem
t_modem = modem.RF()
orig_tx_func = modem.RF.transmit
modem.RF.transmit = t_transmit
t_modem.log = structlog.get_logger("station1_RF")
time.sleep(0.5)
@ -114,7 +117,7 @@ def t_highsnr_arq_short_station1(
{
"data": b64_str,
"dxcallsign": dxcall,
"mode": codec2.freedv_get_mode_value_by_name(freedv_mode),
"mode": codec2.FREEDV_MODE[freedv_mode].value,
"n_frames": 1,
}
],
@ -127,10 +130,16 @@ def t_highsnr_arq_short_station1(
# time.sleep(0.5)
# sock.process_tnc_commands(json.dumps(data))
time.sleep(19)
timeout = time.time() + 45
while "ARQ;TRANSMITTING;SUCCESS" not in static.INFO:
if time.time() > timeout:
log.warning("station1", first=True)
break
time.sleep(0.1)
# Construct disconnect message to dxstation.
data = {"type": "ping", "command": "ping", "dxcallsign": dxcall}
# data = {"type": "ping", "command": "ping", "dxcallsign": dxcall}
# data = {"type": "arq", "command": "disconnect", "dxcallsign": dxcall}
# sock.process_tnc_commands(json.dumps(data))
# time.sleep(0.5)
@ -138,12 +147,23 @@ def t_highsnr_arq_short_station1(
# sock.process_tnc_commands(json.dumps(data))
# time.sleep(0.5)
sock.process_tnc_commands(json.dumps(data))
# sock.process_tnc_commands(json.dumps(data))
# time.sleep(1)
# sock.process_tnc_commands(json.dumps(data))
# time.sleep(0.5)
time.sleep(1)
timeout = time.time() + 40
while static.ARQ_STATE and tnc.data_queue_transmit.queue:
if time.time() > timeout:
log.error("station1", TIMEOUT=True)
break
time.sleep(0.5)
log.info("station1", arq_state=pformat(static.ARQ_STATE))
log.debug("Info: ", info=static.INFO)
log.info("S1 DQT: ", info=pformat(tnc.data_queue_transmit.queue))
log.info("S1 DQR: ", info=pformat(tnc.data_queue_received.queue))
log.info("S1 Info: ", info=static.INFO)
assert "DATACHANNEL;OPENING" in static.INFO
assert "DATACHANNEL;OPEN" in static.INFO
# assert "ARQ;SESSION;CLOSE" in static.INFO
log.debug("station1: Exiting!")
assert "ARQ;TRANSMITTING;SUCCESS" in static.INFO
log.error("station1: Exiting!")

39
test/util_arq_chat_file_2.py
View file

@ -34,7 +34,7 @@ def t_highsnr_arq_short_station2(
message: str,
lowbwmode: bool,
):
log = structlog.get_logger(__name__)
log = structlog.get_logger("station2")
orig_tx_func: object
orig_rx_func: object
@ -44,9 +44,10 @@ def t_highsnr_arq_short_station2(
t_frames = frames
parent_pipe.send(t_frames)
log.debug("S2 TX: ", frames=t_frames)
# frametype = int.from_bytes(t_frames[:1], "big")
# log.debug("S2 RX: ", frametype=frametype)
# log.info("S2 TX: ", frames=t_frames)
for item in t_frames:
frametype = int.from_bytes(item[:1], "big")
log.info("S2 TX: ", frametype=static.FRAME_TYPE(frametype).name)
# Apologies for the Python "magic." "orig_func" is a pointer to the
# original function captured before this one was put in place.
@ -63,7 +64,7 @@ def t_highsnr_arq_short_station2(
bytes_per_frame=bytes_per_frame,
)
frametype = int.from_bytes(t_bytes_out[:1], "big")
log.debug("S2 RX: ", frametype=frametype)
log.info("S2 RX: ", frametype=static.FRAME_TYPE(frametype).name)
# Apologies for the Python "magic." "orig_func" is a pointer to the
# original function captured before this one was put in place.
@ -82,7 +83,7 @@ def t_highsnr_arq_short_station2(
mycallsign = helpers.callsign_to_bytes(mycall)
mycallsign = helpers.bytes_to_callsign(mycallsign)
static.MYCALLSIGN = helpers.bytes_to_callsign(mycallsign)
static.MYCALLSIGN = mycallsign
static.MYCALLSIGN_CRC = helpers.get_crc_24(static.MYCALLSIGN)
dxcallsign = helpers.callsign_to_bytes(dxcall)
@ -94,16 +95,36 @@ def t_highsnr_arq_short_station2(
tnc = data_handler.DATA()
orig_rx_func = data_handler.DATA.process_data
data_handler.DATA.process_data = t_process_data
tnc.log = structlog.get_logger("station2_DATA")
# Create the modem
t_modem = modem.RF()
orig_tx_func = modem.RF.transmit
modem.RF.transmit = t_transmit
t_modem.log = structlog.get_logger("station2_RF")
time.sleep(22)
timeout = time.time() + 45
while "ARQ;RECEIVING;SUCCESS" not in static.INFO or static.ARQ_STATE:
if time.time() > timeout:
log.error("station2", first=True)
break
time.sleep(0.5)
log.info("station2", arq_state=pformat(static.ARQ_STATE))
log.debug("Info: ", info=static.INFO)
# Allow enough time for this side to transmit the final ACK.
timeout = time.time() + 40
while static.ARQ_STATE:
if time.time() > timeout:
log.error("station2", TIMEOUT=True)
break
time.sleep(0.5)
log.info("station2", arq_state=pformat(static.ARQ_STATE))
log.info("S2 DQT: ", info=pformat(tnc.data_queue_transmit.queue))
log.info("S2 DQR: ", info=pformat(tnc.data_queue_received.queue))
log.info("S2 Info: ", info=static.INFO)
assert "DATACHANNEL;RECEIVEDOPENER" in static.INFO
# assert "QRV;SENDING" in static.INFO
# assert "ARQ;SESSION;CLOSE" in static.INFO
log.debug("station2: Exiting!")
assert "ARQ;RECEIVING;SUCCESS" in static.INFO
log.error("station2: Exiting!")