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 Projects Releases Packages Wiki Activity

Work toward unifying the end-to-end tests.

Browse source
This commit is contained in:
Paul Kronenwetter 2022-06-14 18:27:19 -04:00
parent 0c243878ea
commit 9ace3a2491
9 changed files with 161 additions and 16 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

20
test/test_beacon.py
View file

@ -29,6 +29,12 @@ except ImportError:
STATIONS = ["AA2BB", "ZZ9YY"]
bytes_out = b'{"dt":"f","fn":"zeit.txt","ft":"text\\/plain","d":"data:text\\/plain;base64,MyBtb2Rlcywgb2huZSBjbGFzcwowLjAwMDk2OTQ4MTE4MDk5MTg0MTcKCjIgbW9kZXMsIG9obmUgY2xhc3MKMC4wMDA5NjY1NDUxODkxMjI1Mzk0CgoxIG1vZGUsIG9obmUgY2xhc3MKMC4wMDA5NjY5NzY1NTU4Nzc4MjA5CgMyBtb2Rlcywgb2huZSBjbGFzcwowLjAwMDk2OTQ4MTE4MDk5MTg0MTcKCjIgbW9kZXMsIG9obmUgY2xhc3MKMC4wMDA5NjY1NDUxODkxMjI1Mzk0CgoxIG1vZGUsIG9obmUgY2xhc3MKMC4wMDA5NjY5NzY1NTU4Nzc4MjA5Cg=MyBtb2Rlcywgb2huZSBjbGFzcwowLjAwMDk2OTQ4MTE4MDk5MTg0MTcKCjIgbW9kZXMsIG9obmUgY2xhc3MKMC4wMDA5NjY1NDUxODkxMjI1Mzk0CgoxIG1vZGUsIG9obmUgY2xhc3MKMC4wMDA5NjY5NzY1NTU4Nzc4MjA5CgMyBtb2Rlcywgb2huZSBjbGFzcwowLjAwMDk2OTQ4MTE4MDk5MTg0MTcKCjIgbW9kZXMsIG9obmUgY2xhc3MKMC4wMDA5NjY1NDUxODkxMjI1Mzk0CgoxIG1vZGUsIG9obmUgY2xhc3MKMC4wMDA5NjY5NzY1NTU4Nzc4MjA5CgMyBtb2Rlcywgb2huZSBjbGFzcwowLjAwMDk2OTQ4MTE4MDk5MTg0MTcKCjIgbW9kZXMsIG9obmUgY2xhc3MKMC4wMDA5NjY1NDUxODkxMjI1Mzk0CgoxIG1vZGUsIG9obmUgY2xhc3MKMC4wMDA5NjY5NzY1NTU4Nzc4MjA5Cg=","crc":"123123123"}'
messages = [
"This is a test chat...",
"This is a much longer message, hopefully longer than each of the datac1 and datac3 frames available to use in this modem. This should be long enought, but to err on the side of completeness this will string on for many more words before coming to the long awaited conclusion. We are not at the concluding point just yet because there is still more space to be taken up in the datac3 frame. Perhaps now would be a good place to terminate this test message, but perhaps not because we need a few more bytes. Here then we stop. This compresses so well that I need more data, even more stuff than is already here and included in the unreadable diatribe below, or is it a soliloquy? MyBtb2Rlcywgb2huZSBjbGFzcwowLjAwMDk2OTQ4MTE4MDk5MTg0MTcKCjIgbW9kZXMsIG9obmUgY2xhc3MKMC4wMDA5NjY1NDUxODkxMjI1Mzk0CgoxIG1vZGUsIG9obmUgY2xhc3MKMC4wMDA5NjY5NzY1NTU4Nzc4MjA5CgMyBtb2Rlcywgb2huZSBjbGFzcwowLjAwMDk2OTQ4MTE4MDk5MTg0MTcKCjIgbW9kZXMsIG9obmUgY2xhc3MKMC4wMDA5NjY1NDUxODkxMjI1Mzk0CgoxIG1vZGUsIG9obmUgY2xhc3MKMC4wMDA5NjY5NzY1NTU4Nzc4MjA5Cg=MyBtb2Rlcywgb2huZSBjbGFzcwowLjAwMDk2OTQ4MTE4MDk5MTg0MTcKCjIgbW9kZXMsIG9obmUgY2xhc3MKMC4wMDA5NjY1NDUxODkxMjI1Mzk0CgoxIG1vZGUsIG9obmUgY2xhc3MKMC4wMDA5NjY5NzY1NTU4Nzc4MjA5CgMyBtb2Rlcywgb2huZSBjbGFzcwowLjAwMDk2OTQ4MTE4MDk5MTg0MTcKCjIgbW9kZXMsIG9obmUgY2xhc3MKMC4wMDA5NjY1NDUxODkxMjI1Mzk0CgoxIG1vZGUsIG9obmUgY2xhc3MKMC4wMDA5NjY5NzY1NTU4Nzc4MjA5CgMyBtb2Rlcywgb2huZSBjbGFzcwowLjAwMDk2OTQ4MTE4MDk5MTg0MTcKCjIgbW9kZXMsIG9obmUgY2xhc3MKMC4wMDA5NjY1NDUxODkxMjI1Mzk0CgoxIG1vZGUsIG9obmUgY2xhc3MKMC4wMDA5NjY5NzY1NTU4Nzc4MjA5Cg=",
]
PIPE_THREAD_RUNNING = True
@ -94,7 +100,13 @@ def analyze_results(station1: list, station2: list, call_list: list):
locate_data_with_crc(s2, text, data, frametype)
def test_beacon(tmp_path):
@pytest.mark.parametrize("freedv_mode", ["datac1", "datac3"])
@pytest.mark.parametrize("n_frames_per_burst", [1]) # Higher fpb is broken.
@pytest.mark.parametrize("message_no", range(len(messages)))
# @pytest.mark.flaky(reruns=2)
def test_beacon(
freedv_mode: str, n_frames_per_burst: int, message_no: int, tmp_path
):
log_handler.setup_logging(filename=tmp_path / "test_beacon", level="INFO")
log = structlog.get_logger("test_beacon")
@ -128,8 +140,11 @@ def test_beacon(tmp_path):
target=util1.t_beacon1,
args=(
s1_send,
freedv_mode,
n_frames_per_burst,
STATIONS[0],
STATIONS[1],
messages[message_no],
True, # low bandwidth mode
tmp_path,
),
@ -139,8 +154,11 @@ def test_beacon(tmp_path):
target=util2.t_beacon2,
args=(
s2_send,
freedv_mode,
n_frames_per_burst,
STATIONS[1],
STATIONS[0],
messages[message_no],
True, # low bandwidth mode
tmp_path,
),

