Add s7 unified client/server tests, fix EXPLORE parser and server

32 tests for s7.Client and s7.Server using the built-in emulators:

- Legacy: connect, db_read, db_write, db_read_multi, list_datablocks,
  context manager, delegated methods, diagnostic buffer
- S7CommPlus: connect, db_read, db_write, db_read_multi, explore,
  list_datablocks, browse, browse-to-SymbolTable, auto-detection
- Server: context manager, register_db/raw_db, get_db, properties
- Guard tests: browse/explore/subscription require S7CommPlus

Bug fixes:
- s7/client.py: legacy connection optional when S7CommPlus explicit
- s7/_s7commplus_client.py: EXPLORE parser handles WSTRING (0x15),
  UDINT (0x04) datatypes, skips return code VLQ, tracks nesting depth
  to avoid child objects overwriting parent DB name/number
- s7/_s7commplus_server.py: EXPLORE response uses standard S7CommPlus
  IDs (Ids.OBJECT_VARIABLE_TYPE_NAME, Ids.BLOCK_BLOCK_NUMBER) with
  proper flags+datatype encoding

Coverage: s7/client.py 21%→88%, s7/_s7commplus_client.py 40%→67%,
total 78%→81%. mypy clean (0 errors across 79 files).

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

Author: gijzelaerr

s7/_s7commplus_client.py

@@ -592,31 +592,39 @@ def _parse_explore_datablocks(response: bytes) -> list[dict[str, Any]]:
     current_name = ""
     current_number = 0
     current_rid = 0
+    depth = 0
+
+    # Skip return code VLQ at start of response
+    if offset < len(response):
+        _, consumed = _vlq32(response, offset)
+        offset += consumed
 
     while offset < len(response):
-        if offset >= len(response):
-            break
 
         tag = response[offset]
         offset += 1
 
         if tag == 0xA1:  # START_OF_OBJECT
+            depth += 1
             if offset + 4 > len(response):
                 break
-            current_rid = struct.unpack(">I", response[offset : offset + 4])[0]
+            rid = struct.unpack(">I", response[offset : offset + 4])[0]
             offset += 4
             # Skip classId, reserved, reserved (3 VLQ values)
             for _ in range(3):
                 if offset >= len(response):
                     break
                 _, consumed = _vlq32(response, offset)
                 offset += consumed
-            current_name = ""
-            current_number = 0
+            if depth == 1:
+                current_rid = rid
+                current_name = ""
+                current_number = 0
 
         elif tag == 0xA2:  # TERMINATING_OBJECT
-            if current_name and current_number > 0:
+            if depth == 1 and current_name and current_number > 0:
                 datablocks.append({"name": current_name, "number": current_number, "rid": current_rid})
+            depth = max(0, depth - 1)
 
         elif tag == 0xA3:  # ATTRIBUTE
             if offset >= len(response):
@@ -629,24 +637,27 @@ def _parse_explore_datablocks(response: bytes) -> list[dict[str, Any]]:
             datatype = response[offset + 1]
             offset += 2
 
-            if attr_id == Ids.OBJECT_VARIABLE_TYPE_NAME and datatype == 0x13:  # WSTRING
+            if attr_id == Ids.OBJECT_VARIABLE_TYPE_NAME and datatype in (0x13, 0x15):  # S7STRING or WSTRING
                 if offset >= len(response):
                     break
                 str_len, consumed = _vlq32(response, offset)
                 offset += consumed
                 if offset + str_len <= len(response):
-                    try:
-                        current_name = response[offset : offset + str_len].decode("utf-16-be", errors="replace")
-                    except Exception:
-                        current_name = ""
+                    if depth == 1:
+                        try:
+                            current_name = response[offset : offset + str_len].decode("utf-16-be", errors="replace")
+                        except Exception:
+                            current_name = ""
                     offset += str_len
                     continue
 
-            if attr_id == Ids.BLOCK_BLOCK_NUMBER and datatype in (0x07, 0x08):  # UDINT/DWORD
+            if attr_id == Ids.BLOCK_BLOCK_NUMBER and datatype in (0x03, 0x04, 0x0C):  # UINT/UDINT/DWORD
                 if offset >= len(response):
                     break
-                current_number, consumed = _vlq32(response, offset)
+                val, consumed = _vlq32(response, offset)
                 offset += consumed
