From 03f7d0b529234279b524449a53c7245e40160fd0 Mon Sep 17 00:00:00 2001 From: Gijs Molenaar Date: Tue, 18 Aug 2026 14:51:21 +0200 Subject: [PATCH 1/3] feat(s7commplus): add multi-write APIs --- s7commplus/async_client.py | 10 +++++++++- s7commplus/client.py | 14 +++++++++++++- tests/test_s7_server.py | 33 +++++++++++++++++++++++++++++++++ tests/test_s7_unit.py | 10 ++++++++++ 4 files changed, 65 insertions(+), 2 deletions(-) diff --git a/s7commplus/async_client.py b/s7commplus/async_client.py index f632365b..06f8e7c6 100644 --- a/s7commplus/async_client.py +++ b/s7commplus/async_client.py @@ -451,10 +451,18 @@ async def db_read(self, db_number: int, start: int, size: int) -> bytes: async def db_write(self, db_number: int, start: int, data: bytes) -> None: """Write raw bytes to a data block.""" - payload = _build_write_payload([(db_number, start, data)]) + await self.db_write_multi([(db_number, start, data)]) + + async def db_write_multi(self, items: list[tuple[int, int, bytes]]) -> None: + """Write multiple data block regions in a single request.""" + payload = _build_write_payload(items) response = await self._send_request(FunctionCode.SET_MULTI_VARIABLES, payload) _parse_write_response(response) + async def write_multi(self, items: list[tuple[int, int, bytes]]) -> None: + """Alias for :meth:`db_write_multi`.""" + await self.db_write_multi(items) + async def db_read_multi(self, items: list[tuple[int, int, int]]) -> list[bytes]: """Read multiple data block regions in a single request.""" payload = _build_read_payload(items) diff --git a/s7commplus/client.py b/s7commplus/client.py index 7bef905a..2836b382 100644 --- a/s7commplus/client.py +++ b/s7commplus/client.py @@ -137,13 +137,25 @@ def db_write(self, db_number: int, start: int, data: bytes) -> None: start: Start byte offset data: Bytes to write """ + self.db_write_multi([(db_number, start, data)]) + + def db_write_multi(self, items: list[tuple[int, int, bytes]]) -> None: + """Write multiple data block regions in a single request. + + Args: + items: List of ``(db_number, start_offset, data)`` tuples. + """ if self._connection is None: raise RuntimeError("Not connected") - payload = _build_write_payload([(db_number, start, data)]) + payload = _build_write_payload(items) response = self._connection.send_request(FunctionCode.SET_MULTI_VARIABLES, payload) _parse_write_response(response) + def write_multi(self, items: list[tuple[int, int, bytes]]) -> None: + """Alias for :meth:`db_write_multi`.""" + self.db_write_multi(items) + def db_read_multi(self, items: list[tuple[int, int, int]]) -> list[bytes]: """Read multiple data block regions in a single request. diff --git a/tests/test_s7_server.py b/tests/test_s7_server.py index 1ba92bee..3073954f 100644 --- a/tests/test_s7_server.py +++ b/tests/test_s7_server.py @@ -207,6 +207,24 @@ def test_multi_read(self, server: S7CommPlusServer) -> None: finally: client.disconnect() + def test_multi_write(self, server: S7CommPlusServer) -> None: + client = S7CommPlusClient() + client.connect("127.0.0.1", port=TEST_PORT) + try: + client.db_write_multi( + [ + (1, 0, b"first"), + (1, 10, b"second"), + (2, 20, b"third"), + ] + ) + + assert client.db_read(1, 0, 5) == b"first" + assert client.db_read(1, 10, 6) == b"second" + assert client.db_read(2, 20, 5) == b"third" + finally: + client.disconnect() + def test_explore(self, server: S7CommPlusServer) -> None: client = S7CommPlusClient() client.connect("127.0.0.1", port=TEST_PORT) @@ -304,6 +322,21 @@ async def test_multi_read(self, server: S7CommPlusServer) -> None: temp = struct.unpack(">f", results[0])[0] assert abs(temp - 23.5) < 0.1 # May be modified by earlier test + async def test_multi_write(self, server: S7CommPlusServer) -> None: + async with S7CommPlusAsyncClient() as client: + await client.connect("127.0.0.1", port=TEST_PORT) + await client.write_multi( + [ + (1, 0, b"alpha"), + (1, 10, b"beta"), + (2, 20, b"gamma"), + ] + ) + + assert await client.db_read(1, 0, 5) == b"alpha" + assert await client.db_read(1, 10, 4) == b"beta" + assert await client.db_read(2, 20, 5) == b"gamma" + async def test_explore(self, server: S7CommPlusServer) -> None: async with S7CommPlusAsyncClient() as client: await client.connect("127.0.0.1", port=TEST_PORT) diff --git a/tests/test_s7_unit.py b/tests/test_s7_unit.py index 1c03ffc8..57c03862 100644 --- a/tests/test_s7_unit.py +++ b/tests/test_s7_unit.py @@ -501,6 +501,16 @@ def test_db_read_multi_not_connected(self) -> None: with pytest.raises(RuntimeError, match="Not connected"): client.db_read_multi([(1, 0, 4)]) + def test_db_write_multi_not_connected(self) -> None: + client = S7CommPlusClient() + with pytest.raises(RuntimeError, match="Not connected"): + client.db_write_multi([(1, 0, b"data")]) + + def test_write_multi_not_connected(self) -> None: + client = S7CommPlusClient() + with pytest.raises(RuntimeError, match="Not connected"): + client.write_multi([(1, 0, b"data")]) + def test_explore_not_connected(self) -> None: client = S7CommPlusClient() with pytest.raises(RuntimeError, match="Not connected"): From 6ad0fa5c44c0831b8f620b5684c6bfba2c3b934c Mon Sep 17 00:00:00 2001 From: Gijs Molenaar Date: Mon, 7 Sep 2026 09:48:35 +0200 Subject: [PATCH 2/3] fix(s7commplus): encode typed multi-writes Co-Authored-By: Codex --- s7commplus/async_client.py | 18 +++++------ s7commplus/client.py | 64 +++++++++++++++++++++----------------- s7commplus/codec.py | 36 +++++++++++++++++++++ tests/test_s7_codec.py | 10 ++++++ tests/test_s7_unit.py | 37 ++++++++++++---------- 5 files changed, 110 insertions(+), 55 deletions(-) diff --git a/s7commplus/async_client.py b/s7commplus/async_client.py index 7ff53c1a..9197e1a0 100644 --- a/s7commplus/async_client.py +++ b/s7commplus/async_client.py @@ -12,6 +12,7 @@ from . import typeinfo from .blob_decompressor import find_and_decompress from .client import ( + DBWriteItem, _build_area_read_payload, _build_area_write_payload, _build_explore_payload, @@ -434,17 +435,17 @@ async def db_read(self, db_number: int, start: int, size: int) -> bytes: raise RuntimeError("Read failed: PLC returned error for item") return results[0] - async def db_write(self, db_number: int, start: int, data: bytes) -> None: - """Write raw bytes to a data block.""" - await self.db_write_multi([(db_number, start, data)]) + async def db_write(self, db_number: int, start: int, data: bytes, datatype: DataType = DataType.BLOB) -> None: + """Write raw bytes to a data block with an optional explicit PValue datatype.""" + await self.db_write_multi([(db_number, start, data, datatype)]) - async def db_write_multi(self, items: list[tuple[int, int, bytes]]) -> None: - """Write multiple data block regions in a single request.""" + async def db_write_multi(self, items: list[DBWriteItem]) -> None: + """Write multiple regions, optionally adding a DataType as each tuple's fourth item.""" payload = _build_write_payload(items, self._protocol_version) response = await self._send_request(FunctionCode.SET_MULTI_VARIABLES, payload) _parse_write_response(response) - async def write_multi(self, items: list[tuple[int, int, bytes]]) -> None: + async def write_multi(self, items: list[DBWriteItem]) -> None: """Alias for :meth:`db_write_multi`.""" await self.db_write_multi(items) @@ -595,10 +596,7 @@ async def read_symbolic(self, access_area: int, lids: list[int], symbol_crc: int .. warning:: This method is **experimental** and may change. """ - # TODO: Send the correct integrity id once available - payload = _build_symbolic_read_payload( - access_area, lids, symbol_crc, False, self._integrity_id_read, self._protocol_version - ) + payload = _build_symbolic_read_payload(access_area, lids, symbol_crc, self._protocol_version) response = await self._send_request(FunctionCode.GET_MULTI_VARIABLES, payload) results = _parse_read_response(response) if not results or results[0] is None: diff --git a/s7commplus/client.py b/s7commplus/client.py index 6bc9050e..807774d7 100644 --- a/s7commplus/client.py +++ b/s7commplus/client.py @@ -5,7 +5,7 @@ import logging import struct -from typing import Any, Callable, Optional, TypeVar +from typing import Any, Callable, Optional, TypeAlias, TypeVar from snap7.error import S7ConnectionError @@ -16,6 +16,7 @@ encode_item_address, encode_object_qualifier, encode_pvalue_blob, + encode_pvalue_typed, ) from .connection import S7CommPlusConnection from .protocol import DataType, ElementID, FunctionCode, Ids, ObjectId, ProtocolVersion @@ -24,6 +25,15 @@ logger = logging.getLogger(__name__) _T = TypeVar("_T") +DBWriteItem: TypeAlias = tuple[int, int, bytes] | tuple[int, int, bytes, DataType] + + +def _normalize_write_item(item: DBWriteItem) -> tuple[int, int, bytes, DataType]: + if len(item) == 3: + db_number, start, data = item + return db_number, start, data, DataType.BLOB + db_number, start, data, datatype = item + return db_number, start, data, DataType(datatype) class S7CommPlusClient: @@ -199,43 +209,46 @@ def _db_read_substreamed(self, db_number: int, start: int, size: int) -> bytes: response = self._connection.send_request(FunctionCode.GET_VAR_SUBSTREAMED, payload) return _parse_substreamed_read_response(response) - def db_write(self, db_number: int, start: int, data: bytes) -> None: + def db_write(self, db_number: int, start: int, data: bytes, datatype: DataType = DataType.BLOB) -> None: """Write raw bytes to a data block. Args: db_number: Data block number start: Start byte offset data: Bytes to write + datatype: S7CommPlus PValue datatype for ``data``. Defaults to BLOB. """ - self.db_write_multi([(db_number, start, data)]) + self.db_write_multi([(db_number, start, data, datatype)]) - def db_write_multi(self, items: list[tuple[int, int, bytes]]) -> None: + def db_write_multi(self, items: list[DBWriteItem]) -> None: """Write multiple data block regions in a single request. Args: - items: List of ``(db_number, start_offset, data)`` tuples. + items: ``(db_number, start_offset, data)`` tuples for BLOB writes, + or four-tuples adding an explicit :class:`DataType`. """ if self._connection is None: raise RuntimeError("Not connected") if self._connection.requires_substreamed: - for db_number, start, data in items: - self._db_write_substreamed(db_number, start, data) + for item in items: + db_number, start, data, datatype = _normalize_write_item(item) + self._db_write_substreamed(db_number, start, data, datatype) return payload = _build_write_payload(items, self._connection.protocol_version) response = self._connection.send_request(FunctionCode.SET_MULTI_VARIABLES, payload) _parse_write_response(response) - def write_multi(self, items: list[tuple[int, int, bytes]]) -> None: + def write_multi(self, items: list[DBWriteItem]) -> None: """Alias for :meth:`db_write_multi`.""" self.db_write_multi(items) - def _db_write_substreamed(self, db_number: int, start: int, data: bytes) -> None: + def _db_write_substreamed(self, db_number: int, start: int, data: bytes, datatype: DataType = DataType.BLOB) -> None: assert self._connection is not None access_area = Ids.DB_ACCESS_AREA_BASE + (db_number & 0xFFFF) payload = _build_substreamed_write_payload( - self._connection.session_id, access_area, Ids.DB_VALUE_ACTUAL, [start + 1, len(data)], data + self._connection.session_id, access_area, Ids.DB_VALUE_ACTUAL, [start + 1, len(data)], data, datatype ) self._connection.send_request(FunctionCode.SET_VAR_SUBSTREAMED, payload) @@ -348,9 +361,7 @@ def read_symbolic(self, access_area: int, lids: list[int], symbol_crc: int = 0) raise RuntimeError("Not connected") # TODO: Send the correct integrity id once available - payload = _build_symbolic_read_payload( - access_area, lids, symbol_crc, False, protocol_version=self._connection.protocol_version - ) + payload = _build_symbolic_read_payload(access_area, lids, symbol_crc, self._connection.protocol_version) response = self._connection.send_request(FunctionCode.GET_MULTI_VARIABLES, payload) results = _parse_read_response(response) if not results or results[0] is None: @@ -710,7 +721,6 @@ def _build_read_payload(items: list[tuple[int, int, int]], protocol_version: int for addr in addresses: payload += addr payload += encode_object_qualifier(protocol_version=protocol_version) - payload += encode_uint32_vlq(1) payload += struct.pack(">I", 0) return bytes(payload) @@ -771,7 +781,7 @@ def _parse_read_response(response: bytes) -> list[Optional[bytes]]: return results -def _build_write_payload(items: list[tuple[int, int, bytes]], protocol_version: int = ProtocolVersion.V2) -> bytes: +def _build_write_payload(items: list[DBWriteItem], protocol_version: int = ProtocolVersion.V2) -> bytes: """Build a SetMultiVariables request payload. Args: @@ -782,7 +792,8 @@ def _build_write_payload(items: list[tuple[int, int, bytes]], protocol_version: """ addresses: list[bytes] = [] total_field_count = 0 - for db_number, start, data in items: + normalized = [_normalize_write_item(item) for item in items] + for db_number, start, data, _ in normalized: access_area = Ids.DB_ACCESS_AREA_BASE + (db_number & 0xFFFF) addr_bytes, field_count = encode_item_address( access_area=access_area, @@ -798,12 +809,11 @@ def _build_write_payload(items: list[tuple[int, int, bytes]], protocol_version: payload += encode_uint32_vlq(total_field_count) for addr in addresses: payload += addr - for i, (_, _, data) in enumerate(items, 1): + for i, (_, _, data, datatype) in enumerate(normalized, 1): payload += encode_uint32_vlq(i) - payload += encode_pvalue_blob(data) + payload += encode_pvalue_typed(datatype, data) payload += bytes([0x00]) payload += encode_object_qualifier(protocol_version=protocol_version) - payload += encode_uint32_vlq(1) payload += struct.pack(">I", 0) return bytes(payload) @@ -861,7 +871,12 @@ def _build_substreamed_read_payload(session_id: int, access_area: int, access_su def _build_substreamed_write_payload( - session_id: int, access_area: int, access_sub_area: int, lids: list[int], data: bytes + session_id: int, + access_area: int, + access_sub_area: int, + lids: list[int], + data: bytes, + datatype: DataType = DataType.BLOB, ) -> bytes: """Build a SET_VAR_SUBSTREAMED payload for data access on V1-initial PLCs.""" oq = encode_object_qualifier(protocol_version=ProtocolVersion.V1) @@ -879,7 +894,7 @@ def _build_substreamed_write_payload( payload += oq payload += bytes([0x00]) payload += encode_uint32_vlq(1) - payload += encode_pvalue_blob(data) + payload += encode_pvalue_typed(datatype, data) payload += encode_uint32_vlq(1) payload += struct.pack(">I", 0) return bytes(payload) @@ -920,7 +935,6 @@ def _build_area_read_payload(area_rid: int, start: int, size: int, protocol_vers payload += encode_uint32_vlq(field_count) payload += addr_bytes payload += encode_object_qualifier(protocol_version=protocol_version) - payload += encode_uint32_vlq(1) payload += struct.pack(">I", 0) return bytes(payload) @@ -942,7 +956,6 @@ def _build_area_write_payload(area_rid: int, start: int, data: bytes, protocol_v payload += encode_pvalue_blob(data) payload += bytes([0x00]) payload += encode_object_qualifier(protocol_version=protocol_version) - payload += encode_uint32_vlq(1) payload += struct.pack(">I", 0) return bytes(payload) @@ -951,8 +964,6 @@ def _build_symbolic_read_payload( access_area: int, lids: list[int], symbol_crc: int = 0, - with_integrity: bool = True, - integrity_id: int = 1, protocol_version: int = ProtocolVersion.V2, ) -> bytes: """Build a GetMultiVariables payload for symbolic (LID-based) access. @@ -982,8 +993,6 @@ def _build_symbolic_read_payload( payload += encode_uint32_vlq(field_count) payload += addr_bytes payload += encode_object_qualifier(protocol_version=protocol_version) - if with_integrity: - payload += encode_uint32_vlq(integrity_id) payload += struct.pack(">I", 0) return bytes(payload) @@ -1017,7 +1026,6 @@ def _build_symbolic_write_payload( payload += encode_pvalue_blob(data) payload += bytes([0x00]) payload += encode_object_qualifier(protocol_version=protocol_version) - payload += encode_uint32_vlq(1) payload += struct.pack(">I", 0) return bytes(payload) diff --git a/s7commplus/codec.py b/s7commplus/codec.py index 8a9ba3fa..c6630713 100644 --- a/s7commplus/codec.py +++ b/s7commplus/codec.py @@ -379,6 +379,42 @@ def encode_pvalue_blob(data: bytes) -> bytes: return bytes(result) +def encode_pvalue_typed(datatype: DataType, data: bytes) -> bytes: + """Encode already-serialized value bytes with an explicit PValue type. + + Raw DB offsets do not carry enough metadata to infer the PLC datatype. + Callers writing anything other than a byte-array/BLOB must therefore + provide the type reported by the PLC's symbol information. + """ + datatype = DataType(datatype) + fixed_sizes = { + DataType.BOOL: 1, + DataType.USINT: 1, + DataType.BYTE: 1, + DataType.SINT: 1, + DataType.UINT: 2, + DataType.WORD: 2, + DataType.INT: 2, + DataType.DWORD: 4, + DataType.REAL: 4, + DataType.LWORD: 8, + DataType.LREAL: 8, + DataType.TIMESTAMP: 8, + DataType.RID: 4, + } + expected_size = fixed_sizes.get(datatype) + if expected_size is not None and len(data) != expected_size: + raise ValueError(f"{datatype.name} requires {expected_size} encoded bytes, got {len(data)}") + if datatype == DataType.NULL and data: + raise ValueError("NULL requires an empty value") + + result = bytearray((0x00, datatype)) + if datatype in (DataType.BLOB, DataType.WSTRING, DataType.S7STRING): + result += encode_uint32_vlq(len(data)) + result += data + return bytes(result) + + def decode_pvalue_to_bytes(data: bytes, offset: int) -> tuple[bytes, int]: """Decode a PValue from S7CommPlus response to raw bytes. diff --git a/tests/test_s7_codec.py b/tests/test_s7_codec.py index f284d1ba..af3ac4cc 100644 --- a/tests/test_s7_codec.py +++ b/tests/test_s7_codec.py @@ -29,6 +29,7 @@ encode_item_address, encode_object_qualifier, encode_pvalue_blob, + encode_pvalue_typed, encode_request_header, encode_typed_value, encode_uint8, @@ -409,6 +410,15 @@ def test_roundtrip_with_decode(self) -> None: assert consumed == len(encoded) +class TestPValueTyped: + def test_word_uses_raw_big_endian_bytes(self) -> None: + assert encode_pvalue_typed(DataType.WORD, b"\x02\x00") == bytes((0x00, DataType.WORD, 0x02, 0x00)) + + def test_rejects_wrong_fixed_width(self) -> None: + with pytest.raises(ValueError, match="REAL requires 4 encoded bytes"): + encode_pvalue_typed(DataType.REAL, b"\x00\x00") + + class TestDecodePValue: """Test decode_pvalue_to_bytes for all scalar and array type branches.""" diff --git a/tests/test_s7_unit.py b/tests/test_s7_unit.py index 9acb272e..6749c054 100644 --- a/tests/test_s7_unit.py +++ b/tests/test_s7_unit.py @@ -201,40 +201,43 @@ def test_write_read_consistency(self) -> None: assert isinstance(write_payload, bytes) -class TestSequenceNumber: - """Verify all payload builders include a SequenceNumber after ObjectQualifier.""" +class TestIntegrityPlaceholder: + """Verify payload builders leave IntegrityId insertion to the connection.""" @staticmethod - def _has_sequence_number(payload: bytes) -> bool: + def _has_only_trailing_fill(payload: bytes) -> bool: oq = encode_object_qualifier() idx = bytes(payload).find(oq) assert idx >= 0, "ObjectQualifier not found in payload" - seq_offset = idx + len(oq) - return payload[seq_offset : seq_offset + 1] == encode_uint32_vlq(1) + return payload[idx + len(oq) :] == bytes(4) - def test_read_payload_has_sequence_number(self) -> None: + def test_read_payload_has_no_static_integrity_id(self) -> None: payload = _build_read_payload([(1, 0, 4)]) - assert self._has_sequence_number(payload) + assert self._has_only_trailing_fill(payload) - def test_write_payload_has_sequence_number(self) -> None: + def test_write_payload_has_no_static_integrity_id(self) -> None: payload = _build_write_payload([(1, 0, bytes([1, 2, 3, 4]))]) - assert self._has_sequence_number(payload) + assert self._has_only_trailing_fill(payload) - def test_area_read_payload_has_sequence_number(self) -> None: + def test_area_read_payload_has_no_static_integrity_id(self) -> None: payload = _build_area_read_payload(82, 0, 4) - assert self._has_sequence_number(payload) + assert self._has_only_trailing_fill(payload) - def test_area_write_payload_has_sequence_number(self) -> None: + def test_area_write_payload_has_no_static_integrity_id(self) -> None: payload = _build_area_write_payload(82, 0, b"\x00\x00\x00\x00") - assert self._has_sequence_number(payload) + assert self._has_only_trailing_fill(payload) - def test_symbolic_read_payload_has_sequence_number(self) -> None: + def test_symbolic_read_payload_has_no_static_integrity_id(self) -> None: payload = _build_symbolic_read_payload(0x8A0E0001, [1, 4]) - assert self._has_sequence_number(payload) + assert self._has_only_trailing_fill(payload) - def test_symbolic_write_payload_has_sequence_number(self) -> None: + def test_symbolic_write_payload_has_no_static_integrity_id(self) -> None: payload = _build_symbolic_write_payload(0x8A0E0001, [1, 4], b"\x01") - assert self._has_sequence_number(payload) + assert self._has_only_trailing_fill(payload) + + def test_write_payload_encodes_explicit_datatype(self) -> None: + payload = _build_write_payload([(1, 0, struct.pack(">f", 2.0), DataType.REAL)]) + assert bytes((0x00, DataType.REAL)) + struct.pack(">f", 2.0) in payload # -- Connection unit tests -- From 6959ed60d34fadf6b0c72f754f536b74be9d277e Mon Sep 17 00:00:00 2001 From: Gijs Molenaar Date: Tue, 8 Sep 2026 07:35:44 +0200 Subject: [PATCH 3/3] fix(s7commplus): encode integer PValues as VLQ --- s7commplus/codec.py | 17 ++++++++++++++++- tests/test_s7_codec.py | 17 +++++++++++++++++ tests/test_s7_unit.py | 27 +++++++++++++++++++++++++-- 3 files changed, 58 insertions(+), 3 deletions(-) diff --git a/s7commplus/codec.py b/s7commplus/codec.py index c6630713..a9879d9e 100644 --- a/s7commplus/codec.py +++ b/s7commplus/codec.py @@ -395,12 +395,18 @@ def encode_pvalue_typed(datatype: DataType, data: bytes) -> bytes: DataType.UINT: 2, DataType.WORD: 2, DataType.INT: 2, + DataType.UDINT: 4, DataType.DWORD: 4, + DataType.DINT: 4, DataType.REAL: 4, + DataType.ULINT: 8, DataType.LWORD: 8, + DataType.LINT: 8, DataType.LREAL: 8, DataType.TIMESTAMP: 8, + DataType.TIMESPAN: 8, DataType.RID: 4, + DataType.AID: 4, } expected_size = fixed_sizes.get(datatype) if expected_size is not None and len(data) != expected_size: @@ -411,7 +417,16 @@ def encode_pvalue_typed(datatype: DataType, data: bytes) -> bytes: result = bytearray((0x00, datatype)) if datatype in (DataType.BLOB, DataType.WSTRING, DataType.S7STRING): result += encode_uint32_vlq(len(data)) - result += data + if datatype in (DataType.UDINT, DataType.AID): + result += encode_uint32_vlq(int.from_bytes(data, "big")) + elif datatype == DataType.ULINT: + result += encode_uint64_vlq(int.from_bytes(data, "big")) + elif datatype == DataType.DINT: + result += encode_int32_vlq(int.from_bytes(data, "big", signed=True)) + elif datatype in (DataType.LINT, DataType.TIMESPAN): + result += encode_int64_vlq(int.from_bytes(data, "big", signed=True)) + else: + result += data return bytes(result) diff --git a/tests/test_s7_codec.py b/tests/test_s7_codec.py index af3ac4cc..a3ee0b7f 100644 --- a/tests/test_s7_codec.py +++ b/tests/test_s7_codec.py @@ -414,10 +414,27 @@ class TestPValueTyped: def test_word_uses_raw_big_endian_bytes(self) -> None: assert encode_pvalue_typed(DataType.WORD, b"\x02\x00") == bytes((0x00, DataType.WORD, 0x02, 0x00)) + @pytest.mark.parametrize( + ("datatype", "data", "encoded_value"), + [ + (DataType.UDINT, struct.pack(">I", 300), b"\x82\x2c"), + (DataType.DINT, struct.pack(">i", -5), b"\x7b"), + (DataType.ULINT, struct.pack(">Q", 300), b"\x82\x2c"), + (DataType.LINT, struct.pack(">q", -5), b"\x7b"), + (DataType.TIMESPAN, struct.pack(">q", -5), b"\x7b"), + (DataType.AID, struct.pack(">I", 300), b"\x82\x2c"), + ], + ) + def test_variable_length_integer_types_use_vlq(self, datatype: DataType, data: bytes, encoded_value: bytes) -> None: + assert encode_pvalue_typed(datatype, data) == bytes((0x00, datatype)) + encoded_value + def test_rejects_wrong_fixed_width(self) -> None: with pytest.raises(ValueError, match="REAL requires 4 encoded bytes"): encode_pvalue_typed(DataType.REAL, b"\x00\x00") + with pytest.raises(ValueError, match="UDINT requires 4 encoded bytes"): + encode_pvalue_typed(DataType.UDINT, b"\x00\x00") + class TestDecodePValue: """Test decode_pvalue_to_bytes for all scalar and array type branches.""" diff --git a/tests/test_s7_unit.py b/tests/test_s7_unit.py index 6749c054..d060c8b8 100644 --- a/tests/test_s7_unit.py +++ b/tests/test_s7_unit.py @@ -20,10 +20,10 @@ _build_substreamed_write_payload, ) from s7commplus.connection import S7CommPlusConnection, _strip_paom_string_in_session_version -from s7commplus.codec import encode_object_qualifier, encode_pvalue_blob +from s7commplus.codec import encode_header, encode_object_qualifier, encode_pvalue_blob from s7commplus.codec import _pvalue_element_size as _element_size from s7commplus.codec import skip_typed_value, parse_server_session_version -from s7commplus.protocol import DataType, ElementID, FunctionCode, Ids, ObjectId +from s7commplus.protocol import DataType, ElementID, FunctionCode, Ids, ObjectId, Opcode, ProtocolVersion from s7commplus.vlq import ( encode_uint32_vlq, encode_uint64_vlq, @@ -239,6 +239,29 @@ def test_write_payload_encodes_explicit_datatype(self) -> None: payload = _build_write_payload([(1, 0, struct.pack(">f", 2.0), DataType.REAL)]) assert bytes((0x00, DataType.REAL)) + struct.pack(">f", 2.0) in payload + @pytest.mark.parametrize(("with_integrity", "integrity_id"), [(False, 0), (True, 7)]) + def test_connection_conditionally_inserts_integrity_id(self, with_integrity: bool, integrity_id: int) -> None: + payload = _build_read_payload([(1, 0, 4)]) + response = struct.pack(">BHHHHB", Opcode.RESPONSE, 0, FunctionCode.GET_MULTI_VARIABLES, 0, 1, 0x34) + connection = S7CommPlusConnection("127.0.0.1") + connection._connected = True + connection._protocol_version = ProtocolVersion.V2 + connection._with_integrity_id = with_integrity + connection._integrity_id_read = integrity_id + connection._send_s7_data = MagicMock() + connection._recv_s7_data = MagicMock( + return_value=encode_header(ProtocolVersion.V2, len(response)) + + response + + struct.pack(">BBH", 0x72, ProtocolVersion.V2, 0), + ) + + connection.send_request(FunctionCode.GET_MULTI_VARIABLES, payload) + + frame = connection._send_s7_data.call_args.args[0] + sent_payload = frame[4 + 14 : -4] + expected = payload[:-4] + (encode_uint32_vlq(integrity_id) if with_integrity else b"") + payload[-4:] + assert sent_payload == expected + # -- Connection unit tests --