feat: some cleanup and initial work towards tests (#30)

* feat: some cleanup and initial work towards tests

* feat: run pytest as part of CI

.github/workflows/ci.yml

@@ -40,8 +40,8 @@ jobs:
           - "3.11"
         os:
           - ubuntu-latest
-          # - windows-latest
-          # - macOS-latest
+          - windows-latest
+          - macOS-latest
     runs-on: ${{ matrix.os }}
     steps:
       - uses: actions/checkout@v3
@@ -53,9 +53,9 @@ jobs:
       - name: Install Dependencies
         run: poetry install
         shell: bash
-      # - name: Test with Pytest
-      #   run: poetry run pytest
-      #   shell: bash
+      - name: Test with Pytest
+        run: poetry run pytest
+        shell: bash
   release:
     runs-on: ubuntu-latest
     environment: release

.vscode/settings.json

@@ -3,7 +3,12 @@
   "editor.rulers": [140],
   "python.formatting.provider": "black",
   // https://code.visualstudio.com/docs/python/testing#_pytest-configuration-settings
-  "python.testing.pytestEnabled": false,
+  "python.testing.pytestEnabled": true,
+  "python.testing.pytestArgs": [
+    "--cov=pyomnilogic_local",
+    "--cov-report=xml:coverage.xml",
+    "tests"
+  ],
   "python.linting.mypyEnabled": true,
   "python.analysis.typeCheckingMode": "basic"
 }

pyomnilogic_local/models/mspconfig.py

@@ -54,10 +54,9 @@ def propagate_bow_id(self, bow_id: int | None) -> None:
                     # ... then call propagate_bow_id on each of them ...
                     if device is not None:
                         device.propagate_bow_id(bow_id)
-            else:
-                # ... otherwise just call it on the single subdevice
-                if subdevice is not None:
-                    subdevice.propagate_bow_id(bow_id)
+            # ... otherwise just call it on the single subdevice
+            elif subdevice is not None:
+                subdevice.propagate_bow_id(bow_id)
 
 
 class MSPSystem(BaseModel):

pyomnilogic_local/protocol.py

@@ -23,9 +23,9 @@ class OmniLogicMessage:
     payload: bytes
     client_type: ClientType = ClientType.SIMPLE
     version: str = "1.19"
-    timestamp: int | None
+    timestamp: int | None = int(time.time())
     reserved_1: int = 0
-    compressed: int = 0
+    compressed: bool = False
     reserved_2: int = 0
 
     def __init__(self, msg_id: int, msg_type: MessageType, payload: str | None = None, version: str = "1.19") -> None:
@@ -43,31 +43,34 @@ def __bytes__(self) -> bytes:
         header = struct.pack(
             self.header_format,
             self.id,  # Msg id
-            int(time.time_ns() / (10**9)),  # Timestamp
+            self.timestamp,
             bytes(self.version, "ascii"),  # version string
             self.type.value,  # OpID/msgType
             self.client_type.value,  # Client type
             0,  # reserved
-            0,  # compressed
+            self.compressed,  # compressed
             0,  # reserved
         )
         return header + self.payload
 
     def __repr__(self) -> str:
         if self.compressed or self.type is MessageType.MSP_BLOCKMESSAGE:
-            return f"ID: {self.id}, Type: {self.type.name}, Compressed: {self.compressed}"
-        return f"ID: {self.id}, Type: {self.type.name}, Compressed: {self.compressed}, Body: {self.payload[:-1].decode('utf-8')}"
+            return f"ID: {self.id}, Type: {self.type.name}, Compressed: {self.compressed}, Client: {self.client_type.name}"
+        return (
+            f"ID: {self.id}, Type: {self.type.name}, Compressed: {self.compressed}, Client: {self.client_type.name}, "
+            f"Body: {self.payload[:-1].decode('utf-8')}"
+        )
 
     @classmethod
     def from_bytes(cls, data: bytes) -> Self:
         # split the header and data
-        header = data[0:24]
+        header = data[:24]
         rdata: bytes = data[24:]
 
         msg_id, tstamp, vers, msg_type, client_type, res1, compressed, res2 = struct.unpack(cls.header_format, header)
-        message = cls(msg_id=msg_id, msg_type=MessageType(msg_type), version=vers)
+        message = cls(msg_id=msg_id, msg_type=MessageType(msg_type), version=vers.decode("utf-8"))
         message.timestamp = tstamp
-        message.client_type = client_type
+        message.client_type = ClientType(int(client_type))
         message.reserved_1 = res1
         # There are some messages that are ALWAYS compressed although they do not return a 1 in their LeadMessage
         message.compressed = compressed == 1 or message.type in [MessageType.MSP_TELEMETRY_UPDATE]
@@ -176,7 +179,7 @@ async def _receive_file(self) -> str:
         await self._send_ack(message.id)
 
         # If the response is too large, the controller will send a LeadMessage indicating how many follow-up messages will be sent
-        if message.type == MessageType.MSP_LEADMESSAGE:
+        if message.type is MessageType.MSP_LEADMESSAGE:
             leadmsg = LeadMessage.from_orm(ET.fromstring(message.payload[:-1]))
 
             _LOGGER.debug("Will receive %s blockmessages", leadmsg.msg_block_count)
@@ -207,10 +210,7 @@ async def _receive_file(self) -> str:
             for _, data in sorted(data_fragments.items()):
                 retval += data
 
