FreeDATA/test/test_modem.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
Test connect frame commands over a high quality simulated audio channel.

Near end-to-end test for sending / receiving connection control frames through the
Modem and modem and back through on the other station. Data injection initiates from the
queue used by the daemon process into and out of the Modem.

Can be invoked from CMake, pytest, coverage or directly.

Uses util_modem_I[RS]S.py in separate process to perform the data transfer.

@author: DJ2LS, N2KIQ
"""

import multiprocessing
import os
import sys
import time

import log_handler
import pytest
import structlog

try:
    import test.util_modem_IRS as irs
    import test.util_modem_ISS as iss
except ImportError:
    import util_modem_IRS as irs
    import util_modem_ISS as iss


# This test is currently a little inconsistent.
@pytest.mark.parametrize("command", ["CONNECT"])
@pytest.mark.flaky(reruns=2)
def test_modem(command, tmp_path):
    log_handler.setup_logging(filename=tmp_path / "test_modem", level="INFO")
    log = structlog.get_logger("test_modem")

    iss_proc = multiprocessing.Process(target=iss.t_arq_iss, args=[command, tmp_path])
    irs_proc = multiprocessing.Process(target=irs.t_arq_irs, args=[command, tmp_path])
    log.debug("Starting threads.")
    iss_proc.start()
    irs_proc.start()

    time.sleep(12)

    log.debug("Terminating threads.")
    irs_proc.terminate()
    iss_proc.terminate()
    irs_proc.join()
    iss_proc.join()

    for idx in range(2):
        try:
            os.unlink(tmp_path / f"hfchannel{idx+1}")
        except FileNotFoundError as fnfe:
            log.debug(f"Unlinking pipe: {fnfe}")

    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__":
    # Run pytest with the current script as the filename.
    ecode = pytest.main(["-s", "-v", sys.argv[0]])
    if ecode == 0:
        print("errors: 0")
    else:
        print(ecode)