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

Adjust checks to use SOCKET_QUEUE.

Browse source
This commit is contained in:
Paul Kronenwetter 2022-06-06 20:26:41 -04:00
parent 1fcd7dbdec
commit e183f9a419
3 changed files with 30 additions and 9 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

4
test/test_tnc.py
View file

@ -16,8 +16,8 @@ from test import util_tnc_ISS as iss
import pytest
# These do not update static.INFO:
# "CONNECT", "SEND_TEST_FRAME"
# These do not update sock.SOCKET_QUEUE:
# "CONNECT"
# This test is currently a little inconsistent.
@pytest.mark.flaky(reruns=2)
@pytest.mark.parametrize("command", ["CQ", "PING", "BEACON"])

17
test/util_tnc_IRS.py
View file

@ -10,26 +10,37 @@ import os
import sys
import time
import structlog
sys.path.insert(0, "..")
sys.path.insert(0, "../tnc")
import data_handler
import helpers
import modem
import sock
import static
IRS_original_arq_cleanup = object
MESSAGE: str
log = structlog.get_logger("util_tnc_IRS")
def irs_arq_cleanup():
"""Replacement for modem.arq_cleanup to detect when to exit process."""
if "TRANSMISSION;STOPPED" in static.INFO:
print(f"{static.INFO=}")
log.info(
"irs_arq_cleanup", socket_queue=sock.SOCKET_QUEUE.queue, message=MESSAGE.lower()
)
if '"arq":"transmission","status":"stopped"' in str(sock.SOCKET_QUEUE.queue):
# log.info("irs_arq_cleanup", socket_queue=sock.SOCKET_QUEUE.queue)
time.sleep(2)
# sys.exit does not terminate threads.
# pylint: disable=protected-access
if f"{MESSAGE};RECEIVING" not in static.INFO:
if f'"{MESSAGE.lower()}":"receiving"' not in str(
sock.SOCKET_QUEUE.queue
) and f'"{MESSAGE.lower()}":"received"' not in str(sock.SOCKET_QUEUE.queue):
print(f"{MESSAGE} was not received.")
log.info("irs_arq_cleanup", socket_queue=sock.SOCKET_QUEUE.queue)
os._exit(1)
os._exit(0)

18
test/util_tnc_ISS.py
View file

@ -10,26 +10,36 @@ import os
import sys
import time
sys.path.insert(0, "..")
import structlog
sys.path.insert(0, "../tnc")
import data_handler
import helpers
import modem
import sock
import static
ISS_original_arq_cleanup = object
MESSAGE: str
log = structlog.get_logger("util_tnc_ISS")
def iss_arq_cleanup():
"""Replacement for modem.arq_cleanup to detect when to exit process."""
if "TRANSMISSION;STOPPED" in static.INFO:
print(f"{static.INFO=}")
log.info(
"irs_arq_cleanup", socket_queue=sock.SOCKET_QUEUE.queue, message=MESSAGE.lower()
)
if '"arq":"transmission","status":"stopped"' in str(sock.SOCKET_QUEUE.queue):
# log.info("iss_arq_cleanup", socket_queue=sock.SOCKET_QUEUE.queue)
time.sleep(1)
# sys.exit does not terminate threads.
# pylint: disable=protected-access
if f"{MESSAGE};SENDING" not in static.INFO:
if f'"{MESSAGE.lower()}":"transmitting"' not in str(
sock.SOCKET_QUEUE.queue
) and f'"{MESSAGE.lower()}":"sending"' not in str(sock.SOCKET_QUEUE.queue):
print(f"{MESSAGE} was not sent.")
log.info("iss_arq_cleanup", socket_queue=sock.SOCKET_QUEUE.queue)
os._exit(1)
os._exit(0)