summaryrefslogtreecommitdiff
path: root/stacks/daemon/test_kiss_bridge.py
blob: 066d1d01033b28cca5abd7ef8e2d5f33cf641897 (plain)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
#!/usr/bin/env python3
"""Offline unit tests for kiss_bridge (no serial hardware)."""
from __future__ import annotations

import sys
from pathlib import Path

sys.path.insert(0, str(Path(__file__).resolve().parent))

from ax25_codec import ax25_build_ui, ax25_crc, ax25_parse_ui, parse_callsign  # noqa: E402
from kiss_bridge import (  # noqa: E402
    KissDecoder,
    format_rx_line,
    kiss_data_frame,
    kiss_encode,
    serial_profile_for_device,
)


def test_callsign_parse() -> None:
    assert parse_callsign("DG1ABC") == ("DG1ABC", 0)
    assert parse_callsign("DK0WC-7") == ("DK0WC", 7)


def test_ax25_roundtrip() -> None:
    frame = ax25_build_ui("DG1ABC", "CQ", b"73")
    parsed = ax25_parse_ui(frame)
    assert parsed is not None
    src, dst, payload = parsed
    assert src == "DG1ABC"
    assert dst == "CQ"
    assert payload == b"73"


def test_kiss_fcs_strip() -> None:
    frame = ax25_build_ui("CB-0", "QST", b"hi")
    pkt = kiss_data_frame(0, frame)
    assert pkt.startswith(b"\xc0")
    assert pkt.endswith(b"\xc0")
    dec = KissDecoder()
    frames = dec.feed(pkt)
    assert len(frames) == 1
    _port, payload = frames[0]
    assert len(payload) == len(frame) - 2
    parsed = ax25_parse_ui(payload)
    assert parsed is not None
    src, dst, info = parsed
    assert src == "CB"
    assert dst == "QST"
    assert info == b"hi"


def test_kiss_escape_roundtrip() -> None:
    raw = bytes([0xC0, 0xDB, 0x00, 0xFF])
    pkt = kiss_encode(0, 0x00, raw)
    dec = KissDecoder()
    frames = dec.feed(pkt)
    assert frames[0][1] == raw


def test_format_rx() -> None:
    line = format_rx_line("DG1ABC", "QST", b"73", ax25_ui=True)
    assert line == "[AX25 UI DG1ABC>QST] 73"


def test_serial_profile_tnc2c() -> None:
    root = str(Path(__file__).resolve().parents[2])
    prof = serial_profile_for_device("tnc2c", root, {})
    assert prof.baud == 19200
    assert prof.dtr_rts is True
    assert prof.kiss_entry == "kiss_on"


def test_serial_profile_pktnc2() -> None:
    root = str(Path(__file__).resolve().parents[2])
    prof = serial_profile_for_device("pktnc2", root, {})
    assert prof.baud == 9600
    assert prof.dtr_rts is False
    assert prof.kiss_entry == "auto"


def test_serial_profile_pccom_kiss_env() -> None:
    import os
    import tempfile

    tmp = Path(tempfile.mkdtemp())
    env = tmp / "local" / "pccom-kiss-serial.env"
    env.parent.mkdir(parents=True)
    env.write_text(
        "PCCOM_KISS_DEV=/dev/ttyUSB2\nPCCOM_KISS_BAUD=9600\nPCCOM_KISS_DTR_RTS=no\n",
        encoding="utf-8",
    )
    old = os.environ.get("MAX25_ROOT")
    os.environ["MAX25_ROOT"] = str(tmp)
    try:
        prof = serial_profile_for_device("pccom-kiss", str(tmp), {})
        assert prof.device == "/dev/ttyUSB2"
        assert prof.baud == 9600
        assert prof.dtr_rts is False
        assert prof.kiss_entry == "none"
    finally:
        if old is None:
            os.environ.pop("MAX25_ROOT", None)
        else:
            os.environ["MAX25_ROOT"] = old


def test_crc_known_vector() -> None:
    body = bytes(range(256))
    assert ax25_crc(body) == 0x303C


def test_kiss_non_data_ignored() -> None:
    pkt = kiss_encode(0, 0x06, b"ignored")  # TXDELAY cmd
    dec = KissDecoder()
    assert dec.feed(pkt) == []


def test_invalid_callsign_tx() -> None:
    from kiss_bridge import KissBridge, SerialProfile

    bridge = KissBridge(SerialProfile(), on_rx=lambda _m: None)
    bridge._kiss_active = True
    bridge._fd = -1  # not used — validate before write
    ok, msg = bridge.transmit("BAD!", "CQ", "hi", ax25_ui=True)
    assert not ok
    assert "invalid" in msg.lower()


def test_rx_thread_not_started_on_open() -> None:
    from kiss_bridge import KissBridge, SerialProfile

    bridge = KissBridge(SerialProfile(), on_rx=lambda _m: None)
    bridge._fd = 1
    bridge.status = "open"
    assert bridge._thread is None


def main() -> int:
    tests = [
        test_callsign_parse,
        test_ax25_roundtrip,
        test_kiss_fcs_strip,
        test_kiss_escape_roundtrip,
        test_format_rx,
        test_serial_profile_tnc2c,
        test_serial_profile_pktnc2,
        test_serial_profile_pccom_kiss_env,
        test_crc_known_vector,
        test_kiss_non_data_ignored,
        test_invalid_callsign_tx,
        test_rx_thread_not_started_on_open,
    ]
    for fn in tests:
        fn()
    print("OK: kiss_bridge unit tests")
    return 0


if __name__ == "__main__":
    sys.exit(main())
git clone -b <branch> https://cgit.mode42.com/<repo>.git
git clone -b <branch> git://cgit.mode42.com/<repo>.git

info@mode42.com