diff --git a/tcmenu/__init__.py b/tcmenu/__init__.py index 7dd5b27..41a431c 100644 --- a/tcmenu/__init__.py +++ b/tcmenu/__init__.py @@ -6,4 +6,4 @@ "__version__", ] -__version__ = "4.1.1" +__version__ = "4.1.2" diff --git a/tcmenu/remote/commands/dialog_mode.py b/tcmenu/remote/commands/dialog_mode.py index e2bba0f..15aafec 100644 --- a/tcmenu/remote/commands/dialog_mode.py +++ b/tcmenu/remote/commands/dialog_mode.py @@ -15,3 +15,12 @@ class DialogMode(Enum): """ Perform the following action on the dialog. """ ACTION = auto() + + @staticmethod + def from_string(mode: str) -> 'DialogMode': + if mode == "S": + return DialogMode.SHOW + elif mode == "H": + return DialogMode.HIDE + else: + return DialogMode.ACTION diff --git a/tcmenu/remote/commands/menu_button_type.py b/tcmenu/remote/commands/menu_button_type.py index 9a6c584..ba3ba62 100644 --- a/tcmenu/remote/commands/menu_button_type.py +++ b/tcmenu/remote/commands/menu_button_type.py @@ -32,3 +32,10 @@ def button_name(self) -> str: @property def type_value(self) -> int: return self._value_.type_value + + @staticmethod + def from_id(button_id: int): + for button_type in MenuButtonType: + if button_type.type_value == button_id: + return button_type + return MenuButtonType.NONE diff --git a/tcmenu/remote/commands/menu_heartbeat_command.py b/tcmenu/remote/commands/menu_heartbeat_command.py index 461d048..675fe67 100644 --- a/tcmenu/remote/commands/menu_heartbeat_command.py +++ b/tcmenu/remote/commands/menu_heartbeat_command.py @@ -10,9 +10,9 @@ class MenuHeartbeatCommand(MenuCommand): # noinspection PyArgumentList class HeartbeatMode(Enum): - START = auto() - NORMAL = auto() - END = auto() + NORMAL = 0 + START = 1 + END = 2 heartbeat_interval: int @@ -22,5 +22,13 @@ class HeartbeatMode(Enum): def command_type(self) -> MessageField: return MenuCommandType.HEARTBEAT.message_field + @staticmethod + def from_id(heartbeat_id: int): + return ( + MenuHeartbeatCommand.HeartbeatMode(heartbeat_id) + if heartbeat_id in (item.value for item in MenuHeartbeatCommand.HeartbeatMode) + else MenuHeartbeatCommand.HeartbeatMode.NORMAL + ) + def __repr__(self): return "MenuHeartbeatCommand{}" diff --git a/tcmenu/remote/protocol/configurable_protocol_converter.py b/tcmenu/remote/protocol/configurable_protocol_converter.py index ade5bc8..8ac7a12 100644 --- a/tcmenu/remote/protocol/configurable_protocol_converter.py +++ b/tcmenu/remote/protocol/configurable_protocol_converter.py @@ -82,7 +82,7 @@ def add_raw_out_processor(self, field, processor, clazz): def from_channel(self, buffer: io.BytesIO) -> MenuCommand: proto_id = buffer.read(1)[0] - protocol = CommandProtocol(proto_id).value + protocol = CommandProtocol(proto_id) msg_type = self._get_msg_type_from_buffer(buffer) cmd_type = MessageField.from_id(msg_type) @@ -94,11 +94,11 @@ def from_channel(self, buffer: io.BytesIO) -> MenuCommand: parser = TagValTextParser(buffer) logging.debug(f"Protocol convert in: {parser}") return self._tag_val_incoming_parsers[cmd_type](parser) - elif protocol == CommandProtocol.RAW_BIN_PROTOCOL and cmd_type in self._raw_incoming_parsers: + elif protocol.value == CommandProtocol.RAW_BIN_PROTOCOL and cmd_type in self._raw_incoming_parsers: length = int.from_bytes(buffer.read(4), byteorder="big") return self._raw_incoming_parsers[cmd_type](buffer, length) else: - raise TcProtocolException(f"Unknown protocol used in message {protocol}") + raise TcProtocolException(f"Unknown protocol used in message: {protocol.name}") def to_channel(self, buffer: io.BytesIO, command: MenuCommand) -> None: raw_processor = self._raw_output_writers.get(command.command_type) diff --git a/tcmenu/remote/protocol/message_field.py b/tcmenu/remote/protocol/message_field.py index 51f15f9..cbebe9b 100644 --- a/tcmenu/remote/protocol/message_field.py +++ b/tcmenu/remote/protocol/message_field.py @@ -8,12 +8,12 @@ class MessageField: second_byte: str - _ALL_FIELDS_MAP: ClassVar[dict[str, "MessageField"]] = dict() + _ALL_FIELDS_DICT: ClassVar[dict[str, "MessageField"]] = {} def __post_init__(self): - if self.id in MessageField._ALL_FIELDS_MAP.keys(): + if self.id in MessageField._ALL_FIELDS_DICT.keys(): raise ValueError(f"Duplicate key: {self.id}") - MessageField._ALL_FIELDS_MAP[self.id] = self + MessageField._ALL_FIELDS_DICT[self.id] = self @property def high(self) -> str: @@ -29,8 +29,8 @@ def id(self) -> str: @staticmethod def from_id(field_id: str) -> "MessageField": - if field_id in MessageField._ALL_FIELDS_MAP.keys(): - return MessageField._ALL_FIELDS_MAP[field_id] + if field_id in MessageField._ALL_FIELDS_DICT.keys(): + return MessageField._ALL_FIELDS_DICT[field_id] else: raise ValueError("An unknown message type was generated.") diff --git a/tcmenu/remote/protocol/tag_val_menu_command_processors.py b/tcmenu/remote/protocol/tag_val_menu_command_processors.py index e66291e..4df695d 100644 --- a/tcmenu/remote/protocol/tag_val_menu_command_processors.py +++ b/tcmenu/remote/protocol/tag_val_menu_command_processors.py @@ -1,5 +1,7 @@ import io +import uuid +from tcmenu.remote.commands.command_factory import CommandFactory from tcmenu.remote.commands.dialog_mode import DialogMode from tcmenu.remote.commands.menu_acknowledgement_command import MenuAcknowledgementCommand from tcmenu.remote.commands.menu_boot_commands import ( @@ -24,6 +26,9 @@ from tcmenu.remote.commands.menu_heartbeat_command import MenuHeartbeatCommand from tcmenu.remote.commands.menu_join_command import MenuJoinCommand from tcmenu.remote.commands.menu_pairing_command import MenuPairingCommand +from tcmenu.remote.protocol.correlation_id import CorrelationId +from tcmenu.remote.protocol.protocol_util import ProtocolUtil +from tcmenu.remote.protocol.tag_val_menu_fields import TagValMenuFields from tcmenu.remote.protocol.tag_val_text_parser import TagValTextParser @@ -172,11 +177,25 @@ def add_handlers_to_protocol(proto: "ConfigurableProtocolConverter"): @staticmethod def _process_join(parser: TagValTextParser) -> MenuCommand: - pass + uuid_str = parser.get_value(TagValMenuFields.KEY_UUID_FIELD.value, "") + uuid_val = uuid.UUID(uuid_str) if len(uuid_str) > 0 else uuid.uuid4() + + return MenuJoinCommand( + my_name=parser.get_value(TagValMenuFields.KEY_NAME_FIELD.value), + api_version=parser.get_value_as_int(TagValMenuFields.KEY_VER_FIELD.value), + platform=ProtocolUtil.from_key_to_api_platform( + parser.get_value_as_int(TagValMenuFields.KEY_PLATFORM_ID.value) + ), + app_uuid=uuid_val, + serial_number=parser.get_value_as_int(TagValMenuFields.KEY_SERIAL_NO.value, 0), + ) @staticmethod def _process_heartbeat(parser: TagValTextParser) -> MenuCommand: - pass + return CommandFactory.new_heartbeat_command( + frequency=parser.get_value_as_int(TagValMenuFields.HB_FREQUENCY_FIELD.value, 10000), + mode=MenuHeartbeatCommand.from_id(parser.get_value_as_int(TagValMenuFields.HB_MODE_FIELD.value, 0)), + ) @staticmethod def _process_bootstrap(parser: TagValTextParser) -> MenuCommand: @@ -240,15 +259,17 @@ def _process_pairing_request(parser: TagValTextParser) -> MenuCommand: @staticmethod def _process_dialog_update(parser: TagValTextParser) -> MenuCommand: - pass + cor = parser.get_value(TagValMenuFields.KEY_CORRELATION_FIELD.value, "") + correlation_id = CorrelationId.EMPTY_CORRELATION if len(cor) == 0 else CorrelationId.from_string(cor) - @staticmethod - def _as_dialog_mode(mode: str) -> DialogMode: - pass - - @staticmethod - def _as_button(req: int) -> MenuButtonType: - pass + return CommandFactory.new_dialog_command( + mode=DialogMode.from_string(parser.get_value(TagValMenuFields.KEY_MODE_FIELD.value)), + header=parser.get_value(TagValMenuFields.KEY_HEADER_FIELD.value, ""), + message=parser.get_value(TagValMenuFields.KEY_BUFFER_FIELD.value, ""), + button1=MenuButtonType.from_id(parser.get_value_as_int(TagValMenuFields.KEY_BUTTON1_FIELD.value, 0)), + button2=MenuButtonType.from_id(parser.get_value_as_int(TagValMenuFields.KEY_BUTTON2_FIELD.value, 0)), + correlation_id=correlation_id, + ) @staticmethod def _write_join(buffer: io.StringIO, command: MenuCommand) -> None: diff --git a/test/remote/commands/test_dialog_mode.py b/test/remote/commands/test_dialog_mode.py index c238387..8cfef51 100644 --- a/test/remote/commands/test_dialog_mode.py +++ b/test/remote/commands/test_dialog_mode.py @@ -10,3 +10,10 @@ def test_enum_values(): def test_enum_uniqueness(): values = [mode.value for mode in DialogMode] assert len(values) == len(set(values)), "Enum values are not unique" + + +def test_dialog_mode_from_string(): + assert DialogMode.SHOW == DialogMode.from_string("S") + assert DialogMode.HIDE == DialogMode.from_string("H") + assert DialogMode.ACTION == DialogMode.from_string("") + assert DialogMode.ACTION == DialogMode.from_string("AA") diff --git a/test/remote/commands/test_menu_button_type.py b/test/remote/commands/test_menu_button_type.py index 868f2c6..f9b06b4 100644 --- a/test/remote/commands/test_menu_button_type.py +++ b/test/remote/commands/test_menu_button_type.py @@ -13,3 +13,13 @@ def test_button_not_set(): assert type(button.button_name) == str assert len(button.button_name) == 0 assert type(button.type_value) == int + + +def test_button_type_from_id(): + assert MenuButtonType.OK == MenuButtonType.from_id(0) + assert MenuButtonType.ACCEPT == MenuButtonType.from_id(1) + assert MenuButtonType.CANCEL == MenuButtonType.from_id(2) + assert MenuButtonType.CLOSE == MenuButtonType.from_id(3) + assert MenuButtonType.NONE == MenuButtonType.from_id(4) + assert MenuButtonType.NONE == MenuButtonType.from_id(5) + assert MenuButtonType.NONE == MenuButtonType.from_id(100) diff --git a/test/remote/commands/test_menu_heartbeat_command.py b/test/remote/commands/test_menu_heartbeat_command.py index 7e0bb5e..836fc11 100644 --- a/test/remote/commands/test_menu_heartbeat_command.py +++ b/test/remote/commands/test_menu_heartbeat_command.py @@ -9,3 +9,10 @@ def test_menu_heartbeat_command(): assert command.mode == MenuHeartbeatCommand.HeartbeatMode.START assert isinstance(command.command_type, MessageField) is True assert command.command_type.id == "HB" + + +def test_menu_heartbeat_command_from_id(): + assert MenuHeartbeatCommand.HeartbeatMode.START == MenuHeartbeatCommand.from_id(1) + assert MenuHeartbeatCommand.HeartbeatMode.END == MenuHeartbeatCommand.from_id(2) + assert MenuHeartbeatCommand.HeartbeatMode.NORMAL == MenuHeartbeatCommand.from_id(0) + assert MenuHeartbeatCommand.HeartbeatMode.NORMAL == MenuHeartbeatCommand.from_id(100) diff --git a/test/remote/protocol/test_message_field.py b/test/remote/protocol/test_message_field.py index e91560f..cc8d96a 100644 --- a/test/remote/protocol/test_message_field.py +++ b/test/remote/protocol/test_message_field.py @@ -3,8 +3,25 @@ from tcmenu.remote.protocol.message_field import MessageField +# noinspection PyProtectedMember +@pytest.fixture(autouse=True) +def message_field_fixture(): + """ + This code temporarily changes internal dictionary of a MessageField. + We have to ensure that original entries are set back to avoid + test failures. + """ + # Setup + entries: dict[str, "MessageField"] = MessageField._ALL_FIELDS_DICT.copy() + MessageField._ALL_FIELDS_DICT.clear() + + yield + + # Teardown + MessageField._ALL_FIELDS_DICT = entries + + def test_message_field(): - MessageField._ALL_FIELDS_MAP.clear() message = MessageField("R", "V") assert message.first_byte is "R" @@ -16,7 +33,6 @@ def test_message_field(): def test_message_field_duplicate_entry(): - MessageField._ALL_FIELDS_MAP.clear() MessageField("A", "A") with pytest.raises(ValueError): @@ -24,7 +40,6 @@ def test_message_field_duplicate_entry(): def test_message_from_id(): - MessageField._ALL_FIELDS_MAP.clear() MessageField("A", "A") message = MessageField.from_id("AA") diff --git a/test/remote/protocol/test_tag_val_menu_command_protocol.py b/test/remote/protocol/test_tag_val_menu_command_protocol.py index 240c33a..91bf40b 100644 --- a/test/remote/protocol/test_tag_val_menu_command_protocol.py +++ b/test/remote/protocol/test_tag_val_menu_command_protocol.py @@ -1,5 +1,13 @@ import io +import uuid +from tcmenu.remote.commands.dialog_mode import DialogMode +from tcmenu.remote.commands.menu_button_type import MenuButtonType +from tcmenu.remote.commands.menu_command_type import MenuCommandType +from tcmenu.remote.commands.menu_dialog_command import MenuDialogCommand +from tcmenu.remote.commands.menu_heartbeat_command import MenuHeartbeatCommand +from tcmenu.remote.commands.menu_join_command import MenuJoinCommand +from tcmenu.remote.protocol.api_platform import ApiPlatform from tcmenu.remote.protocol.command_protocol import CommandProtocol from tcmenu.remote.protocol.configurable_protocol_converter import ConfigurableProtocolConverter from tcmenu.remote.protocol.message_field import MessageField @@ -8,23 +16,75 @@ def test_receive_join_command(): - # command = protocol.from_channel( - # to_buffer( - # MenuCommandType.JOIN.message_field, - # "NM=IoTdevice|UU=07cd8bc6-734d-43da-84e7-6084990becfc|US=987654321|VE=1223|PF=1|\u0002", - # ) - # ) - - # assert type(command) is MenuJoinCommand - # # noinspection PyTypeChecker - # join: MenuJoinCommand = command - # assert "07cd8bc6-734d-43da-84e7-6084990becfc" == str(join.app_uuid) - # assert "IoTdevice" == join.my_name - # assert 1223 == join.api_version - # assert 987654321 == join - # assert ApiPlatform.JAVA_API == join.platform - # assert MenuCommandType.JOIN.message_field == join.command_type - pass + command = protocol.from_channel( + to_buffer( + MenuCommandType.JOIN.message_field, + "NM=IoTdevice|UU=07cd8bc6-734d-43da-84e7-6084990becfc|US=987654321|VE=1223|PF=1|\u0002", + ) + ) + + assert type(command) is MenuJoinCommand + # noinspection PyTypeChecker + join: MenuJoinCommand = command + assert "07cd8bc6-734d-43da-84e7-6084990becfc" == str(join.app_uuid) + assert "IoTdevice" == join.my_name + assert 1223 == join.api_version + assert 987654321 == join.serial_number + assert ApiPlatform.JAVA_API == join.platform + assert MenuCommandType.JOIN.message_field == join.command_type + + +def test_receive_join_command_with_no_uuid_and_sn(): + command = protocol.from_channel( + to_buffer( + MenuCommandType.JOIN.message_field, + "NM=IoTdevice|VE=1223|PF=1|\u0002", + ) + ) + + assert type(command) is MenuJoinCommand + # noinspection PyTypeChecker + join: MenuJoinCommand = command + assert isinstance(join.app_uuid, uuid.UUID) + assert "IoTdevice" == join.my_name + assert 1223 == join.api_version + assert 0 == join.serial_number + assert ApiPlatform.JAVA_API == join.platform + assert MenuCommandType.JOIN.message_field == join.command_type + + +def test_receive_dialog_command(): + command = protocol.from_channel( + to_buffer( + MenuCommandType.DIALOG_UPDATE.message_field, + "MO=S|HF=Hello\\||BU=Buffer\\=|B1=0|B2=4|\u0002", + ) + ) + + assert type(command) is MenuDialogCommand + # noinspection PyTypeChecker + dlg: MenuDialogCommand = command + assert DialogMode.SHOW == dlg.dialog_mode + assert "Hello|" == dlg.header + assert "Buffer=" == dlg.buffer + assert MenuButtonType.OK, dlg.button1 + assert MenuButtonType.NONE, dlg.button2 + + +def test_receive_heartbeat_command(): + command = protocol.from_channel( + to_buffer( + MenuCommandType.HEARTBEAT.message_field, + "\u0002", + ) + ) + + assert type(command) is MenuHeartbeatCommand + # noinspection PyTypeChecker + hb: MenuHeartbeatCommand = command + assert MenuCommandType.HEARTBEAT.message_field == hb.command_type + assert hb.mode == MenuHeartbeatCommand.HeartbeatMode.NORMAL + assert hb.heartbeat_interval == 10000 def to_buffer(message_type: MessageField, s: str) -> io.BytesIO: