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

Use a better way to exit the infinite while loop.

Browse source
This commit is contained in:
Paul Kronenwetter 2022-06-10 16:54:20 -04:00
parent ab772d1f70
commit 84e74d31e2
3 changed files with 14 additions and 12 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

4
test/test_tnc.py
View file

@ -43,8 +43,8 @@ def test_tnc(command):
except FileNotFoundError as fnfe:
print(f"Unlinking pipe: {fnfe}")
assert iss_proc.exitcode == 0, f"Transmit side failed test. {iss_proc}"
assert irs_proc.exitcode == 0, f"Receive side failed test. {irs_proc}"
assert iss_proc.exitcode in [0, -15], f"Transmit side failed test. {iss_proc}"
assert irs_proc.exitcode in [0, -15], f"Receive side failed test. {irs_proc}"
if __name__ == "__main__":

11
test/util_tnc_IRS.py
View file

@ -7,8 +7,10 @@ Created on Wed Dec 23 07:04:24 2020
"""
import os
import signal
import sys
import time
from typing import Callable
import structlog
@ -20,7 +22,7 @@ import modem
import sock
import static
IRS_original_arq_cleanup = object
IRS_original_arq_cleanup: Callable
MESSAGE: str
log = structlog.get_logger("util_tnc_IRS")
@ -34,16 +36,15 @@ def irs_arq_cleanup():
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.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)
# sys.exit does not terminate threads, and os_exit doesn't allow coverage collection.
signal.raise_signal(signal.SIGKILL)
os._exit(0)
signal.raise_signal(signal.SIGTERM)
IRS_original_arq_cleanup()

11
test/util_tnc_ISS.py
View file

@ -7,8 +7,10 @@ Created on Wed Dec 23 07:04:24 2020
"""
import os
import signal
import sys
import time
from typing import Callable
import structlog
@ -19,7 +21,7 @@ import modem
import sock
import static
ISS_original_arq_cleanup = object
ISS_original_arq_cleanup: Callable
MESSAGE: str
log = structlog.get_logger("util_tnc_ISS")
@ -33,16 +35,15 @@ def iss_arq_cleanup():
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.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)
# sys.exit does not terminate threads, and os_exit doesn't allow coverage collection.
signal.raise_signal(signal.SIGKILL)
os._exit(0)
signal.raise_signal(signal.SIGTERM)
ISS_original_arq_cleanup()