-
Notifications
You must be signed in to change notification settings - Fork 25
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Split protocol tests into separate files
- Loading branch information
Showing
11 changed files
with
839 additions
and
702 deletions.
There are no files selected for viewing
This file was deleted.
Oops, something went wrong.
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,55 @@ | ||
"""Unit tests for the UDP protocol handlers.""" # noqa: INP001 | ||
|
||
# ruff: noqa: E501 | ||
|
||
import unittest | ||
import unittest.mock | ||
|
||
from context import ( | ||
GeckoGetChannelProtocolHandler, | ||
) | ||
|
||
PARMS = (1, 2, b"SRCID", b"DESTID") | ||
|
||
|
||
class TestGeckoGetChannelHandler(unittest.TestCase): | ||
"""Channel handler tests.""" | ||
|
||
def test_send_construct_request(self) -> None: | ||
handler = GeckoGetChannelProtocolHandler.request(1, parms=PARMS) | ||
self.assertEqual( | ||
handler.send_bytes, | ||
b"<PACKT><SRCCN>DESTID</SRCCN><DESCN>SRCID</DESCN>" | ||
b"<DATAS>CURCH\x01</DATAS></PACKT>", | ||
) | ||
|
||
def test_send_construct_response(self) -> None: | ||
handler = GeckoGetChannelProtocolHandler.response(10, 33, parms=PARMS) | ||
self.assertEqual( | ||
handler.send_bytes, | ||
b"<PACKT><SRCCN>DESTID</SRCCN><DESCN>SRCID</DESCN>" | ||
b"<DATAS>CHCUR\x0a\x21</DATAS></PACKT>", | ||
) | ||
|
||
def test_recv_can_handle(self) -> None: | ||
handler = GeckoGetChannelProtocolHandler() | ||
self.assertTrue(handler.can_handle(b"CURCH", PARMS)) | ||
self.assertTrue(handler.can_handle(b"CHCUR", PARMS)) | ||
self.assertFalse(handler.can_handle(b"OTHER", PARMS)) | ||
|
||
def test_recv_handle_request(self) -> None: | ||
handler = GeckoGetChannelProtocolHandler() | ||
handler.handle(b"CURCH\x01", PARMS) | ||
self.assertFalse(handler.should_remove_handler) | ||
self.assertEqual(handler._sequence, 1) | ||
|
||
def test_recv_handle_response(self) -> None: | ||
handler = GeckoGetChannelProtocolHandler() | ||
handler.handle(b"CHCUR\x0a\x21", PARMS) | ||
self.assertTrue(handler.should_remove_handler) | ||
self.assertEqual(handler.channel, 10) | ||
self.assertEqual(handler.signal_strength, 33) | ||
|
||
|
||
if __name__ == "__main__": | ||
unittest.main() |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,61 @@ | ||
"""Unit tests for the UDP protocol handlers.""" # noqa: INP001 | ||
|
||
# ruff: noqa: E501 | ||
|
||
import unittest | ||
import unittest.mock | ||
|
||
from context import ( | ||
GeckoConfigFileProtocolHandler, | ||
) | ||
|
||
PARMS = (1, 2, b"SRCID", b"DESTID") | ||
|
||
|
||
class TestGeckoConfigFileHandler(unittest.TestCase): | ||
"""Config file tests.""" | ||
|
||
def test_send_construct_request(self) -> None: | ||
handler = GeckoConfigFileProtocolHandler.request(1, parms=PARMS) | ||
self.assertEqual( | ||
handler.send_bytes, | ||
b"<PACKT><SRCCN>DESTID</SRCCN><DESCN>SRCID</DESCN>" | ||
b"<DATAS>SFILE\x01</DATAS></PACKT>", | ||
) | ||
|
||
def test_send_construct_response(self) -> None: | ||
handler = GeckoConfigFileProtocolHandler.response("inXM", 7, 8, parms=PARMS) | ||
self.assertEqual( | ||
handler.send_bytes, | ||
b"<PACKT><SRCCN>DESTID</SRCCN><DESCN>SRCID</DESCN>" | ||
b"<DATAS>FILES,inXM_C07.xml,inXM_S08.xml</DATAS></PACKT>", | ||
) | ||
|
||
def test_recv_can_handle(self) -> None: | ||
handler = GeckoConfigFileProtocolHandler() | ||
self.assertTrue(handler.can_handle(b"SFILE", PARMS)) | ||
self.assertTrue(handler.can_handle(b"FILES", PARMS)) | ||
self.assertFalse(handler.can_handle(b"OTHER", PARMS)) | ||
|
||
def test_recv_handle_request(self) -> None: | ||
handler = GeckoConfigFileProtocolHandler() | ||
handler.handle(b"SFILE\x01", PARMS) | ||
self.assertFalse(handler.should_remove_handler) | ||
self.assertEqual(handler._sequence, 1) | ||
|
||
def test_recv_handle_response(self) -> None: | ||
handler = GeckoConfigFileProtocolHandler() | ||
handler.handle(b"FILES,inXM_C09.xml,inXM_S09.xml", PARMS) | ||
self.assertTrue(handler.should_remove_handler) | ||
self.assertEqual(handler.plateform_key, "inXM") | ||
self.assertEqual(handler.config_version, 9) | ||
self.assertEqual(handler.log_version, 9) | ||
|
||
@unittest.expectedFailure | ||
def test_recv_handle_response_error(self) -> None: | ||
handler = GeckoConfigFileProtocolHandler() | ||
self.assertTrue(handler.handle(b"FILES,inXM_C09.xml,inYE_S09.xml", PARMS)) | ||
|
||
|
||
if __name__ == "__main__": | ||
unittest.main() |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,79 @@ | ||
"""Unit tests for the UDP protocol handlers.""" # noqa: INP001 | ||
|
||
# ruff: noqa: E501 | ||
|
||
import unittest | ||
import unittest.mock | ||
|
||
from context import ( | ||
GeckoHelloProtocolHandler, | ||
) | ||
|
||
PARMS = (1, 2, b"SRCID", b"DESTID") | ||
|
||
|
||
class TestGeckoHelloHandler(unittest.TestCase): | ||
"""Test the GeckoHelloProtocol classes.""" | ||
|
||
def setUp(self) -> None: | ||
"""Set up the class.""" | ||
self.handler = GeckoHelloProtocolHandler(b"") | ||
|
||
def test_send_broadcast_construct(self) -> None: | ||
handler = GeckoHelloProtocolHandler.broadcast() | ||
self.assertEqual(handler.send_bytes, b"<HELLO>1</HELLO>") | ||
|
||
def test_send_client_construct(self) -> None: | ||
handler = GeckoHelloProtocolHandler.client(b"CLIENT") | ||
self.assertEqual(handler.send_bytes, b"<HELLO>CLIENT</HELLO>") | ||
|
||
def test_send_response_construct(self) -> None: | ||
handler = GeckoHelloProtocolHandler.response(b"SPA", "Name") | ||
self.assertEqual(handler.send_bytes, b"<HELLO>SPA|Name</HELLO>") | ||
|
||
def test_recv_can_handle(self) -> None: | ||
self.assertTrue(self.handler.can_handle(b"<HELLO></HELLO>", ())) | ||
self.assertFalse(self.handler.can_handle(b"<HELLO></HELLO", ())) | ||
self.assertFalse(self.handler.can_handle(b"<HELLO></HELLO> ", ())) | ||
self.assertFalse(self.handler.can_handle(b"<GOODBYE>", ())) | ||
|
||
def test_recv_broadcast(self) -> None: | ||
self.assertFalse(self.handler.handle(b"<HELLO>1</HELLO>", ())) | ||
self.assertTrue(self.handler.was_broadcast_discovery) | ||
self.assertIsNone(self.handler._client_identifier) | ||
self.assertIsNone(self.handler._spa_identifier) | ||
self.assertIsNone(self.handler._spa_name) | ||
|
||
def test_recv_client_ios(self) -> None: | ||
self.assertFalse(self.handler.handle(b"<HELLO>IOSCLIENT</HELLO>", ())) | ||
self.assertFalse(self.handler.was_broadcast_discovery) | ||
self.assertEqual(self.handler.client_identifier, b"IOSCLIENT") | ||
self.assertIsNone(self.handler._spa_identifier) | ||
self.assertIsNone(self.handler._spa_name) | ||
|
||
def test_recv_client_android(self) -> None: | ||
self.assertFalse(self.handler.handle(b"<HELLO>ANDCLIENT</HELLO>", ())) | ||
self.assertFalse(self.handler.was_broadcast_discovery) | ||
self.assertEqual(self.handler.client_identifier, b"ANDCLIENT") | ||
self.assertIsNone(self.handler._spa_identifier) | ||
self.assertIsNone(self.handler._spa_name) | ||
|
||
@unittest.expectedFailure | ||
def test_recv_client_unknown(self) -> None: | ||
self.handler.handle(b"<HELLO>UNKCLIENT</HELLO>", ()) | ||
|
||
def test_recv_response(self) -> None: | ||
self.assertFalse(self.handler.handle(b"<HELLO>SPA|Name</HELLO>", ())) | ||
self.assertFalse(self.handler.was_broadcast_discovery) | ||
self.assertIsNone(self.handler._client_identifier) | ||
self.assertEqual(self.handler.spa_identifier, b"SPA") | ||
self.assertEqual(self.handler.spa_name, "Name") | ||
|
||
def test_recv_can_handle_multiple(self) -> None: | ||
self.test_recv_response() | ||
self.test_recv_client_android() | ||
self.test_recv_broadcast() | ||
|
||
|
||
if __name__ == "__main__": | ||
unittest.main() |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,78 @@ | ||
"""Unit tests for the UDP protocol handlers.""" # noqa: INP001 | ||
|
||
# ruff: noqa: E501 | ||
|
||
import unittest | ||
import unittest.mock | ||
|
||
from context import ( | ||
GeckoPackCommandProtocolHandler, | ||
) | ||
|
||
PARMS = (1, 2, b"SRCID", b"DESTID") | ||
|
||
|
||
class TestGeckoPackCommandHandlers(unittest.TestCase): | ||
"""Pack Command tests.""" | ||
|
||
def test_send_construct_key_press(self) -> None: | ||
handler = GeckoPackCommandProtocolHandler.keypress(1, 6, 1, parms=PARMS) | ||
self.assertEqual( | ||
handler.send_bytes, | ||
b"<PACKT><SRCCN>DESTID</SRCCN><DESCN>SRCID</DESCN>" | ||
b"<DATAS>SPACK\x01\x06\x02\x39\x01</DATAS></PACKT>", | ||
) | ||
|
||
def test_send_construct_set_value(self) -> None: | ||
handler = GeckoPackCommandProtocolHandler.set_value( | ||
1, 6, 9, 9, 15, 2, 702, parms=PARMS | ||
) | ||
self.assertEqual( | ||
handler.send_bytes, | ||
b"<PACKT><SRCCN>DESTID</SRCCN><DESCN>SRCID</DESCN>" | ||
b"<DATAS>SPACK\x01\x06\x07\x46\x09\x09\x00\x0f\x02\xbe</DATAS></PACKT>", | ||
) | ||
|
||
def test_send_construct_response(self) -> None: | ||
handler = GeckoPackCommandProtocolHandler.response(parms=PARMS) | ||
self.assertEqual( | ||
handler.send_bytes, | ||
b"<PACKT><SRCCN>DESTID</SRCCN><DESCN>SRCID</DESCN>" | ||
b"<DATAS>PACKS</DATAS></PACKT>", | ||
) | ||
|
||
def test_recv_can_handle(self) -> None: | ||
handler = GeckoPackCommandProtocolHandler() | ||
self.assertTrue(handler.can_handle(b"SPACK", PARMS)) | ||
self.assertTrue(handler.can_handle(b"PACKS", PARMS)) | ||
self.assertFalse(handler.can_handle(b"OTHER", PARMS)) | ||
|
||
def test_recv_handle_key_press(self) -> None: | ||
handler = GeckoPackCommandProtocolHandler() | ||
handler.handle(b"SPACK\x01\x06\x02\x39\x01", PARMS) | ||
self.assertFalse(handler.should_remove_handler) | ||
self.assertEqual(handler._sequence, 1) | ||
self.assertTrue(handler.is_key_press) | ||
self.assertEqual(handler.keycode, 1) | ||
self.assertFalse(handler.is_set_value) | ||
|
||
def test_recv_handle_set_value(self) -> None: | ||
handler = GeckoPackCommandProtocolHandler() | ||
handler.handle(b"SPACK\x01\x06\x07\x46\x09\x09\x00\x0f\x02\xbe", PARMS) | ||
self.assertFalse(handler.should_remove_handler) | ||
self.assertEqual(handler._sequence, 1) | ||
self.assertFalse(handler.is_key_press) | ||
self.assertTrue(handler.is_set_value) | ||
self.assertEqual(handler.position, 15) | ||
self.assertEqual(handler.new_data, b"\x02\xbe") | ||
|
||
def test_recv_handle_response(self) -> None: | ||
handler = GeckoPackCommandProtocolHandler() | ||
handler.handle(b"PACKS", PARMS) | ||
self.assertTrue(handler.should_remove_handler) | ||
self.assertFalse(handler.is_key_press) | ||
self.assertFalse(handler.is_set_value) | ||
|
||
|
||
if __name__ == "__main__": | ||
unittest.main() |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,52 @@ | ||
"""Unit tests for the UDP protocol handlers.""" # noqa: INP001 | ||
|
||
# ruff: noqa: E501 | ||
|
||
import unittest | ||
import unittest.mock | ||
|
||
from context import ( | ||
GeckoPacketProtocolHandler, | ||
) | ||
|
||
PARMS = (1, 2, b"SRCID", b"DESTID") | ||
|
||
|
||
class TestGeckoPacketHandler(unittest.TestCase): | ||
"""Packet handler tests.""" | ||
|
||
def setUp(self) -> None: | ||
"""Set up the class.""" | ||
self.handler = GeckoPacketProtocolHandler(content=b"CONTENT", parms=PARMS) | ||
|
||
def test_recv_can_handle(self) -> None: | ||
self.assertTrue(self.handler.can_handle(b"<PACKT></PACKT>", ())) | ||
self.assertFalse(self.handler.can_handle(b"<PACKT></PACKT", ())) | ||
self.assertFalse(self.handler.can_handle(b"<PACKT></PACKT> ", ())) | ||
self.assertFalse(self.handler.can_handle(b"<SOMETHING>", ())) | ||
|
||
def test_recv_extract_ok(self) -> None: | ||
self.assertFalse( | ||
self.handler.handle( | ||
b"<PACKT><SRCCN>SRCID</SRCCN><DESCN>DESTID</DESCN>" | ||
b"<DATAS>DATA</DATAS></PACKT>", | ||
(1, 2), | ||
) | ||
) | ||
assert self.handler.parms is not None | ||
self.assertTupleEqual(self.handler.parms, PARMS) | ||
self.assertEqual(self.handler.packet_content, b"DATA") | ||
|
||
def test_send_construct(self) -> None: | ||
self.assertEqual( | ||
self.handler.send_bytes, | ||
b"<PACKT><SRCCN>DESTID</SRCCN><DESCN>SRCID</DESCN>" | ||
b"<DATAS>CONTENT</DATAS></PACKT>", | ||
) | ||
assert self.handler.parms is not None | ||
self.assertEqual(self.handler.parms[0], 1) | ||
self.assertEqual(self.handler.parms[1], 2) | ||
|
||
|
||
if __name__ == "__main__": | ||
unittest.main() |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,47 @@ | ||
"""Unit tests for the UDP protocol handlers.""" # noqa: INP001 | ||
|
||
# ruff: noqa: E501 | ||
|
||
import unittest | ||
import unittest.mock | ||
|
||
from context import ( | ||
GeckoPingProtocolHandler, | ||
) | ||
|
||
PARMS = (1, 2, b"SRCID", b"DESTID") | ||
|
||
|
||
class TestGeckoPingHandler(unittest.TestCase): | ||
"""Ping handler tests.""" | ||
|
||
def test_send_construct_request(self) -> None: | ||
handler = GeckoPingProtocolHandler.request(parms=PARMS) | ||
self.assertEqual( | ||
handler.send_bytes, | ||
b"<PACKT><SRCCN>DESTID</SRCCN><DESCN>SRCID</DESCN>" | ||
b"<DATAS>APING</DATAS></PACKT>", | ||
) | ||
|
||
def test_send_construct_response(self) -> None: | ||
handler = GeckoPingProtocolHandler.response(parms=PARMS) | ||
self.assertEqual( | ||
handler.send_bytes, | ||
b"<PACKT><SRCCN>DESTID</SRCCN><DESCN>SRCID</DESCN>" | ||
b"<DATAS>APING\x00</DATAS></PACKT>", | ||
) | ||
|
||
def test_recv_can_handle(self) -> None: | ||
handler = GeckoPingProtocolHandler.request(parms=PARMS) | ||
self.assertTrue(handler.can_handle(b"APING", PARMS)) | ||
self.assertIsNone(handler._sequence) | ||
|
||
def test_recv_handle(self) -> None: | ||
handler = GeckoPingProtocolHandler.request(parms=PARMS) | ||
handler.handle(b"APING\x00", PARMS) | ||
self.assertFalse(handler.should_remove_handler) | ||
self.assertEqual(handler._sequence, 0) | ||
|
||
|
||
if __name__ == "__main__": | ||
unittest.main() |
Oops, something went wrong.