2
test/test_chat_text.py
View file

@ -103,7 +103,7 @@ def analyze_results(station1: list, station2: list, call_list: list):
@pytest.mark.parametrize("freedv_mode", ["datac1", "datac3"])
@pytest.mark.parametrize("n_frames_per_burst", [1]) # Higher fpb is broken.
@pytest.mark.parametrize("message_no", range(len(messages)))
@pytest.mark.flaky(reruns=2)
# @pytest.mark.flaky(reruns=2)
def test_chat_text(
freedv_mode: str, n_frames_per_burst: int, message_no: int, tmp_path
):

21
test/test_cq.py
View file

@ -29,6 +29,12 @@ except ImportError:
STATIONS = ["AA2BB", "ZZ9YY"]
bytes_out = b'{"dt":"f","fn":"zeit.txt","ft":"text\\/plain","d":"data:text\\/plain;base64,MyBtb2Rlcywgb2huZSBjbGFzcwowLjAwMDk2OTQ4MTE4MDk5MTg0MTcKCjIgbW9kZXMsIG9obmUgY2xhc3MKMC4wMDA5NjY1NDUxODkxMjI1Mzk0CgoxIG1vZGUsIG9obmUgY2xhc3MKMC4wMDA5NjY5NzY1NTU4Nzc4MjA5CgMyBtb2Rlcywgb2huZSBjbGFzcwowLjAwMDk2OTQ4MTE4MDk5MTg0MTcKCjIgbW9kZXMsIG9obmUgY2xhc3MKMC4wMDA5NjY1NDUxODkxMjI1Mzk0CgoxIG1vZGUsIG9obmUgY2xhc3MKMC4wMDA5NjY5NzY1NTU4Nzc4MjA5Cg=MyBtb2Rlcywgb2huZSBjbGFzcwowLjAwMDk2OTQ4MTE4MDk5MTg0MTcKCjIgbW9kZXMsIG9obmUgY2xhc3MKMC4wMDA5NjY1NDUxODkxMjI1Mzk0CgoxIG1vZGUsIG9obmUgY2xhc3MKMC4wMDA5NjY5NzY1NTU4Nzc4MjA5CgMyBtb2Rlcywgb2huZSBjbGFzcwowLjAwMDk2OTQ4MTE4MDk5MTg0MTcKCjIgbW9kZXMsIG9obmUgY2xhc3MKMC4wMDA5NjY1NDUxODkxMjI1Mzk0CgoxIG1vZGUsIG9obmUgY2xhc3MKMC4wMDA5NjY5NzY1NTU4Nzc4MjA5CgMyBtb2Rlcywgb2huZSBjbGFzcwowLjAwMDk2OTQ4MTE4MDk5MTg0MTcKCjIgbW9kZXMsIG9obmUgY2xhc3MKMC4wMDA5NjY1NDUxODkxMjI1Mzk0CgoxIG1vZGUsIG9obmUgY2xhc3MKMC4wMDA5NjY5NzY1NTU4Nzc4MjA5Cg=","crc":"123123123"}'
messages = [
"This is a test chat...",
"This is a much longer message, hopefully longer than each of the datac1 and datac3 frames available to use in this modem. This should be long enought, but to err on the side of completeness this will string on for many more words before coming to the long awaited conclusion. We are not at the concluding point just yet because there is still more space to be taken up in the datac3 frame. Perhaps now would be a good place to terminate this test message, but perhaps not because we need a few more bytes. Here then we stop. This compresses so well that I need more data, even more stuff than is already here and included in the unreadable diatribe below, or is it a soliloquy? MyBtb2Rlcywgb2huZSBjbGFzcwowLjAwMDk2OTQ4MTE4MDk5MTg0MTcKCjIgbW9kZXMsIG9obmUgY2xhc3MKMC4wMDA5NjY1NDUxODkxMjI1Mzk0CgoxIG1vZGUsIG9obmUgY2xhc3MKMC4wMDA5NjY5NzY1NTU4Nzc4MjA5CgMyBtb2Rlcywgb2huZSBjbGFzcwowLjAwMDk2OTQ4MTE4MDk5MTg0MTcKCjIgbW9kZXMsIG9obmUgY2xhc3MKMC4wMDA5NjY1NDUxODkxMjI1Mzk0CgoxIG1vZGUsIG9obmUgY2xhc3MKMC4wMDA5NjY5NzY1NTU4Nzc4MjA5Cg=MyBtb2Rlcywgb2huZSBjbGFzcwowLjAwMDk2OTQ4MTE4MDk5MTg0MTcKCjIgbW9kZXMsIG9obmUgY2xhc3MKMC4wMDA5NjY1NDUxODkxMjI1Mzk0CgoxIG1vZGUsIG9obmUgY2xhc3MKMC4wMDA5NjY5NzY1NTU4Nzc4MjA5CgMyBtb2Rlcywgb2huZSBjbGFzcwowLjAwMDk2OTQ4MTE4MDk5MTg0MTcKCjIgbW9kZXMsIG9obmUgY2xhc3MKMC4wMDA5NjY1NDUxODkxMjI1Mzk0CgoxIG1vZGUsIG9obmUgY2xhc3MKMC4wMDA5NjY5NzY1NTU4Nzc4MjA5CgMyBtb2Rlcywgb2huZSBjbGFzcwowLjAwMDk2OTQ4MTE4MDk5MTg0MTcKCjIgbW9kZXMsIG9obmUgY2xhc3MKMC4wMDA5NjY1NDUxODkxMjI1Mzk0CgoxIG1vZGUsIG9obmUgY2xhc3MKMC4wMDA5NjY5NzY1NTU4Nzc4MjA5Cg=",
]
PIPE_THREAD_RUNNING = True
@ -94,8 +100,13 @@ def analyze_results(station1: list, station2: list, call_list: list):
locate_data_with_crc(s2, text, data, frametype)
@pytest.mark.flaky(reruns=2)
def test_cq(tmp_path):
@pytest.mark.parametrize("freedv_mode", ["datac1", "datac3"])
@pytest.mark.parametrize("n_frames_per_burst", [1]) # Higher fpb is broken.
@pytest.mark.parametrize("message_no", range(len(messages)))
# @pytest.mark.flaky(reruns=2)
def test_cq(
freedv_mode: str, n_frames_per_burst: int, message_no: int, tmp_path
):
log_handler.setup_logging(filename=tmp_path / "test_cq", level="INFO")
log = structlog.get_logger("test_cq")
@ -129,8 +140,11 @@ def test_cq(tmp_path):
target=util1.t_cq1,
args=(
s1_send,
freedv_mode,
n_frames_per_burst,
STATIONS[0],
STATIONS[1],
messages[message_no],
True, # low bandwidth mode
tmp_path,
),
@ -140,8 +154,11 @@ def test_cq(tmp_path):
target=util2.t_cq2,
args=(
s2_send,
freedv_mode,
n_frames_per_burst,
STATIONS[1],
STATIONS[0],
messages[message_no],
True, # low bandwidth mode
tmp_path,
),