-        # If we did not receive a LeadMessage, but the message is compressed anyway...
-        elif message.compressed:
-            retval = message.payload
-        # A short response, no LeadMessage and no compression...
+        # We did not receive a LeadMessage, so our payload is just this one packet
         else:
             retval = message.payload
 

pyomnilogic_local/types.py

@@ -32,6 +32,7 @@ class MessageType(Enum):
 class ClientType(Enum):
     XML = 0
     SIMPLE = 1
+    OMNI = 3
 
 
 class OmniType(str, Enum):

pyproject.toml

@@ -21,7 +21,6 @@ omnilogic = "pyomnilogic_local.cli:main"
 [tool.poetry.dependencies]
 python = "^3.10"
 pydantic = "^1.10.7"
-pydantic-xml = "^0.6.1"
 xmltodict = "^0.13.0"
 
 [tool.poetry.group.dev.dependencies]
@@ -30,6 +29,12 @@ mypy = "^1.2.0"
 pylint = "^2.17.0"
 pydantic = "^1.10.7"
 pytest = "^7.3.1"
+pytest-cov = "^4.1.0"
+
+[tool.pytest.ini_options]
+addopts = [
+    "--import-mode=importlib",
+]
 
 [build-system]
 requires = ["poetry-core"]

tests/test_protocol.py

@@ -0,0 +1,61 @@
+from pyomnilogic_local.protocol import OmniLogicMessage
+from pyomnilogic_local.types import ClientType, MessageType
+
+
+def test_parse_basic_ack() -> None:
+    """Validate that we can parse a basic ACK packet"""
+    bytes_ack = b"\x99_\xd1l\x00\x00\x00\x00dv\x8f\xc11.20\x00\x00\x03\xea\x03\x00\x00\x00"
+    message = OmniLogicMessage.from_bytes(bytes_ack)
+    assert message.id == 2573193580
+    assert message.type is MessageType.ACK
+    assert message.compressed is False
+    assert str(message) == "ID: 2573193580, Type: ACK, Compressed: False, Client: OMNI, Body: "
+
+
+def test_create_basic_ack() -> None:
+    """Validate that we can create a valid basic ACK packet"""
+    bytes_ack = b"\x99_\xd1l\x00\x00\x00\x00dv\x8f\xc11.20\x00\x00\x03\xea\x03\x00\x00\x00"
+    message = OmniLogicMessage(2573193580, MessageType.ACK, payload=None, version="1.20")
+    message.client_type = ClientType.OMNI
+    message.timestamp = 1685491649
+    assert bytes(message) == bytes_ack
+
+
+def test_parse_leadmessate() -> None:
+    """Validate that we can parse an MSP LeadMessage."""
+    bytes_leadmessage = (
+        b'\x00\x00\x90v\x00\x00\x00\x00dv\x92\xc11.20\x00\x00\x07\xce\x03\x00\x01\x00<?xml version="1.0" encoding="UTF-8" ?>'
+        b'<Response xmlns="http://nextgen.hayward.com/api"><Name>LeadMessage</Name><Parameters>'
+        b'<Parameter name="SourceOpId" dataType="int">1003</Parameter><Parameter name="MsgSize" dataType="int">3361</Parameter>'
+        b'<Parameter name="MsgBlockCount" dataType="int">4</Parameter><Parameter name="Type" dataType="int">0</Parameter>'
+        b"</Parameters></Response>\x00"
+    )
+    message = OmniLogicMessage.from_bytes(bytes_leadmessage)
+    print(message.timestamp)
+    assert message.id == 36982
+    assert message.type is MessageType.MSP_LEADMESSAGE
+    assert message.timestamp == 1685492417
+    assert message.compressed is True
+    assert str(message) == "ID: 36982, Type: MSP_LEADMESSAGE, Compressed: True, Client: OMNI"
+
+
+def test_create_leadmessage() -> None:
+    """Validate that we can create a valid MSP LeadMessage"""
+    bytes_leadmessage = (
+        b'\x00\x00\x90v\x00\x00\x00\x00dv\x92\xc11.20\x00\x00\x07\xce\x03\x00\x01\x00<?xml version="1.0" encoding="UTF-8" ?>'
+        b'<Response xmlns="http://nextgen.hayward.com/api"><Name>LeadMessage</Name><Parameters>'
+        b'<Parameter name="SourceOpId" dataType="int">1003</Parameter><Parameter name="MsgSize" dataType="int">3361</Parameter>'
+        b'<Parameter name="MsgBlockCount" dataType="int">4</Parameter><Parameter name="Type" dataType="int">0</Parameter></Parameters>'
+        b"</Response>\x00"
+    )
+    payload_leadmessage = (
+        '<?xml version="1.0" encoding="UTF-8" ?><Response xmlns="http://nextgen.hayward.com/api"><Name>LeadMessage</Name><Parameters>'
+        '<Parameter name="SourceOpId" dataType="int">1003</Parameter><Parameter name="MsgSize" dataType="int">3361</Parameter>'
+        '<Parameter name="MsgBlockCount" dataType="int">4</Parameter><Parameter name="Type" dataType="int">0</Parameter></Parameters>'
+        "</Response>"
+    )
+    message = OmniLogicMessage(36982, MessageType.MSP_LEADMESSAGE, payload=payload_leadmessage, version="1.20")
+    message.client_type = ClientType.OMNI
+    message.timestamp = 1685492417
+    message.compressed = True
+    assert bytes(message) == bytes_leadmessage