|
1 | 1 | """Meshtastic unit tests for stream_interface.py""" |
2 | 2 |
|
3 | 3 | import logging |
4 | | -import re |
| 4 | +#import re |
5 | 5 |
|
6 | 6 | from unittest.mock import MagicMock |
7 | 7 | import pytest |
@@ -35,48 +35,49 @@ def test_StreamInterface_with_noProto(caplog): |
35 | 35 | assert data == test_data |
36 | 36 |
|
37 | 37 |
|
38 | | -## Note: This takes a bit, so moving from unit to slow |
39 | | -## Tip: If you want to see the print output, run with '-s' flag: |
40 | | -## pytest -s meshtastic/tests/test_stream_interface.py::test_sendToRadioImpl |
41 | | -@pytest.mark.unitslow |
42 | | -@pytest.mark.usefixtures("reset_globals") |
43 | | -def test_sendToRadioImpl(caplog): |
44 | | - """Test _sendToRadioImpl()""" |
45 | | - |
46 | | -# def add_header(b): |
47 | | -# """Add header stuffs for radio""" |
48 | | -# bufLen = len(b) |
49 | | -# header = bytes([START1, START2, (bufLen >> 8) & 0xff, bufLen & 0xff]) |
50 | | -# return header + b |
51 | | - |
52 | | - # captured raw bytes of a Heltec2.1 radio with 2 channels (primary and a secondary channel named "gpio") |
53 | | - raw_1_my_info = b'\x1a,\x08\xdc\x8c\xd5\xc5\x02\x18\r2\x0e1.2.49.5354c49P\x15]\xe1%\x17Eh\xe0\xa7\x12p\xe8\x9d\x01x\x08\x90\x01\x01' |
54 | | - raw_2_node_info = b'"9\x08\xdc\x8c\xd5\xc5\x02\x12(\n\t!28b5465c\x12\x0cUnknown 465c\x1a\x03?5C"\x06$o(\xb5F\\0\n\x1a\x02 1%M<\xc6a' |
55 | | - # pylint: disable=C0301 |
56 | | - raw_3_node_info = b'"C\x08\xa4\x8c\xd5\xc5\x02\x12(\n\t!28b54624\x12\x0cUnknown 4624\x1a\x03?24"\x06$o(\xb5F$0\n\x1a\x07 5MH<\xc6a%G<\xc6a=\x00\x00\xc0@' |
57 | | - raw_4_complete = b'@\xcf\xe5\xd1\x8c\x0e' |
58 | | - # pylint: disable=C0301 |
59 | | - raw_5_prefs = b'Z6\r\\F\xb5(\x15\\F\xb5("\x1c\x08\x06\x12\x13*\x11\n\x0f0\x84\x07P\xac\x02\x88\x01\x01\xb0\t#\xb8\t\x015]$\xddk5\xd5\x7f!b=M<\xc6aP\x03`F' |
60 | | - # pylint: disable=C0301 |
61 | | - raw_6_channel0 = b'Z.\r\\F\xb5(\x15\\F\xb5("\x14\x08\x06\x12\x0b:\t\x12\x05\x18\x01"\x01\x01\x18\x015^$\xddk5\xd6\x7f!b=M<\xc6aP\x03`F' |
62 | | - # pylint: disable=C0301 |
63 | | - raw_7_channel1 = b'ZS\r\\F\xb5(\x15\\F\xb5("9\x08\x06\x120:.\x08\x01\x12(" \xb4&\xb3\xc7\x06\xd8\xe39%\xba\xa5\xee\x8eH\x06\xf6\xf4H\xe8\xd5\xc1[ao\xb5Y\\\xb4"\xafmi*\x04gpio\x18\x025_$\xddk5\xd7\x7f!b=M<\xc6aP\x03`F' |
64 | | - raw_8_channel2 = b'Z)\r\\F\xb5(\x15\\F\xb5("\x0f\x08\x06\x12\x06:\x04\x08\x02\x12\x005`$\xddk5\xd8\x7f!b=M<\xc6aP\x03`F' |
65 | | - raw_blank = b'' |
66 | | - |
67 | | - test_data = b'hello' |
68 | | - stream = MagicMock() |
69 | | - #stream.read.return_value = add_header(test_data) |
70 | | - stream.read.side_effect = [ raw_1_my_info, raw_2_node_info, raw_3_node_info, raw_4_complete, |
71 | | - raw_5_prefs, raw_6_channel0, raw_7_channel1, raw_8_channel2, |
72 | | - raw_blank, raw_blank] |
73 | | - toRadio = MagicMock() |
74 | | - toRadio.SerializeToString.return_value = test_data |
75 | | - with caplog.at_level(logging.DEBUG): |
76 | | - iface = StreamInterface(noProto=True, connectNow=False) |
77 | | - iface.stream = stream |
78 | | - iface.connect() |
79 | | - iface._sendToRadioImpl(toRadio) |
80 | | - assert re.search(r'Sending: ', caplog.text, re.MULTILINE) |
81 | | - assert re.search(r'reading character', caplog.text, re.MULTILINE) |
82 | | - assert re.search(r'In reader loop', caplog.text, re.MULTILINE) |
| 38 | +# TODO |
| 39 | +### Note: This takes a bit, so moving from unit to slow |
| 40 | +### Tip: If you want to see the print output, run with '-s' flag: |
| 41 | +### pytest -s meshtastic/tests/test_stream_interface.py::test_sendToRadioImpl |
| 42 | +#@pytest.mark.unitslow |
| 43 | +#@pytest.mark.usefixtures("reset_globals") |
| 44 | +#def test_sendToRadioImpl(caplog): |
| 45 | +# """Test _sendToRadioImpl()""" |
| 46 | +# |
| 47 | +## def add_header(b): |
| 48 | +## """Add header stuffs for radio""" |
| 49 | +## bufLen = len(b) |
| 50 | +## header = bytes([START1, START2, (bufLen >> 8) & 0xff, bufLen & 0xff]) |
| 51 | +## return header + b |
| 52 | +# |
| 53 | +# # captured raw bytes of a Heltec2.1 radio with 2 channels (primary and a secondary channel named "gpio") |
| 54 | +# raw_1_my_info = b'\x1a,\x08\xdc\x8c\xd5\xc5\x02\x18\r2\x0e1.2.49.5354c49P\x15]\xe1%\x17Eh\xe0\xa7\x12p\xe8\x9d\x01x\x08\x90\x01\x01' |
| 55 | +# raw_2_node_info = b'"9\x08\xdc\x8c\xd5\xc5\x02\x12(\n\t!28b5465c\x12\x0cUnknown 465c\x1a\x03?5C"\x06$o(\xb5F\\0\n\x1a\x02 1%M<\xc6a' |
| 56 | +# # pylint: disable=C0301 |
| 57 | +# raw_3_node_info = b'"C\x08\xa4\x8c\xd5\xc5\x02\x12(\n\t!28b54624\x12\x0cUnknown 4624\x1a\x03?24"\x06$o(\xb5F$0\n\x1a\x07 5MH<\xc6a%G<\xc6a=\x00\x00\xc0@' |
| 58 | +# raw_4_complete = b'@\xcf\xe5\xd1\x8c\x0e' |
| 59 | +# # pylint: disable=C0301 |
| 60 | +# raw_5_prefs = b'Z6\r\\F\xb5(\x15\\F\xb5("\x1c\x08\x06\x12\x13*\x11\n\x0f0\x84\x07P\xac\x02\x88\x01\x01\xb0\t#\xb8\t\x015]$\xddk5\xd5\x7f!b=M<\xc6aP\x03`F' |
| 61 | +# # pylint: disable=C0301 |
| 62 | +# raw_6_channel0 = b'Z.\r\\F\xb5(\x15\\F\xb5("\x14\x08\x06\x12\x0b:\t\x12\x05\x18\x01"\x01\x01\x18\x015^$\xddk5\xd6\x7f!b=M<\xc6aP\x03`F' |
| 63 | +# # pylint: disable=C0301 |
| 64 | +# raw_7_channel1 = b'ZS\r\\F\xb5(\x15\\F\xb5("9\x08\x06\x120:.\x08\x01\x12(" \xb4&\xb3\xc7\x06\xd8\xe39%\xba\xa5\xee\x8eH\x06\xf6\xf4H\xe8\xd5\xc1[ao\xb5Y\\\xb4"\xafmi*\x04gpio\x18\x025_$\xddk5\xd7\x7f!b=M<\xc6aP\x03`F' |
| 65 | +# raw_8_channel2 = b'Z)\r\\F\xb5(\x15\\F\xb5("\x0f\x08\x06\x12\x06:\x04\x08\x02\x12\x005`$\xddk5\xd8\x7f!b=M<\xc6aP\x03`F' |
| 66 | +# raw_blank = b'' |
| 67 | +# |
| 68 | +# test_data = b'hello' |
| 69 | +# stream = MagicMock() |
| 70 | +# #stream.read.return_value = add_header(test_data) |
| 71 | +# stream.read.side_effect = [ raw_1_my_info, raw_2_node_info, raw_3_node_info, raw_4_complete, |
| 72 | +# raw_5_prefs, raw_6_channel0, raw_7_channel1, raw_8_channel2, |
| 73 | +# raw_blank, raw_blank] |
| 74 | +# toRadio = MagicMock() |
| 75 | +# toRadio.SerializeToString.return_value = test_data |
| 76 | +# with caplog.at_level(logging.DEBUG): |
| 77 | +# iface = StreamInterface(noProto=True, connectNow=False) |
| 78 | +# iface.stream = stream |
| 79 | +# iface.connect() |
| 80 | +# iface._sendToRadioImpl(toRadio) |
| 81 | +# assert re.search(r'Sending: ', caplog.text, re.MULTILINE) |
| 82 | +# assert re.search(r'reading character', caplog.text, re.MULTILINE) |
| 83 | +# assert re.search(r'In reader loop', caplog.text, re.MULTILINE) |
0 commit comments