Add s7 unified client/server tests, fix mypy issues

29 new tests for s7.Client and s7.Server using the built-in server
emulator (no real PLC needed):

- Legacy protocol: connect, db_read, db_write, db_read_multi,
  list_datablocks, context manager, repr, delegated methods,
  diagnostic buffer, attribute errors
- S7CommPlus protocol: connect, db_read, db_write, db_read_multi,
  explore, auto-detection
- S7CommPlus guards: browse/explore/subscription require S7CommPlus
- Server: context manager, register_db, register_raw_db, get_db,
  property accessors
- Protocol enum values

Bug fix in s7/client.py: legacy connection is now optional when
S7CommPlus is explicitly requested — PLCs with PUT/GET disabled and
test emulators don't support both protocols on the same port.

Coverage improvements:
- s7/client.py: 21% -> 87%
- s7/server.py: 43% -> 84%

Also fixed mypy issues in test_logging.py.

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

Author: gijzelaerr

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"