+                if depth == 1:
+                    current_number = val
                 continue
 
             # Skip unknown attribute value
@@ -680,11 +691,17 @@ def _parse_explore_fields(response: bytes, db_number: int, db_name: str) -> list
     """
     from .vlq import decode_uint32_vlq as _vlq32
 
+
     fields: list[dict[str, Any]] = []
     offset = 0
     field_name = ""
     byte_offset = 0
 
+    # Skip return code VLQ at start of response
+    if offset < len(response):
+        _, consumed = _vlq32(response, offset)
+        offset += consumed
+
     while offset < len(response):
         tag = response[offset]
         offset += 1

s7/_s7commplus_server.py

@@ -35,6 +35,7 @@
     DataType,
     ElementID,
     FunctionCode,
+    Ids,
     Opcode,
     ProtocolVersion,
     READ_FUNCTION_CODES,
@@ -716,34 +717,51 @@ def _handle_explore(self, seq_num: int, session_id: int, request_data: bytes) ->
         )
         response += encode_uint32_vlq(0)  # Return code: success
 
-        # Return list of data blocks as objects
+        # Return list of data blocks as objects using standard S7CommPlus IDs
         for db_num, db in sorted(self._data_blocks.items()):
             response += bytes([ElementID.START_OF_OBJECT])
             response += struct.pack(">I", db.object_id)  # Relation ID
             response += encode_uint32_vlq(0x00000100)  # Class: DataBlock
             response += encode_uint32_vlq(0x00000000)  # Class flags
             response += encode_uint32_vlq(0x00000000)  # Attribute ID
 
-            # DB number attribute
+            # ObjectVariableTypeName (233) -- DB name as WSTRING
             response += bytes([ElementID.ATTRIBUTE])
-            response += encode_uint32_vlq(0x0001)  # DB number attribute ID
-            response += encode_typed_value(DataType.UINT, db_num)
+            response += encode_uint32_vlq(Ids.OBJECT_VARIABLE_TYPE_NAME)
+            name_bytes = f"DB{db_num}".encode("utf-16-be")
+            response += bytes([0x00, DataType.WSTRING])
+            response += encode_uint32_vlq(len(name_bytes))
+            response += name_bytes
 
-            # DB size attribute
+            # Block_BlockNumber (2521) -- DB number as UDINT
             response += bytes([ElementID.ATTRIBUTE])
-            response += encode_uint32_vlq(0x0002)  # DB size attribute ID
-            response += encode_typed_value(DataType.UDINT, len(db.data))
+            response += encode_uint32_vlq(Ids.BLOCK_BLOCK_NUMBER)
+            response += bytes([0x00, DataType.UDINT])
+            response += encode_uint32_vlq(db_num)
 
-            # Variable list
+            # DB size attribute (non-standard, for backward compat)
+            response += bytes([ElementID.ATTRIBUTE])
+            response += encode_uint32_vlq(0x0002)
+            response += bytes([0x00, DataType.UDINT])
+            response += encode_uint32_vlq(len(db.data))
+
+            # Variable list -- used by browse to resolve field names
             if db.variables:
-                response += bytes([ElementID.VARNAME_LIST])
-                response += encode_uint32_vlq(len(db.variables))
                 for var_name, var in db.variables.items():
-                    name_bytes = var_name.encode("utf-8")
-                    response += encode_uint32_vlq(len(name_bytes))
-                    response += name_bytes
-                    response += encode_uint32_vlq(var.soft_datatype)
-                    response += encode_uint32_vlq(var.byte_offset)
+                    response += bytes([ElementID.START_OF_OBJECT])
+                    response += struct.pack(">I", 0)  # child RID
+                    response += encode_uint32_vlq(0)
+                    response += encode_uint32_vlq(0)
+                    response += encode_uint32_vlq(0)
+
+                    response += bytes([ElementID.ATTRIBUTE])
+                    response += encode_uint32_vlq(Ids.OBJECT_VARIABLE_TYPE_NAME)
+                    vname_bytes = var_name.encode("utf-16-be")
+                    response += bytes([0x00, DataType.WSTRING])
+                    response += encode_uint32_vlq(len(vname_bytes))
+                    response += vname_bytes
+
+                    response += bytes([ElementID.TERMINATING_OBJECT])
 
             response += bytes([ElementID.TERMINATING_OBJECT])
 

s7/client.py

@@ -127,10 +127,23 @@ def connect(
         else:
             self._protocol = Protocol.LEGACY
 
-        # Always connect legacy client (needed for block ops, PLC control, etc.)
-        self._legacy = LegacyClient()
-        self._legacy.connect(address, rack, slot, tcp_port)
-        logger.info(f"Legacy S7 connected to {address}:{tcp_port}")
+        # Connect legacy client for block ops, PLC control, etc.
+        # Skip when S7CommPlus was explicitly requested — the target may not
+        # support legacy S7 (e.g. PUT/GET disabled) or use a different port
+        # (e.g. test emulators).
+        if self._protocol != Protocol.S7COMMPLUS:
+            self._legacy = LegacyClient()
+            self._legacy.connect(address, rack, slot, tcp_port)
+            logger.info(f"Legacy S7 connected to {address}:{tcp_port}")
+        elif protocol == Protocol.AUTO:
+            # AUTO mode with S7CommPlus: also try legacy for block ops
+            try:
+                self._legacy = LegacyClient()
+                self._legacy.connect(address, rack, slot, tcp_port)
+                logger.info(f"Legacy S7 connected to {address}:{tcp_port}")
+            except Exception as e:
+                logger.debug(f"Legacy S7 connection failed (S7CommPlus available): {e}")
+                self._legacy = None
 
         return self
 

tests/test_logging.py

@@ -3,18 +3,20 @@
 import json
 import logging
 
+import pytest
+
 from snap7.log import PLCLoggerAdapter, OperationLogger, JSONFormatter
 
 
 class TestPLCLoggerAdapter:
-    def test_prefix_added(self, caplog: logging.LogRecord) -> None:  # type: ignore[type-arg]
+    def test_prefix_added(self, caplog: pytest.LogCaptureFixture) -> None:
         base = logging.getLogger("test.adapter")
         adapter = PLCLoggerAdapter(base, plc_host="10.0.0.1", rack=0, slot=1)
         with caplog.at_level(logging.INFO, logger="test.adapter"):
             adapter.info("Connected")
         assert "[10.0.0.1 R0/S1] Connected" in caplog.text
 
-    def test_no_prefix_without_host(self, caplog: logging.LogRecord) -> None:  # type: ignore[type-arg]
+    def test_no_prefix_without_host(self, caplog: pytest.LogCaptureFixture) -> None:
         base = logging.getLogger("test.nohost")
         adapter = PLCLoggerAdapter(base)
         with caplog.at_level(logging.INFO, logger="test.nohost"):
@@ -40,7 +42,7 @@ def test_update_context_partial(self) -> None:
 
 
 class TestOperationLogger:
-    def test_logs_timing(self, caplog: logging.LogRecord) -> None:  # type: ignore[type-arg]
+    def test_logs_timing(self, caplog: pytest.LogCaptureFixture) -> None:
         base = logging.getLogger("test.oplog")
         with caplog.at_level(logging.DEBUG, logger="test.oplog"):
             with OperationLogger(base, "db_read", db=1, start=0, size=4):
@@ -49,7 +51,7 @@ def test_logs_timing(self, caplog: logging.LogRecord) -> None:  # type: ignore[t
         assert "db=1" in caplog.text
         assert "ms)" in caplog.text
 
-    def test_works_with_adapter(self, caplog: logging.LogRecord) -> None:  # type: ignore[type-arg]
+    def test_works_with_adapter(self, caplog: pytest.LogCaptureFixture) -> None:
         base = logging.getLogger("test.oplog_adapter")
         adapter = PLCLoggerAdapter(base, plc_host="10.0.0.1", rack=0, slot=1)
         with caplog.at_level(logging.DEBUG, logger="test.oplog_adapter"):
@@ -90,9 +92,9 @@ def test_plc_context_included(self) -> None:
             args=None,
             exc_info=None,
         )
-        record.plc_host = "192.168.1.10"  # type: ignore[attr-defined]
-        record.plc_rack = 0  # type: ignore[attr-defined]
-        record.plc_slot = 1  # type: ignore[attr-defined]
+        record.plc_host = "192.168.1.10"
+        record.plc_rack = 0
+        record.plc_slot = 1
         output = formatter.format(record)
         data = json.loads(output)
         assert data["plc_host"] == "192.168.1.10"