Refractor of messages and added tests

iec62056_21/client.py

@@ -201,9 +201,7 @@ def send_break(self):
         communication.
         """
         logger.info("Sending BREAK message to end communication")
-        break_msg = messages.CommandMessage(
-            command="B", command_type="0", data_set=None
-        )
+        break_msg = messages.CommandMessage(command="B", command_type=0, data_set=None)
         self.transport.send(break_msg.to_bytes())
 
     def ack_with_option_select(self, mode):

iec62056_21/messages.py

@@ -50,7 +50,7 @@ class DataSet(Iec6205621Data):
 
     EXCLUDE_CHARS = ["(", ")", "/", "!"]
 
-    def __init__(self, value, address=None, unit=None, no_address=False):
+    def __init__(self, address, value, unit=None, no_address=False):
 
         # TODO: in programming mode, protocol mode C the value can be up to 128 chars
 
@@ -175,18 +175,8 @@ def from_representation(cls, string_data: str):
         if not utils.bcc_valid(string_data):
             raise ValueError("BCC not valid")
 
-        _in_data = _in_data[:-1]  # remove bcc
+        _in_data = _in_data[1:-5]  # remove stx and !<cr><lf>ETX bcc
 
-        if _in_data.startswith(constants.STX):
-            _in_data = _in_data[1:]
-        else:
-            raise ValueError("ReadoutDataMessage does not start with STX")
-        if _in_data.endswith(
-            f"{constants.END_CHAR}{constants.LINE_END}{constants.ETX}"
-        ):
-            _in_data = _in_data[:-3]
-        else:
-            raise ValueError("ReadoutDataMessage does not end with !<cr><lf>ETX")
         data_block = DataBlock.from_representation(_in_data)
 
         return cls(data_block=data_block)
@@ -204,6 +194,11 @@ def __init__(self, command, command_type, data_set):
         self.command_type = command_type
         self.data_set = data_set
 
+        if command not in self.allowed_commands:
+            raise ValueError(f"{command} is not an allowed command")
+        if command_type not in self.allowed_command_types:
+            raise ValueError(f"{command_type} is not an allowed command type")
+
     def to_representation(self):
         header = f"{constants.SOH}{self.command}{self.command_type}"
         if self.data_set:
@@ -227,11 +222,6 @@ def from_representation(cls, string_data):
         command_type = int(header[2])
         data_set = DataSet.from_representation(body[1:-1])
 
-        if command not in cls.allowed_commands:
-            raise ValueError(f"{command} is not an allowed command")
-        if command_type not in cls.allowed_command_types:
-            raise ValueError(f"{command_type} is not an allowed command type")
-
         return cls(command, command_type, data_set)
 
     @classmethod

tests/test_data.py

@@ -150,18 +150,6 @@ def test_to_representation_single_line(self):
             assert db.to_representation() == "3:14(314*kWh)4:15(415*kWh)\r\n"
 
 
-class TestCommandMessage:
-    def test_parse_p0_messge(self):
-        data = b"\x01P0\x02(1234567)\x03P"
-        cm = messages.CommandMessage.from_bytes(data)
-
-        assert cm.command == "P"
-        assert cm.command_type == 0
-        assert cm.data_set.value == "1234567"
-        assert cm.data_set.address is None
-        assert cm.data_set.unit is None
-
-
 class TestAnswerDataMessage:
     def test_parse_read_response(self):
         data = b"\x023:171.0(0)\x03\x12"
@@ -190,3 +178,55 @@ def test_to_representation(self):
         )
 
         assert rdm.to_representation() == '\x023:14(314*kWh)4:15(415*kWh)\r\n!\r\n\x03"'
+
+    def test_from_representation(self):
+
+        rdm = messages.ReadoutDataMessage.from_representation(
+            '\x023:14(314*kWh)4:15(415*kWh)\r\n!\r\n\x03"'
+        )
+        print(rdm)
+        assert len(rdm.data_block.data_lines) == 1
+        assert len(rdm.data_block.data_lines[0].data_sets) == 2
+
+    def test_invalid_bcc_raises_error(self):
+        with pytest.raises(ValueError):
+            rdm = messages.ReadoutDataMessage.from_representation(
+                "\x023:14(314*kWh)4:15(415*kWh)\r\n!\r\n\x03x"
+            )
+
+
+class TestCommandMessage:
+    def test_command_message_to_representation(self):
+        cm = messages.CommandMessage(
+            command="R",
+            command_type=1,
+            data_set=messages.DataSet(address="1.8.0", value=1),
+        )
+
+        assert cm.to_representation() == "\x01R1\x021.8.0(1)\x03k"
+
+    def test_from_representation(self):
+        data = "\x01P0\x02(1234567)\x03P"
+        cm = messages.CommandMessage.from_representation(data)
+
+        assert cm.command == "P"
+        assert cm.command_type == 0
+        assert cm.data_set.value == "1234567"
+        assert cm.data_set.address is None
+        assert cm.data_set.unit is None
+
+    def test_invalid_command_raises_value_error(self):
+        with pytest.raises(ValueError):
+            cm = messages.CommandMessage(
+                command="X",
+                command_type=1,
+                data_set=messages.DataSet(address="1.8.0", value=1),
+            )
+
+    def test_invalid_command_type_raises_value_error(self):
+        with pytest.raises(ValueError):
+            cm = messages.CommandMessage(
+                command="R",
+                command_type=12,
+                data_set=messages.DataSet(address="1.8.0", value=1),
+            )