fix(s7commplus): remove misleading TLS hint

s7commplus/async_client.py

@@ -40,6 +40,7 @@
     _build_get_var_substreamed_payload,
     _build_set_variable_payload,
     _check_set_variable_response,
+    _log_create_object_return_value,
     _parse_get_var_substreamed_response,
     _parse_protection_level_response,
     _set_s7_groups,
@@ -939,8 +940,7 @@ async def _create_session(self) -> None:
             self._session_id = struct.unpack_from(">I", response, 9)[0]
         self._protocol_version = version
 
-        if return_value != 0:
-            logger.warning(f"CreateObject returned error 0x{return_value:X} — PLC may require TLS (use_tls=True)")
+        _log_create_object_return_value(return_value, self._tls_active)
 
         self._server_session_version = parse_server_session_version(response[10 + obj_end :])
         if self._server_session_version is not None:

s7commplus/connection.py

@@ -68,6 +68,25 @@
 
 logger = logging.getLogger(__name__)
 
+
+def _log_create_object_return_value(return_value: int, tls_active: bool) -> None:
+    """Log a non-zero CreateObject status without guessing at TLS requirements."""
+    if return_value == 0:
+        return
+    if tls_active:
+        # Some firmware (e.g. S7-1200 FW V4.1) returns a non-zero value on a
+        # usable TLS session, so keep this informational.
+        logger.debug(
+            "CreateObject returned non-zero 0x%X on an active TLS session; continuing to parse the returned session data",
+            return_value,
+        )
+        return
+    logger.warning(
+        "CreateObject returned non-zero 0x%X; continuing to parse the returned session data",
+        return_value,
+    )
+
+
 # TLS cipher suites for S7 PLC compatibility.
 # ECDHE suites are preferred (forward secrecy); RSA-kx kept as fallback for
 # older firmware.  The key to Siemens PLC compatibility is restricting the
@@ -1093,13 +1112,7 @@ def _wstring_attr(attr_id: int, s: str) -> bytes:
         logger.debug(f"CreateObject response: return_value={return_value} object_ids={[hex(i) for i in object_ids]}")
         logger.debug(f"Session created: id=0x{self._session_id:08X} ({self._session_id}), version=V{version}")
 
-        if return_value != 0:
-            if self._tls_active:
-                # Some firmware (e.g. S7-1200 FW V4.1) returns a non-zero CreateObject
-                # value on a perfectly usable TLS session, so this is informational only.
-                logger.debug(f"CreateObject returned non-zero 0x{return_value:X} on an active TLS session (session still usable)")
-            else:
-                logger.warning(f"CreateObject returned error 0x{return_value:X} — PLC may require TLS (use_tls=True)")
+        _log_create_object_return_value(return_value, self._tls_active)
 
         # Parse remaining payload (the ResponseObject tree) for session attributes
         attrs = parse_create_object_attributes(response[offset:])

tests/test_s7_v2.py

@@ -5,6 +5,7 @@
 """
 
 import hashlib
+import logging
 import struct
 from unittest.mock import AsyncMock, MagicMock
 
@@ -17,6 +18,7 @@
     _build_get_var_substreamed_payload,
     _build_set_variable_payload,
     _check_set_variable_response,
+    _log_create_object_return_value,
     _parse_get_var_substreamed_response,
     _parse_protection_level_response,
 )
@@ -534,3 +536,20 @@ def test_authenticate_requires_tls(self) -> None:
         conn._tls_active = False
         with pytest.raises(S7ConnectionError, match="requires TLS"):
             conn.authenticate("password")
+
+
+class TestCreateObjectStatusLogging:
+    """A CreateObject status alone does not identify a TLS requirement."""
+
+    def test_plain_connection_does_not_recommend_tls(self, caplog: pytest.LogCaptureFixture) -> None:
+        with caplog.at_level(logging.WARNING, logger="s7commplus.connection"):
+            _log_create_object_return_value(0x4000800000000011, tls_active=False)
+
+        assert "continuing to parse the returned session data" in caplog.text
+        assert "TLS" not in caplog.text
+
+    def test_success_is_silent(self, caplog: pytest.LogCaptureFixture) -> None:
+        with caplog.at_level(logging.DEBUG, logger="s7commplus.connection"):
+            _log_create_object_return_value(0, tls_active=False)
+
+        assert caplog.text == ""