42
test/util_beacon_1.py
View file

@ -6,11 +6,13 @@ Created on Wed Dec 23 07:04:24 2020
@author: DJ2LS
"""
import base64
import json
import time
from pprint import pformat
from typing import Callable
import codec2
import data_handler
import helpers
import modem
@ -71,8 +73,11 @@ def t_setup(
def t_beacon1(
parent_pipe,
freedv_mode: str,
n_frames_per_burst: int,
mycall: str,
dxcall: str,
message: str,
lowbwmode: bool,
tmp_path,
):
@ -121,17 +126,38 @@ def t_beacon1(
log.info("t_beacon1:", RXCHANNEL=modem.RXCHANNEL)
log.info("t_beacon1:", TXCHANNEL=modem.TXCHANNEL)
# Construct message to dxstation.
# b64_str = str(base64.b64encode(bytes(message, "UTF-8")), "UTF-8").strip()
# data = {
# "type": "arq",
# "command": "send_raw",
# "parameter": [
# {
# "data": b64_str,
# "dxcallsign": dxcall,
# "mode": codec2.FREEDV_MODE[freedv_mode].value,
# "n_frames": n_frames_per_burst,
# }
# ],
# }
# Construct message to start beacon.
data = {"type": "command", "command": "start_beacon", "parameter": "5"}
# Construct message to start cq.
# data = {"type": "command", "command": "cqcqcq"}
sock.process_tnc_commands(json.dumps(data, indent=None))
time.sleep(0.5)
sock.process_tnc_commands(json.dumps(data, indent=None))
# This transaction should take less than 14 sec.
timeout = time.time() + 10
# Assure the test completes.
timeout = time.time() + 10 # 25
# Compare with the string conversion instead of repeatedly dumping
# the queue to an object for comparisons.
# while '"arq":"transmission","status":"transmitted"' not in str(
# sock.SOCKET_QUEUE.queue
# ):
while '"beacon":"transmitted"' not in str(sock.SOCKET_QUEUE.queue):
# while '"cq":"transmitting"' not in str(sock.SOCKET_QUEUE.queue):
if time.time() > timeout:
log.warning("station1 TIMEOUT", first=True)
break
@ -140,9 +166,11 @@ def t_beacon1(
data = {"type": "arq", "command": "disconnect", "dxcallsign": dxcall}
sock.process_tnc_commands(json.dumps(data, indent=None))
time.sleep(0.5)
sock.process_tnc_commands(json.dumps(data, indent=None))
# Allow enough time for this side to process the disconnect frame.
timeout = time.time() + 10
timeout = time.time() + 10 # 20
while static.ARQ_STATE or tnc.data_queue_transmit.queue:
if time.time() > timeout:
log.error("station1", TIMEOUT=True)
@ -153,8 +181,16 @@ def t_beacon1(
# log.info("S1 DQT: ", DQ_Tx=pformat(tnc.data_queue_transmit.queue))
# log.info("S1 DQR: ", DQ_Rx=pformat(tnc.data_queue_received.queue))
# log.info("S1 Socket: ", socket_queue=pformat(sock.SOCKET_QUEUE.queue))
# assert '"arq":"transmission","status":"transmitting"' in str(
# sock.SOCKET_QUEUE.queue
# )
# assert '"arq":"transmission","status":"transmitted"' in str(sock.SOCKET_QUEUE.queue)
# assert '"arq":"transmission","status":"failed"' not in str(sock.SOCKET_QUEUE.queue)
# assert '"percent":100' in str(sock.SOCKET_QUEUE.queue)
assert '"beacon":"transmitting"' in str(sock.SOCKET_QUEUE.queue)
assert '"beacon":"failed"' not in str(sock.SOCKET_QUEUE.queue)
# assert '"cq":"transmitting"' in str(sock.SOCKET_QUEUE.queue)
# assert '"cq":"failed"' not in str(sock.SOCKET_QUEUE.queue)
assert '"command_response":"disconnect","status":"OK"' in str(
sock.SOCKET_QUEUE.queue
)

17
test/util_beacon_2.py
View file

@ -70,8 +70,11 @@ def t_setup(
def t_beacon2(
parent_pipe,
freedv_mode: str,
n_frames_per_burst: int,
mycall: str,
dxcall: str,
message: str,
lowbwmode: bool,
tmp_path,
):
@ -120,11 +123,16 @@ def t_beacon2(
log.info("t_beacon2:", RXCHANNEL=modem.RXCHANNEL)
log.info("t_beacon2:", TXCHANNEL=modem.TXCHANNEL)
# This transaction should take less than 14 sec.
timeout = time.time() + 20
# Assure the test completes.
timeout = time.time() + 20 # 25
# Compare with the string conversion instead of repeatedly dumping
# the queue to an object for comparisons.
# while (
# '"arq":"transmission","status":"received"' not in str(sock.SOCKET_QUEUE.queue)
# or static.ARQ_STATE
# ):
while '"beacon":"received"' not in str(sock.SOCKET_QUEUE.queue):
# while '"cq":"received"' not in str(sock.SOCKET_QUEUE.queue):
if time.time() > timeout:
log.warning("station2 TIMEOUT", first=True)
break
@ -132,7 +140,7 @@ def t_beacon2(
log.info("station2, first", arq_state=pformat(static.ARQ_STATE))
# Allow enough time for this side to receive the disconnect frame.
timeout = time.time() + 15
timeout = time.time() + 15 # 20
while '"arq":"session","status":"close"' not in str(sock.SOCKET_QUEUE.queue):
if time.time() > timeout:
log.error("station2", TIMEOUT=True)
@ -143,6 +151,9 @@ def t_beacon2(
# log.info("S2 DQT: ", DQ_Tx=pformat(tnc.data_queue_transmit.queue))
# log.info("S2 DQR: ", DQ_Rx=pformat(tnc.data_queue_received.queue))
# log.info("S2 Socket: ", socket_queue=pformat(sock.SOCKET_QUEUE.queue))
# assert '"arq":"transmission","status":"received"' in str(sock.SOCKET_QUEUE.queue)
assert '"beacon":"received"' in str(sock.SOCKET_QUEUE.queue)
# assert '"cq":"received"' in str(sock.SOCKET_QUEUE.queue)
# assert '"qrv":"transmitting"' in str(sock.SOCKET_QUEUE.queue)
assert '"arq":"session","status":"close"' in str(sock.SOCKET_QUEUE.queue)
log.error("station2: Exiting!")

15
test/util_chat_text_1.py
View file

@ -140,17 +140,24 @@ def t_highsnr_arq_short_station1(
}
],
}
# Construct message to start beacon.
# data = {"type": "command", "command": "start_beacon", "parameter": "5"}
# Construct message to start cq.
# data = {"type": "command", "command": "cqcqcq"}
sock.process_tnc_commands(json.dumps(data, indent=None))
time.sleep(0.5)
sock.process_tnc_commands(json.dumps(data, indent=None))
# This transaction should take less than 14 sec.
# Assure the test completes.
timeout = time.time() + 25
# Compare with the string conversion instead of repeatedly dumping
# the queue to an object for comparisons.
while '"arq":"transmission","status":"transmitted"' not in str(
sock.SOCKET_QUEUE.queue
):
# while '"beacon":"transmitted"' not in str(sock.SOCKET_QUEUE.queue):
# while '"cq":"transmitting"' not in str(sock.SOCKET_QUEUE.queue):
if time.time() > timeout:
log.warning("station1 TIMEOUT", first=True)
break
@ -159,6 +166,8 @@ def t_highsnr_arq_short_station1(
data = {"type": "arq", "command": "disconnect", "dxcallsign": dxcall}
sock.process_tnc_commands(json.dumps(data, indent=None))
time.sleep(0.5)
sock.process_tnc_commands(json.dumps(data, indent=None))
# Allow enough time for this side to process the disconnect frame.
timeout = time.time() + 20
@ -178,6 +187,10 @@ def t_highsnr_arq_short_station1(
assert '"arq":"transmission","status":"transmitted"' in str(sock.SOCKET_QUEUE.queue)
assert '"arq":"transmission","status":"failed"' not in str(sock.SOCKET_QUEUE.queue)
assert '"percent":100' in str(sock.SOCKET_QUEUE.queue)
# assert '"beacon":"transmitting"' in str(sock.SOCKET_QUEUE.queue)
# assert '"beacon":"failed"' not in str(sock.SOCKET_QUEUE.queue)
# assert '"cq":"transmitting"' in str(sock.SOCKET_QUEUE.queue)
# assert '"cq":"failed"' not in str(sock.SOCKET_QUEUE.queue)
assert '"command_response":"disconnect","status":"OK"' in str(
sock.SOCKET_QUEUE.queue
)

7
test/util_chat_text_2.py
View file

@ -123,7 +123,7 @@ def t_highsnr_arq_short_station2(
log.info("t_highsnr_arq_short_station2:", RXCHANNEL=modem.RXCHANNEL)
log.info("t_highsnr_arq_short_station2:", TXCHANNEL=modem.TXCHANNEL)
# This transaction should take less than 14 sec.
# Assure the test completes.
timeout = time.time() + 25
# Compare with the string conversion instead of repeatedly dumping
# the queue to an object for comparisons.
@ -131,6 +131,8 @@ def t_highsnr_arq_short_station2(
'"arq":"transmission","status":"received"' not in str(sock.SOCKET_QUEUE.queue)
or static.ARQ_STATE
):
# while '"beacon":"received"' not in str(sock.SOCKET_QUEUE.queue):
# while '"cq":"received"' not in str(sock.SOCKET_QUEUE.queue):
if time.time() > timeout:
log.warning("station2 TIMEOUT", first=True)
break
@ -150,5 +152,8 @@ def t_highsnr_arq_short_station2(
# log.info("S2 DQR: ", DQ_Rx=pformat(tnc.data_queue_received.queue))
# log.info("S2 Socket: ", socket_queue=pformat(sock.SOCKET_QUEUE.queue))
assert '"arq":"transmission","status":"received"' in str(sock.SOCKET_QUEUE.queue)
# assert '"beacon":"received"' in str(sock.SOCKET_QUEUE.queue)
# assert '"cq":"received"' in str(sock.SOCKET_QUEUE.queue)
# assert '"qrv":"transmitting"' in str(sock.SOCKET_QUEUE.queue)
assert '"arq":"session","status":"close"' in str(sock.SOCKET_QUEUE.queue)
log.error("station2: Exiting!")

38
test/util_cq_1.py
View file

@ -6,11 +6,13 @@ Created on Wed Dec 23 07:04:24 2020
@author: DJ2LS
"""
import base64
import json
import time
from pprint import pformat
from typing import Callable
import codec2
import data_handler
import helpers
import modem
@ -71,8 +73,11 @@ def t_setup(
def t_cq1(
parent_pipe,
freedv_mode: str,
n_frames_per_burst: int,
mycall: str,
dxcall: str,
message: str,
lowbwmode: bool,
tmp_path,
):
@ -121,6 +126,22 @@ def t_cq1(
log.info("t_cq1:", RXCHANNEL=modem.RXCHANNEL)
log.info("t_cq1:", TXCHANNEL=modem.TXCHANNEL)
# Construct message to dxstation.
# b64_str = str(base64.b64encode(bytes(message, "UTF-8")), "UTF-8").strip()
# data = {
# "type": "arq",
# "command": "send_raw",
# "parameter": [
# {
# "data": b64_str,
# "dxcallsign": dxcall,
# "mode": codec2.FREEDV_MODE[freedv_mode].value,
# "n_frames": n_frames_per_burst,
# }
# ],
# }
# Construct message to start beacon.
# data = {"type": "command", "command": "start_beacon", "parameter": "5"}
# Construct message to start cq.
data = {"type": "command", "command": "cqcqcq"}
@ -128,9 +149,14 @@ def t_cq1(
time.sleep(0.5)
sock.process_tnc_commands(json.dumps(data, indent=None))
timeout = time.time() + 5
# Assure the test completes.
timeout = time.time() + 5 # 25
# Compare with the string conversion instead of repeatedly dumping
# the queue to an object for comparisons.
# while '"arq":"transmission","status":"transmitted"' not in str(
# sock.SOCKET_QUEUE.queue
# ):
# while '"beacon":"transmitted"' not in str(sock.SOCKET_QUEUE.queue):
while '"cq":"transmitting"' not in str(sock.SOCKET_QUEUE.queue):
if time.time() > timeout:
log.warning("station1 TIMEOUT", first=True)
@ -144,7 +170,7 @@ def t_cq1(
sock.process_tnc_commands(json.dumps(data, indent=None))
# Allow enough time for this side to process the disconnect frame.
timeout = time.time() + 5
timeout = time.time() + 5 # 20
while static.ARQ_STATE or tnc.data_queue_transmit.queue:
if time.time() > timeout:
log.error("station1", TIMEOUT=True)
@ -155,6 +181,14 @@ def t_cq1(
# log.info("S1 DQT: ", DQ_Tx=pformat(tnc.data_queue_transmit.queue))
# log.info("S1 DQR: ", DQ_Rx=pformat(tnc.data_queue_received.queue))
# log.info("S1 Socket: ", socket_queue=pformat(sock.SOCKET_QUEUE.queue))
# assert '"arq":"transmission","status":"transmitting"' in str(
# sock.SOCKET_QUEUE.queue
# )
# assert '"arq":"transmission","status":"transmitted"' in str(sock.SOCKET_QUEUE.queue)
# assert '"arq":"transmission","status":"failed"' not in str(sock.SOCKET_QUEUE.queue)
# assert '"percent":100' in str(sock.SOCKET_QUEUE.queue)
# assert '"beacon":"transmitting"' in str(sock.SOCKET_QUEUE.queue)
# assert '"beacon":"failed"' not in str(sock.SOCKET_QUEUE.queue)
assert '"cq":"transmitting"' in str(sock.SOCKET_QUEUE.queue)
assert '"cq":"failed"' not in str(sock.SOCKET_QUEUE.queue)
assert '"command_response":"disconnect","status":"OK"' in str(

15
test/util_cq_2.py
View file

@ -70,8 +70,11 @@ def t_setup(
def t_cq2(
parent_pipe,
freedv_mode: str,
n_frames_per_burst: int,
mycall: str,
dxcall: str,
message: str,
lowbwmode: bool,
tmp_path,
):
@ -120,9 +123,15 @@ def t_cq2(
log.info("t_cq2:", RXCHANNEL=modem.RXCHANNEL)
log.info("t_cq2:", TXCHANNEL=modem.TXCHANNEL)
timeout = time.time() + 5
# Assure the test completes.
timeout = time.time() + 5 # 25
# Compare with the string conversion instead of repeatedly dumping
# the queue to an object for comparisons.
# while (
# '"arq":"transmission","status":"received"' not in str(sock.SOCKET_QUEUE.queue)
# or static.ARQ_STATE
# ):
# while '"beacon":"received"' not in str(sock.SOCKET_QUEUE.queue):
while '"cq":"received"' not in str(sock.SOCKET_QUEUE.queue):
if time.time() > timeout:
log.warning("station2 TIMEOUT", first=True)
@ -131,7 +140,7 @@ def t_cq2(
log.info("station2, first", arq_state=pformat(static.ARQ_STATE))
# Allow enough time for this side to receive the disconnect frame.
timeout = time.time() + 6
timeout = time.time() + 6 # 20
while '"arq":"session","status":"close"' not in str(sock.SOCKET_QUEUE.queue):
if time.time() > timeout:
log.error("station2", TIMEOUT=True)
@ -142,6 +151,8 @@ def t_cq2(
# log.info("S2 DQT: ", DQ_Tx=pformat(tnc.data_queue_transmit.queue))
# log.info("S2 DQR: ", DQ_Rx=pformat(tnc.data_queue_received.queue))
# log.info("S2 Socket: ", socket_queue=pformat(sock.SOCKET_QUEUE.queue))
# assert '"arq":"transmission","status":"received"' in str(sock.SOCKET_QUEUE.queue)
# assert '"beacon":"received"' in str(sock.SOCKET_QUEUE.queue)
assert '"cq":"received"' in str(sock.SOCKET_QUEUE.queue)
assert '"qrv":"transmitting"' in str(sock.SOCKET_QUEUE.queue)
assert '"arq":"session","status":"close"' in str(sock.SOCKET_QUEUE.queue)