70 lines
1.9 KiB
Python
70 lines
1.9 KiB
Python
|
from modjam import protocol
|
||
|
|
|
||
|
|
|
||
|
|
def test_parse_simple():
|
||
|
|
c = protocol.parse("START|name:foo")
|
||
|
|
assert c.cmd == "START"
|
||
|
|
assert c.target is None
|
||
|
|
assert c.args == {"name": ["foo"]}
|
||
|
|
|
||
|
|
|
||
|
|
def test_parse_with_target():
|
||
|
|
c = protocol.parse("RESULTS:A|name:t1")
|
||
|
|
assert c.cmd == "RESULTS"
|
||
|
|
assert c.target == "A"
|
||
|
|
assert c.args == {"name": ["t1"]}
|
||
|
|
|
||
|
|
|
||
|
|
def test_parse_multi_value():
|
||
|
|
c = protocol.parse("START|bw:62.5,500|sf:7,8,9|name:x")
|
||
|
|
assert c.args["bw"] == ["62.5", "500"]
|
||
|
|
assert c.args["sf"] == ["7", "8", "9"]
|
||
|
|
|
||
|
|
|
||
|
|
def test_parse_empty_returns_none():
|
||
|
|
assert protocol.parse("") is None
|
||
|
|
assert protocol.parse(" \n") is None
|
||
|
|
|
||
|
|
|
||
|
|
def test_parse_lowercase_cmd_normalized():
|
||
|
|
assert protocol.parse("stop").cmd == "STOP"
|
||
|
|
|
||
|
|
|
||
|
|
def test_parse_arg_without_colon_skipped():
|
||
|
|
c = protocol.parse("START|name:x|brokenpart|sf:7")
|
||
|
|
assert c.args == {"name": ["x"], "sf": ["7"]}
|
||
|
|
|
||
|
|
|
||
|
|
def test_encode_scalar_and_list():
|
||
|
|
out = protocol.encode("START", name="x", sf=[7, 8], bw=500)
|
||
|
|
assert out == "START|name:x|sf:7,8|bw:500"
|
||
|
|
|
||
|
|
|
||
|
|
def test_encode_with_target():
|
||
|
|
assert protocol.encode("RESULTS", target="A", name="t1") == "RESULTS:A|name:t1"
|
||
|
|
|
||
|
|
|
||
|
|
def test_encode_skips_none():
|
||
|
|
assert protocol.encode("STOP", target=None, foo=None) == "STOP"
|
||
|
|
|
||
|
|
|
||
|
|
def test_encode_roundtrip():
|
||
|
|
src = protocol.encode("START", name="x", stations=["A", "B", "C"], sf=[7, 8])
|
||
|
|
parsed = protocol.parse(src)
|
||
|
|
assert parsed.cmd == "START"
|
||
|
|
assert parsed.args["name"] == ["x"]
|
||
|
|
assert parsed.args["stations"] == ["A", "B", "C"]
|
||
|
|
assert parsed.args["sf"] == ["7", "8"]
|
||
|
|
|
||
|
|
|
||
|
|
def test_heartbeat_format():
|
||
|
|
assert protocol.encode_heartbeat(123, "A", "IDLE", -100) == "123|A|IDLE|-100"
|
||
|
|
|
||
|
|
|
||
|
|
def test_next_format():
|
||
|
|
assert protocol.encode_next(123, "A", 915.1, 500, 7, 5, 22, 40) == "123|A|next:915.1/500/7/5/22/40"
|
||
|
|
|
||
|
|
|
||
|
|
def test_done_format():
|
||
|
|
assert protocol.encode_done(123, "A", 915.1, 500, 7, 5, 22, 40, 100, 88) == "123|A|done:915.1/500/7/5/22/40 100/88"
|