diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml index 7bb2df3..15e0b1d 100644 --- a/.pre-commit-config.yaml +++ b/.pre-commit-config.yaml @@ -16,10 +16,12 @@ repos: - id: check-docstring-first - id: detect-private-key - # - repo: https://github.com/pre-commit/mirrors-mypy - # rev: 'v1.10.1' - # hooks: - # - id: mypy + - repo: https://github.com/pre-commit/mirrors-mypy + rev: 'v1.10.1' + hooks: + - id: mypy + additional_dependencies: + - pytest - repo: https://github.com/astral-sh/ruff-pre-commit rev: 'v0.5.1' diff --git a/src/python_s7comm/async_client.py b/src/python_s7comm/async_client.py index db1538d..c4c85d6 100644 --- a/src/python_s7comm/async_client.py +++ b/src/python_s7comm/async_client.py @@ -1,5 +1,4 @@ import datetime -from typing import List, Optional, Tuple from .s7comm import AsyncS7Comm, enums from .s7comm.packets.variable_address import VariableAddress @@ -50,12 +49,12 @@ async def get_cpu_state(self) -> enums.CPUStatus: async def read_area(self, address: str) -> bytes: return await self.s7comm.read_area(VariableAddress.from_string(address)) - async def write_area(self, address: str, data: bytes) -> int: - return await self.s7comm.write_area(address=VariableAddress.from_string(address), data=data) + async def write_area(self, address: str, data: bytes) -> None: + await self.s7comm.write_area(address=VariableAddress.from_string(address), data=data) - async def get_order_code(self) -> Optional[str]: + async def get_order_code(self) -> str | None: response = await self.s7comm.read_szl(szl_id=0x0011, szl_index=0x0000) - szl_data = SZLResponseData.parse(response.data) + szl_data = SZLResponseData.parse(response.data.data) for date_tree in szl_data.szl_data_tree_list: module_identification = ModuleIdentificationDataTree.parse(date_tree) if module_identification.index == ModuleIdentificationIndex.MODULE_IDENTIFICATION: @@ -63,18 +62,19 @@ async def get_order_code(self) -> Optional[str]: return None async def read_szl(self, szl_id: int, szl_index: int = 0x0000) -> SZLResponseData: - return await self.s7comm.read_szl(szl_id=szl_id, szl_index=szl_index) + response_data = await self.s7comm.read_szl(szl_id=szl_id, szl_index=szl_index) + return SZLResponseData.parse(response_data.data.data) async def read_szl_list(self) -> SZLResponseData: response_data = await self.s7comm.read_szl(szl_id=0x0000, szl_index=0x0000) return SZLResponseData.parse(response_data.data.data) - async def read_multi_vars(self, items: List[str]) -> List[bytes]: + async def read_multi_vars(self, items: list[str]) -> list[bytes]: vars_ = [VariableAddress.from_string(item) for item in items] response = await self.s7comm.read_multi_vars(items=vars_) return response.values() - async def write_multi_vars(self, items: List[Tuple[str, bytes]]) -> bool: + async def write_multi_vars(self, items: list[tuple[str, bytes]]) -> bool: vars_ = [(VariableAddress.from_string(address), data) for address, data in items] response = await self.s7comm.write_multi_vars(items=vars_) return response.check_result() @@ -85,7 +85,7 @@ async def set_plc_system_datetime(self) -> int: async def delete(self, block_type: str, block_num: int) -> int: raise NotImplementedError - async def full_upload(self, _type: str, block_num: int) -> Tuple[bytearray, int]: + async def full_upload(self, _type: str, block_num: int) -> tuple[bytearray, int]: raise NotImplementedError async def upload(self, block_num: int) -> bytearray: diff --git a/src/python_s7comm/s7comm/__init__.py b/src/python_s7comm/s7comm/__init__.py index 2d06ac9..826a415 100644 --- a/src/python_s7comm/s7comm/__init__.py +++ b/src/python_s7comm/s7comm/__init__.py @@ -1,5 +1,6 @@ from .async_client import AsyncS7Comm from .client import S7Comm +from .exceptions import PacketLostError, StalePacketError -__all__ = ["S7Comm", "AsyncS7Comm"] +__all__ = ["S7Comm", "AsyncS7Comm", "PacketLostError", "StalePacketError"] diff --git a/src/python_s7comm/s7comm/async_client.py b/src/python_s7comm/s7comm/async_client.py index 5a12a91..acf59aa 100644 --- a/src/python_s7comm/s7comm/async_client.py +++ b/src/python_s7comm/s7comm/async_client.py @@ -4,7 +4,7 @@ from typing import cast from .base import BaseS7Comm -from .enums import HeaderError, ItemReturnCode, MessageType, SubfunctionCode, UserdataFunction +from .enums import HeaderError, ItemReturnCode, MessageType, SubfunctionCode, UserdataFunction, UserdataLastPDU from .error_messages import ERROR_MESSAGES from .exceptions import PacketLostError, StalePacketError from .packets import ( @@ -12,6 +12,7 @@ S7AckDataHeader, S7Packet, SetupCommunicationRequest, + SZLResponseData, UserDataContinuationRequest, UserDataRequest, UserDataResponse, @@ -38,6 +39,7 @@ def __init__( transport: AsyncTransport | None = None, ): super().__init__(pdu_length=pdu_length) + self.transport: AsyncTransport if transport is None: self.transport = AsyncCOTP(tpdu_size=tpdu_size, source_tsap=source_tsap, dest_tsap=dest_tsap) else: @@ -132,7 +134,7 @@ async def read_area(self, address: VariableAddress) -> bytes: result += response_item.data return result - async def write_area(self, address: VariableAddress, data: bytes) -> int: + async def write_area(self, address: VariableAddress, data: bytes) -> WriteVariableResponse: """ Writes data to PLC memory area with automatic splitting into multiple requests if data exceeds PDU size. @@ -177,7 +179,7 @@ async def write_multi_vars(self, items: list[tuple[VariableAddress, bytes]]) -> raise ValueError("Invalid response class") return response - async def read_szl(self, szl_id: int, szl_index: int) -> S7Packet: + async def read_szl(self, szl_id: int, szl_index: int) -> UserDataResponse: data = struct.pack("!HH", szl_id, szl_index) request = UserDataRequest.create( function_group=UserdataFunction.CPU_FUNCTION, @@ -186,6 +188,8 @@ async def read_szl(self, szl_id: int, szl_index: int) -> S7Packet: data=data, ) response = await self.send(request=request) + if not isinstance(response, UserDataResponse): + raise ValueError("Invalid response class") return response async def plc_stop(self) -> S7Packet: diff --git a/src/python_s7comm/s7comm/base.py b/src/python_s7comm/s7comm/base.py index 78dfe23..876f3d0 100644 --- a/src/python_s7comm/s7comm/base.py +++ b/src/python_s7comm/s7comm/base.py @@ -47,6 +47,7 @@ def _create_packet(request: S7Packet, pdu_reference: int) -> bytes: def _validate_pdu_reference(self, response: S7Packet) -> None: """Raises if PDU reference is invalid.""" + assert response.header is not None if response.header.pdu_reference > self._pdu_reference: raise PacketLostError(f"Expected {self._pdu_reference}, got {response.header.pdu_reference}") elif response.header.pdu_reference < self._pdu_reference: diff --git a/src/python_s7comm/s7comm/client.py b/src/python_s7comm/s7comm/client.py index e8631b5..7fa534c 100644 --- a/src/python_s7comm/s7comm/client.py +++ b/src/python_s7comm/s7comm/client.py @@ -37,6 +37,7 @@ def __init__( transport: Transport | None = None, ) -> None: super().__init__(pdu_length=pdu_length) + self.transport: Transport if transport is None: self.transport = COTP(tpdu_size=tpdu_size, source_tsap=source_tsap, dest_tsap=dest_tsap) else: diff --git a/src/python_s7comm/s7comm/packets/__init__.py b/src/python_s7comm/s7comm/packets/__init__.py index a192448..2e4a88b 100644 --- a/src/python_s7comm/s7comm/packets/__init__.py +++ b/src/python_s7comm/s7comm/packets/__init__.py @@ -1,19 +1,23 @@ +from .data_item import DataItem from .headers import S7AckDataHeader, S7Header from .packet import S7Packet from .plc_command import RequestPLCStop from .rw_variable import VariableReadRequest, VariableWriteRequest from .setup_communication import SetupCommunicationParameter, SetupCommunicationRequest +from .szl import SZLResponseData from .user_data import UserDataContinuationRequest, UserDataRequest, UserDataResponse from .variable_address import VariableAddress __all__ = [ + "DataItem", "S7AckDataHeader", "S7Parameter", "RequestPLCStop", "SetupCommunicationRequest", "SetupCommunicationParameter", "S7Packet", + "SZLResponseData", "UserDataContinuationRequest", "UserDataRequest", "UserDataResponse", diff --git a/src/python_s7comm/s7comm/packets/error.py b/src/python_s7comm/s7comm/packets/error.py index e4c83f0..cbc2b1c 100644 --- a/src/python_s7comm/s7comm/packets/error.py +++ b/src/python_s7comm/s7comm/packets/error.py @@ -16,3 +16,7 @@ def serialize_parameter(self) -> bytes: def serialize_data(self) -> bytes: return b"" + + @classmethod + def parse(cls, packet: bytes) -> "S7Error": + raise NotImplementedError("S7Error is created from header, not parsed from bytes") diff --git a/src/python_s7comm/s7comm/packets/packet.py b/src/python_s7comm/s7comm/packets/packet.py index 15ec0a4..babb6fd 100644 --- a/src/python_s7comm/s7comm/packets/packet.py +++ b/src/python_s7comm/s7comm/packets/packet.py @@ -1,4 +1,4 @@ -from typing import Any, ClassVar, Protocol +from typing import Any, ClassVar, Protocol, Self from .headers import S7AckDataHeader, S7Header @@ -7,16 +7,23 @@ class S7Data(Protocol): def serialize(self) -> bytes: ... +class S7Parameter(Protocol): + def serialize(self) -> bytes: ... + + class S7Packet(Protocol): MESSAGE_TYPE: ClassVar header: S7Header | None - parameter: Any # Optional[S7Parameter] - data: Any # Optional[S7Data] # Union[DataItem, List[DataItem], List[enums.ItemReturnCode], None] + parameter: Any # S7Parameter | None + data: Any # S7Data | list[S7Data] | None def serialize_parameter(self) -> bytes: ... def serialize_data(self) -> bytes: ... + @classmethod + def parse(cls, packet: bytes) -> "S7Packet": ... + class S7Response(S7Packet): def create_packet(self, pdu_reference: int) -> bytes: diff --git a/src/python_s7comm/s7comm/packets/parser.py b/src/python_s7comm/s7comm/packets/parser.py index e1370b8..e7d64e4 100644 --- a/src/python_s7comm/s7comm/packets/parser.py +++ b/src/python_s7comm/s7comm/packets/parser.py @@ -14,13 +14,13 @@ from .user_data import UserDataResponse -response_code_to_packet: dict[FunctionCode, S7Packet] = { +response_code_to_packet: dict[FunctionCode, type[S7Packet]] = { FunctionCode.SetupCommunication: SetupCommunicationRequest, FunctionCode.ReadVariable: ReadVariableResponse, FunctionCode.WriteVariable: WriteVariableResponse, } -request_code_to_packet: dict[FunctionCode, S7Packet] = { +request_code_to_packet: dict[FunctionCode, type[S7Packet]] = { FunctionCode.SetupCommunication: SetupCommunicationRequest, FunctionCode.ReadVariable: VariableReadRequest, FunctionCode.WriteVariable: VariableWriteRequest, @@ -47,12 +47,12 @@ def parse(packet: bytes) -> S7Packet: # parse parameter parameter = packet[header.LENGTH :] - response = response_code_to_packet.get(function_code).parse(parameter) + response = response_code_to_packet[function_code].parse(parameter) elif message_type == MessageType.JobRequest: header = S7Header.parse(packet) function_code = struct.unpack_from("!B", packet, header.LENGTH)[0] parameter = packet[header.LENGTH :] - response = request_code_to_packet.get(function_code).parse(parameter) + response = request_code_to_packet[function_code].parse(parameter) elif message_type == MessageType.Userdata: header = S7Header.parse(packet) response = UserDataResponse.parse(packet[header.LENGTH :]) diff --git a/src/python_s7comm/s7comm/packets/plc_command.py b/src/python_s7comm/s7comm/packets/plc_command.py index 428577f..c8ac8e1 100644 --- a/src/python_s7comm/s7comm/packets/plc_command.py +++ b/src/python_s7comm/s7comm/packets/plc_command.py @@ -26,6 +26,10 @@ def serialize_parameter(self) -> bytes: def serialize_data(self) -> bytes: return b"" + @classmethod + def parse(cls, packet: bytes) -> "RequestPLCStop": + raise NotImplementedError("RequestPLCStop is created from constructor, not parsed from bytes") + class PIServiceRequest: UNKNOWN_BYTES = b"\x00\x00\x00\x00\x00\x00\xfd" diff --git a/src/python_s7comm/s7comm/packets/rw_variable.py b/src/python_s7comm/s7comm/packets/rw_variable.py index 0ae7c2d..3b5e11b 100644 --- a/src/python_s7comm/s7comm/packets/rw_variable.py +++ b/src/python_s7comm/s7comm/packets/rw_variable.py @@ -1,6 +1,6 @@ import struct from dataclasses import dataclass -from typing import AsyncGenerator, ClassVar, Generator, List, Tuple +from typing import AsyncGenerator, ClassVar, Generator from ..enums import ( Area, @@ -105,9 +105,9 @@ def parse(cls, packet: bytes) -> "RequestParameterItem": class VariableRequestParameter: LENGTH = 2 - def __init__(self, function_code: FunctionCode, items: List[RequestParameterItem]): + def __init__(self, function_code: FunctionCode, items: list[RequestParameterItem]): self.function_code = function_code - self.items: List[RequestParameterItem] = items + self.items: list[RequestParameterItem] = items def serialize(self) -> bytes: parameter = struct.pack("!BB", self.function_code, len(self.items)) @@ -127,13 +127,13 @@ def __init__(self, parameter: VariableRequestParameter) -> None: self.data = None @classmethod - def create(cls, items: List[VariableAddress]) -> "VariableReadRequest": + def create(cls, items: list[VariableAddress]) -> "VariableReadRequest": parameter_items = cls._create_parameter_item_list(items=items) parameter = VariableRequestParameter(function_code=cls.FUNCTION_CODE, items=parameter_items) return cls(parameter=parameter) @staticmethod - def _create_parameter_item_list(items: List[VariableAddress]) -> List[RequestParameterItem]: + def _create_parameter_item_list(items: list[VariableAddress]) -> list[RequestParameterItem]: parameter_items = [] for item in items: parameter_item = RequestParameterItem(address=item) @@ -194,13 +194,13 @@ class VariableWriteRequest(S7Packet): MESSAGE_TYPE = MessageType.JobRequest FUNCTION_CODE = FunctionCode.WriteVariable - def __init__(self, parameter: VariableRequestParameter, data: List[DataItem]) -> None: + def __init__(self, parameter: VariableRequestParameter, data: list[DataItem]) -> None: self.header = None self.parameter = parameter self.data = data @classmethod - def create(cls, items: List[Tuple[VariableAddress, bytes]]) -> "VariableWriteRequest": + def create(cls, items: list[tuple[VariableAddress, bytes]]) -> "VariableWriteRequest": parameter_items = [] data_items = [] for item, data in items: @@ -314,7 +314,8 @@ def serialize(self) -> bytes: class ReadVariableResponse(S7Packet): MESSAGE_TYPE = MessageType.Response - def __init__(self, parameter: VariableParameter, data: List[DataItem]) -> None: + def __init__(self, parameter: VariableParameter, data: list[DataItem]) -> None: + self.header = None self.parameter = parameter self.data = data @@ -346,7 +347,7 @@ def parse(cls, packet: bytes) -> "ReadVariableResponse": parameter = VariableParameter(function_code=function_code, item_count=item_count) return cls(parameter=parameter, data=items) - def values(self) -> List[bytes]: + def values(self) -> list[bytes]: """Возвращает только значения в виде списка, после проверки на результат ответа""" result = [] for item in self.data: @@ -355,20 +356,27 @@ def values(self) -> List[bytes]: result.append(item.data) return result + def serialize_parameter(self) -> bytes: + return self.parameter.serialize() + + def serialize_data(self) -> bytes: + return b"".join(item.serialize() for item in self.data) + class WriteVariableResponse(S7Packet): MESSAGE_TYPE = MessageType.Response - def __init__(self, parameter: VariableParameter, data: List[ItemReturnCode]) -> None: + def __init__(self, parameter: VariableParameter, data: list[ItemReturnCode]) -> None: + self.header = None self.parameter = parameter - self.data: List[ItemReturnCode] = data + self.data: list[ItemReturnCode] = data @classmethod def parse(cls, packet: bytes) -> "WriteVariableResponse": function_code, item_count = struct.unpack_from("!BB", packet) offset = 2 parameter = VariableParameter(function_code=function_code, item_count=item_count) - data: List[ItemReturnCode] = [ItemReturnCode(item) for item in packet[offset:item_count]] + data: list[ItemReturnCode] = [ItemReturnCode(item) for item in packet[offset:item_count]] return cls(parameter=parameter, data=data) def serialize_parameter(self) -> bytes: diff --git a/src/python_s7comm/s7comm/packets/szl.py b/src/python_s7comm/s7comm/packets/szl.py index dd3167f..fd5f88c 100644 --- a/src/python_s7comm/s7comm/packets/szl.py +++ b/src/python_s7comm/s7comm/packets/szl.py @@ -1,6 +1,5 @@ import struct from enum import IntEnum -from typing import List from ..enums import CPUStatus @@ -81,7 +80,7 @@ def parse(cls, packet: bytes) -> "ModuleIdentificationDataTree": class SZLResponseData: DATA_TREE_OFFSET = 8 - def __init__(self, szl_id: int, szl_index: int, szl_data_tree_list: List[bytes]) -> None: + def __init__(self, szl_id: int, szl_index: int, szl_data_tree_list: list[bytes]) -> None: self.szl_id = szl_id self.szl_index = szl_index self.szl_data_tree_list = szl_data_tree_list diff --git a/src/python_s7comm/s7comm/packets/user_data.py b/src/python_s7comm/s7comm/packets/user_data.py index 3dbb12b..f366d85 100644 --- a/src/python_s7comm/s7comm/packets/user_data.py +++ b/src/python_s7comm/s7comm/packets/user_data.py @@ -59,7 +59,7 @@ def __init__( self.subfunction = subfunction self.sequence_number = sequence_number self.data_unit_reference = 0 - self.last_data_unit = 0 + self.last_data_unit = UserdataLastPDU.YES self.error_code = 0 def serialize(self) -> bytes: @@ -82,7 +82,7 @@ class UserDataResponseParameter(UserDataRequestParameter): def __init__( self, data_unit_reference: int, - last_data_unit: bool, + last_data_unit: UserdataLastPDU, error_code: int, method: UserdataMethod, type: TransmissionType, diff --git a/src/python_s7comm/s7comm/szl.py b/src/python_s7comm/s7comm/szl.py index acf300d..bae738d 100644 --- a/src/python_s7comm/s7comm/szl.py +++ b/src/python_s7comm/s7comm/szl.py @@ -1,6 +1,5 @@ import struct from enum import IntEnum -from typing import List from .enums import CPUStatus @@ -81,7 +80,7 @@ def parse(cls, packet: bytes) -> "ModuleIdentificationDataTree": class SZLResponseData: DATA_TREE_OFFSET = 8 - def __init__(self, szl_id: int, szl_index: int, szl_data_tree_list: List[bytes]) -> None: + def __init__(self, szl_id: int, szl_index: int, szl_data_tree_list: list[bytes]) -> None: self.szl_id = szl_id self.szl_index = szl_index self.szl_data_tree_list = szl_data_tree_list diff --git a/src/python_s7comm/s7comm/transport/transport.py b/src/python_s7comm/s7comm/transport/transport.py index ec234ec..459f81d 100644 --- a/src/python_s7comm/s7comm/transport/transport.py +++ b/src/python_s7comm/s7comm/transport/transport.py @@ -2,7 +2,7 @@ class Transport(Protocol): - def connect(self) -> None: ... + def connect(self, address: str, rack: int, slot: int, port: int = 102) -> None: ... def close(self) -> None: ... def send(self, payload: bytes) -> None: ... def receive(self) -> bytes: ... diff --git a/src/python_s7comm/sync_client.py b/src/python_s7comm/sync_client.py index 5b2f7b3..29807f4 100644 --- a/src/python_s7comm/sync_client.py +++ b/src/python_s7comm/sync_client.py @@ -1,5 +1,4 @@ import datetime -from typing import List, Optional, Tuple from .s7comm import S7Comm, enums from .s7comm.packets import S7Packet @@ -26,7 +25,12 @@ def __init__( def __enter__(self) -> "Client": return self - def __exit__(self, exc_type, exc_val, exc_tb) -> None: + def __exit__( + self, + exc_type: type[BaseException] | None, + exc_val: BaseException | None, + exc_tb: object, + ) -> None: self.close() def connect(self, address: str, rack: int, slot: int, port: int = 102) -> None: @@ -55,7 +59,7 @@ def read_area(self, address: str) -> bytes: def write_area(self, address: str, data: bytes) -> None: self.s7comm.write_area(address=VariableAddress.from_string(address), data=data) - def get_order_code(self) -> Optional[str]: + def get_order_code(self) -> str | None: response = self.s7comm.read_szl(szl_id=0x0011, szl_index=0x0000) szl_data = SZLResponseData.parse(response.data.data) for date_tree in szl_data.szl_data_tree_list: @@ -71,12 +75,12 @@ def read_szl_list(self) -> SZLResponseData: response_data = self.s7comm.read_szl(szl_id=0x0000, szl_index=0x0000) return SZLResponseData.parse(response_data.data.data) - def read_multi_vars(self, items: List[str]) -> List[bytes]: + def read_multi_vars(self, items: list[str]) -> list[bytes]: items_ = [VariableAddress.from_string(item) for item in items] response = self.s7comm.read_multi_vars(items=items_) return response.values() - def write_multi_vars(self, items: List[Tuple[str, bytes]]) -> bool: + def write_multi_vars(self, items: list[tuple[str, bytes]]) -> bool: vars_ = [(VariableAddress.from_string(address), data) for address, data in items] response = self.s7comm.write_multi_vars(items=vars_) return response.check_result() @@ -90,7 +94,7 @@ def set_plc_system_datetime(self) -> int: def delete(self, block_type: str, block_num: int) -> int: raise NotImplementedError - def full_upload(self, _type: str, block_num: int) -> Tuple[bytearray, int]: + def full_upload(self, _type: str, block_num: int) -> tuple[bytearray, int]: raise NotImplementedError def upload(self, block_num: int) -> bytearray: diff --git a/tests/real_plc_s1200/conftest.py b/tests/real_plc_s1200/conftest.py index 23e5534..bdd1ebc 100644 --- a/tests/real_plc_s1200/conftest.py +++ b/tests/real_plc_s1200/conftest.py @@ -1,7 +1,5 @@ import os - -# import asyncio -from typing import Dict +from typing import AsyncGenerator, Generator, TypedDict import pytest from dotenv import load_dotenv @@ -16,7 +14,14 @@ SERVER_PORT = "102" -def get_plc_address() -> Dict: +class PLCConnectionParams(TypedDict): + address: str + rack: int + slot: int + port: int + + +def get_plc_address() -> PLCConnectionParams: IP = str(os.getenv("PLC_IP", SERVER_IP)) RACK = int(os.getenv("PLC_RACK", "0")) SLOT = int(os.getenv("PLC_SLOT", "1")) @@ -48,7 +53,7 @@ def get_plc_address() -> Dict: @pytest.fixture -async def async_client(): +async def async_client() -> AsyncGenerator[AsyncClient, None]: params = get_plc_address() async_client = AsyncClient() await async_client.connect(**params) @@ -57,7 +62,7 @@ async def async_client(): @pytest.fixture -def client(): +def client() -> Generator[Client, None, None]: params = get_plc_address() client = Client() client.connect(**params) diff --git a/tests/real_plc_s1200/test_async_read_area.py b/tests/real_plc_s1200/test_async_read_area.py index 622aced..9e83270 100644 --- a/tests/real_plc_s1200/test_async_read_area.py +++ b/tests/real_plc_s1200/test_async_read_area.py @@ -9,7 +9,7 @@ @pytest.mark.asyncio -async def test_read_int(async_client): +async def test_read_int(async_client: AsyncClient) -> None: response = await async_client.read_area("DB1.0 INT 2") result = struct.unpack(">2H", response) expected = (10, 255) @@ -17,33 +17,33 @@ async def test_read_int(async_client): @pytest.mark.asyncio -async def test_read_area_real(async_client): +async def test_read_area_real(async_client: AsyncClient) -> None: expected = data_items.RO_REAL_VALUE response = await async_client.read_area("DB1.4 REAL 2") response == expected @pytest.mark.asyncio -async def test_read_byte(async_client): +async def test_read_byte(async_client: AsyncClient) -> None: response = await async_client.read_area("DB1.12 BYTE 2") assert response == data_items.RO_BYTE_VALUE @pytest.mark.asyncio -async def test_read_word(async_client): +async def test_read_word(async_client: AsyncClient) -> None: response = await async_client.read_area(data_items.RO_WORD) assert response == data_items.RO_WORD_VALUE @pytest.mark.asyncio -async def test_read_dword(async_client): +async def test_read_dword(async_client: AsyncClient) -> None: response = await async_client.read_area(data_items.RO_DWORD) expected = data_items.RO_DWORD_VALUE assert response == expected @pytest.mark.asyncio -async def test_read_dint(async_client): +async def test_read_dint(async_client: AsyncClient) -> None: response = await async_client.read_area(data_items.RO_DINT) result = struct.unpack(">2I", response) expected = (2147483647, 42) @@ -51,7 +51,7 @@ async def test_read_dint(async_client): @pytest.mark.asyncio -async def test_read_char(async_client): +async def test_read_char(async_client: AsyncClient) -> None: expected = data_items.RO_CHAR_VALUE response = await async_client.read_area(data_items.RO_CHAR) assert response == expected @@ -59,13 +59,13 @@ async def test_read_char(async_client): @pytest.mark.asyncio @pytest.mark.parametrize("start_bit, expected", [(0, b"\x01"), (1, b"\x00")]) -async def test_read_bool(async_client, start_bit, expected): +async def test_read_bool(async_client: AsyncClient, start_bit: int, expected: bytes) -> None: response = await async_client.read_area(f"DB1.36.{start_bit} BOOL 1") assert response == expected @pytest.mark.asyncio -async def test_read_area_multiple_pdu(async_client): +async def test_read_area_multiple_pdu(async_client: AsyncClient) -> None: amount = 500 world_length = ParameterTransportSize.REAL response = await async_client.read_area(f"DB4.0 REAL {amount}") @@ -73,7 +73,7 @@ async def test_read_area_multiple_pdu(async_client): @pytest.mark.asyncio -async def test_read_multi_vars(async_client): +async def test_read_multi_vars(async_client: AsyncClient) -> None: expected = [ data_items.RO_REAL_VALUE, data_items.RO_BYTE_VALUE, diff --git a/tests/real_plc_s1200/test_async_read_szl.py b/tests/real_plc_s1200/test_async_read_szl.py index ac7c2b1..1b222f8 100644 --- a/tests/real_plc_s1200/test_async_read_szl.py +++ b/tests/real_plc_s1200/test_async_read_szl.py @@ -1,11 +1,12 @@ import pytest +from python_s7comm.async_client import AsyncClient from python_s7comm.s7comm import enums from python_s7comm.s7comm.szl import SZLResponseData @pytest.mark.asyncio -async def test_read_szl_list(async_client): +async def test_read_szl_list(async_client: AsyncClient) -> None: expected = SZLResponseData( szl_id=0, szl_index=0, @@ -19,7 +20,7 @@ async def test_read_szl_list(async_client): @pytest.mark.asyncio -async def test_read_cpu_state(async_client): +async def test_read_cpu_state(async_client: AsyncClient) -> None: expected = enums.CPUStatus.RUN response = await async_client.get_cpu_state() assert response == expected.value diff --git a/tests/real_plc_s1200/test_async_write_area.py b/tests/real_plc_s1200/test_async_write_area.py index 8e99639..3b80802 100644 --- a/tests/real_plc_s1200/test_async_write_area.py +++ b/tests/real_plc_s1200/test_async_write_area.py @@ -2,79 +2,81 @@ import pytest +from python_s7comm.async_client import AsyncClient + @pytest.mark.asyncio -async def test_write_area_int(async_client): +async def test_write_area_int(async_client: AsyncClient) -> None: current_value = await async_client.read_area("DB2.0 INT 2") current_value current_value a, b = struct.unpack("!2H", current_value) expected = struct.pack("!2H", a + 1, b + 1) - response = await async_client.write_area("DB2.0 INT 2", data=expected) + await async_client.write_area("DB2.0 INT 2", data=expected) response = await async_client.read_area("DB2.0 INT 2") assert response == expected @pytest.mark.asyncio -async def test_write_area_bytes(async_client): +async def test_write_area_bytes(async_client: AsyncClient) -> None: expected = b"\x1f\xf1" - response = await async_client.write_area("DB2.12 BYTE 2", data=expected) + await async_client.write_area("DB2.12 BYTE 2", data=expected) response = await async_client.read_area("DB2.12 BYTE 2") assert response == expected @pytest.mark.asyncio -async def test_write_area_real(async_client): +async def test_write_area_real(async_client: AsyncClient) -> None: current_value = await async_client.read_area("DB2.4 REAL 2") a, b = struct.unpack("!ff", current_value) expected = struct.pack("!ff", a + 1, b + 1) - response = await async_client.write_area("DB2.4 REAL 2", data=expected) + await async_client.write_area("DB2.4 REAL 2", data=expected) response = await async_client.read_area("DB2.4 REAL 2") assert response == expected @pytest.mark.asyncio -async def test_write_word(async_client): +async def test_write_word(async_client: AsyncClient) -> None: current_value = await async_client.read_area("DB2.14 WORD 2") a, b = struct.unpack("!HH", current_value) expected = struct.pack("!HH", a + 1, b + 1) - response = await async_client.write_area("DB2.14 WORD 2", data=expected) + await async_client.write_area("DB2.14 WORD 2", data=expected) response = await async_client.read_area("DB2.14 WORD 2") assert response == expected @pytest.mark.asyncio -async def test_write_dword(async_client): +async def test_write_dword(async_client: AsyncClient) -> None: current_value = await async_client.read_area("DB2.18 DWORD 2") a, b = struct.unpack("!II", current_value) expected = struct.pack("!II", a + 1, b + 1) - response = await async_client.write_area("DB2.18 DWORD 2", data=expected) + await async_client.write_area("DB2.18 DWORD 2", data=expected) response = await async_client.read_area("DB2.18 DWORD 2") assert response == expected @pytest.mark.asyncio -async def test_write_dint(async_client): +async def test_write_dint(async_client: AsyncClient) -> None: current_value = await async_client.read_area("DB2.26 DINT 2") a, b = struct.unpack("!II", current_value) expected = struct.pack("!II", a + 1, b + 1) - response = await async_client.write_area("DB2.26 DINT 2", data=expected) + await async_client.write_area("DB2.26 DINT 2", data=expected) response = await async_client.read_area("DB2.26 DINT 2") assert response == expected @pytest.mark.asyncio -async def test_write_char(async_client): +async def test_write_char(async_client: AsyncClient) -> None: current_value = await async_client.read_area("DB2.34 CHAR 2") a, b = struct.unpack("!BB", current_value) expected = struct.pack("!BB", a + 1, b + 1) - response = await async_client.write_area("DB2.34 CHAR 2", data=expected) + await async_client.write_area("DB2.34 CHAR 2", data=expected) response = await async_client.read_area("DB2.34 CHAR 2") assert response == expected @@ -82,18 +84,27 @@ async def test_write_char(async_client): @pytest.mark.asyncio @pytest.mark.parametrize("start_bit", [0, 1, 2, 3, 4, 5, 6, 7]) -async def test_write_bool(async_client, start_bit): +async def test_write_bool(async_client: AsyncClient, start_bit: int) -> None: current_value = await async_client.read_area(f"DB2.36.{start_bit} BOOL 1") a = struct.unpack("!?", current_value)[0] expected = struct.pack("!?", not a) - response = await async_client.write_area(f"DB2.36.{start_bit} BOOL 1", data=expected) + await async_client.write_area(f"DB2.36.{start_bit} BOOL 1", data=expected) response = await async_client.read_area(f"DB2.36.{start_bit} BOOL 1") assert response == expected @pytest.mark.asyncio -async def test_write_multi_vars(async_client): +async def test_write_multi_vars(async_client: AsyncClient) -> None: + addresses: list[str] = [ + "DB2.4 REAL 2", + "DB2.12 BYTE 2", + "DB2.14 WORD 2", + "DB2.18 DWORD 2", + "DB2.26 DINT 2", + "DB2.34 CHAR 2", + "DB2.36 BOOL 1", + ] expected = [ b"B\xf6\xe6fD\x07\xcdq", b"\x0f\xf0", @@ -103,25 +114,8 @@ async def test_write_multi_vars(async_client): b"F-", b"\x00", ] - items = [ - ("DB2.4 REAL 2", expected[0]), - ("DB2.12 BYTE 2", expected[1]), - ("DB2.14 WORD 2", expected[2]), - ("DB2.18 DWORD 2", expected[3]), - ("DB2.26 DINT 2", expected[4]), - ("DB2.34 CHAR 2", expected[5]), - ("DB2.36 BOOL 1", expected[6]), - ] + items = list(zip(addresses, expected)) response = await async_client.write_multi_vars(items=items) assert response - items = [ - "DB2.4 REAL 2", - "DB2.12 BYTE 2", - "DB2.14 WORD 2", - "DB2.18 DWORD 2", - "DB2.26 DINT 2", - "DB2.34 CHAR 2", - "DB2.36 BOOL 1", - ] - assert await async_client.read_multi_vars(items=items) == expected + assert await async_client.read_multi_vars(items=addresses) == expected diff --git a/tests/real_plc_s1200/test_read_area.py b/tests/real_plc_s1200/test_read_area.py index 03003fb..93d1de6 100644 --- a/tests/real_plc_s1200/test_read_area.py +++ b/tests/real_plc_s1200/test_read_area.py @@ -5,64 +5,65 @@ from python_s7comm.s7comm.enums import DataSizeByte, ParameterTransportSize from python_s7comm.s7comm.packets import exceptions +from python_s7comm.sync_client import Client -def test_read_int(client): +def test_read_int(client: Client) -> None: response = client.read_area(data_items.RO_INT) result = struct.unpack(">2H", response) expected = (10, 255) assert result == expected -def test_read_area_real(client): +def test_read_area_real(client: Client) -> None: expected = data_items.RO_REAL_VALUE response = client.read_area(data_items.RO_REAL) assert response == expected -def test_read_byte(client): +def test_read_byte(client: Client) -> None: response = client.read_area(data_items.RO_BYTE) assert response == data_items.RO_BYTE_VALUE -def test_read_word(client): +def test_read_word(client: Client) -> None: response = client.read_area(data_items.RO_WORD) assert response == data_items.RO_WORD_VALUE -def test_read_dword(client): +def test_read_dword(client: Client) -> None: response = client.read_area(data_items.RO_DWORD) expected = data_items.RO_DWORD_VALUE assert response == expected -def test_read_dint(client): +def test_read_dint(client: Client) -> None: response = client.read_area(data_items.RO_DINT) result = struct.unpack(">2I", response) expected = (2147483647, 42) assert result == expected -def test_read_char(client): +def test_read_char(client: Client) -> None: expected = data_items.RO_CHAR_VALUE response = client.read_area(data_items.RO_CHAR) assert response == expected @pytest.mark.parametrize("start_bit, expected", [(0, b"\x01"), (1, b"\x00")]) -def test_read_bool(client, start_bit, expected): +def test_read_bool(client: Client, start_bit: int, expected: bytes) -> None: response = client.read_area(f"DB1.36.{start_bit} BOOL 1") assert response == expected -def test_read_area_multiple_pdu(client): +def test_read_area_multiple_pdu(client: Client) -> None: amount = 500 world_length = ParameterTransportSize.REAL response = client.read_area("DB4.0 REAL 500") assert len(response) == amount * DataSizeByte[world_length.name] -def test_read_multi_vars(client): +def test_read_multi_vars(client: Client) -> None: expected = [ data_items.RO_REAL_VALUE, data_items.RO_BYTE_VALUE, @@ -85,27 +86,27 @@ def test_read_multi_vars(client): assert response == expected -def test_read_area_timer(client): +def test_read_area_timer(client: Client) -> None: with pytest.raises(exceptions.ReadVariableException): client.read_area("T0 TIMER 1") -def test_read_area_counter(client): +def test_read_area_counter(client: Client) -> None: with pytest.raises(exceptions.ReadVariableException): client.read_area("C0 COUNTER 1") -def test_read_area_inputs(client): +def test_read_area_inputs(client: Client) -> None: response = client.read_area("I0 BYTE 1") assert response == b"\x00" -def test_read_area_outputs(client): +def test_read_area_outputs(client: Client) -> None: response = client.read_area("Q0 BYTE 1") assert response == b"\x00" -def test_read_area_marker(client): +def test_read_area_marker(client: Client) -> None: response = client.read_area("M4 REAL 2") assert response == b"B\xf6\xe6fD\x07\xcdq" diff --git a/tests/real_plc_s1200/test_read_szl.py b/tests/real_plc_s1200/test_read_szl.py index 432614f..56c31e4 100644 --- a/tests/real_plc_s1200/test_read_szl.py +++ b/tests/real_plc_s1200/test_read_szl.py @@ -3,7 +3,7 @@ from python_s7comm.sync_client import Client -def test_read_szl_list(client: Client): +def test_read_szl_list(client: Client) -> None: expected = SZLResponseData( szl_id=0, szl_index=0, @@ -16,13 +16,13 @@ def test_read_szl_list(client: Client): assert response.szl_data_tree_list == expected.szl_data_tree_list -def test_read_cpu_state(client: Client): +def test_read_cpu_state(client: Client) -> None: expected = enums.CPUStatus.RUN response = client.get_cpu_state() assert response == expected.value -def test_read_cpu_order_code(client: Client): +def test_read_cpu_order_code(client: Client) -> None: expected = "6ES7 214-1HG40-0XB0 " response = client.get_order_code() assert response == expected diff --git a/tests/real_plc_s1200/test_write_area.py b/tests/real_plc_s1200/test_write_area.py index 6c4dc49..5d07839 100644 --- a/tests/real_plc_s1200/test_write_area.py +++ b/tests/real_plc_s1200/test_write_area.py @@ -5,88 +5,97 @@ from python_s7comm.sync_client import Client -def test_write_area_int(client: Client): +def test_write_area_int(client: Client) -> None: current_value = client.read_area("DB2.0 INT 2") a, b = struct.unpack("!2H", current_value) expected = struct.pack("!2H", a + 1, b + 1) - response = client.write_area("DB2.0 INT 2", data=expected) + client.write_area("DB2.0 INT 2", data=expected) response = client.read_area("DB2.0 INT 2") assert response == expected -def test_write_area_bytes(client): +def test_write_area_bytes(client: Client) -> None: current_value = client.read_area("DB2.12 BYTE 2") a, b = struct.unpack("!BB", current_value) expected = struct.pack("!BB", a + 1, b + 1) - response = client.write_area("DB2.12 BYTE 2", data=expected) + client.write_area("DB2.12 BYTE 2", data=expected) response = client.read_area("DB2.12 BYTE 2") assert response == expected -def test_write_area_real(client): +def test_write_area_real(client: Client) -> None: current_value = client.read_area("DB2.4 REAL 2") a, b = struct.unpack("!ff", current_value) expected = struct.pack("!ff", a + 1, b + 1) - response = client.write_area("DB2.4 REAL 2", data=expected) + client.write_area("DB2.4 REAL 2", data=expected) response = client.read_area("DB2.4 REAL 2") assert response == expected -def test_write_word(client): +def test_write_word(client: Client) -> None: current_value = client.read_area("DB2.14 WORD 2") a, b = struct.unpack("!HH", current_value) expected = struct.pack("!HH", a + 1, b + 1) - response = client.write_area("DB2.14 WORD 2", data=expected) + client.write_area("DB2.14 WORD 2", data=expected) response = client.read_area("DB2.14 WORD 2") assert response == expected -def test_write_dword(client): +def test_write_dword(client: Client) -> None: current_value = client.read_area("DB2.18 DWORD 2") a, b = struct.unpack("!II", current_value) expected = struct.pack("!II", a + 1, b + 1) - response = client.write_area("DB2.18 DWORD 2", data=expected) + client.write_area("DB2.18 DWORD 2", data=expected) response = client.read_area("DB2.18 DWORD 2") assert response == expected -def test_write_dint(client): +def test_write_dint(client: Client) -> None: current_value = client.read_area("DB2.26 DINT 2") a, b = struct.unpack("!II", current_value) expected = struct.pack("!II", a + 1, b + 1) - response = client.write_area("DB2.26 DINT 2", data=expected) + client.write_area("DB2.26 DINT 2", data=expected) response = client.read_area("DB2.26 DINT 2") assert response == expected -def test_write_char(client): +def test_write_char(client: Client) -> None: current_value = client.read_area("DB2.34 CHAR 2") a, b = struct.unpack("!BB", current_value) expected = struct.pack("!BB", a + 1, b + 1) - response = client.write_area("DB2.34 CHAR 2", data=expected) + client.write_area("DB2.34 CHAR 2", data=expected) response = client.read_area("DB2.34 CHAR 2") assert response == expected @pytest.mark.parametrize("start_bit", [0, 1, 2, 3, 4, 5, 6, 7]) -def test_write_bool(client, start_bit): +def test_write_bool(client: Client, start_bit: int) -> None: current_value = client.read_area(f"DB2.36.{start_bit} BOOL 1") a = struct.unpack("!?", current_value)[0] expected = struct.pack("!?", not a) - response = client.write_area(f"DB2.36.{start_bit} BOOL 1", data=expected) + client.write_area(f"DB2.36.{start_bit} BOOL 1", data=expected) response = client.read_area(f"DB2.36.{start_bit} BOOL 1") assert response == expected -def test_write_multi_vars(client): +def test_write_multi_vars(client: Client) -> None: + addresses: list[str] = [ + "DB2.4 REAL 2", + "DB2.12 BYTE 2", + "DB2.14 WORD 2", + "DB2.18 DWORD 2", + "DB2.26 DINT 2", + "DB2.34 CHAR 2", + "DB2.36 BOOL 1", + ] expected = [ b"B\xf6\xe6fD\x07\xcdq", b"\x0f\xf0", @@ -96,25 +105,8 @@ def test_write_multi_vars(client): b"F-", b"\x00", ] - items = [ - ("DB2.4 REAL 2", expected[0]), - ("DB2.12 BYTE 2", expected[1]), - ("DB2.14 WORD 2", expected[2]), - ("DB2.18 DWORD 2", expected[3]), - ("DB2.26 DINT 2", expected[4]), - ("DB2.34 CHAR 2", expected[5]), - ("DB2.36 BOOL 1", expected[6]), - ] + items = list(zip(addresses, expected)) response = client.write_multi_vars(items=items) assert response - items = [ - "DB2.4 REAL 2", - "DB2.12 BYTE 2", - "DB2.14 WORD 2", - "DB2.18 DWORD 2", - "DB2.26 DINT 2", - "DB2.34 CHAR 2", - "DB2.36 BOOL 1", - ] - assert client.read_multi_vars(items=items) == expected + assert client.read_multi_vars(items=addresses) == expected diff --git a/tests/s1200/test_plc_stop.py b/tests/s1200/test_plc_stop.py index fd45437..21370bb 100644 --- a/tests/s1200/test_plc_stop.py +++ b/tests/s1200/test_plc_stop.py @@ -4,7 +4,7 @@ from python_s7comm.s7comm.transport.cotp.tpkt import TPKT -def test_create_plc_stop_packet(): +def test_create_plc_stop_packet() -> None: s1200_request = ( b"\x03\x00\x00\x21" + b"\x02\xf0\x80" diff --git a/tests/s1200/test_read_area_pdu.py b/tests/s1200/test_read_area_pdu.py index 907f948..9909db1 100644 --- a/tests/s1200/test_read_area_pdu.py +++ b/tests/s1200/test_read_area_pdu.py @@ -1,10 +1,10 @@ from python_s7comm.s7comm import enums +from python_s7comm.s7comm.packets import DataItem, VariableReadRequest from python_s7comm.s7comm.packets.parser import S7PacketParser -from python_s7comm.s7comm.packets.rw_variable import DataItem, VariableReadRequest from python_s7comm.s7comm.packets.variable_address import VariableAddress -def test_read_area_real_pdu_request(): +def test_read_area_real_pdu_request() -> None: # tpkt_request = b"\x03\x00\x00\x1f" # cotp_request = b"\x02\xf0\x80" # s7comm_request = b"\x32\x01\x00\x00\x00\x00\x00\x0e\x00\x00\x04\x01\x12\x0a\x10\x08\x00\x02\x00\x01\x84\x00\x00\x20" @@ -21,14 +21,16 @@ def test_read_area_real_pdu_request(): assert expected == result.serialize() -def test_read_area_real_pdu_response(): +def test_read_area_real_pdu_response() -> None: # tpkt_response = b"\x03\x00\x00\x21" # cotp_response = b"\x02\xf0\x80" s7comm_response = ( b"\x32\x03\x00\x00\x01\x00\x00\x02\x00\x0c\x00\x00\x04\x01\xff\x07\00\x08\x42\xf6\xe6\x66\x44\x07\xcd\x71" ) response = S7PacketParser.parse(s7comm_response) - expected = DataItem(return_code=0xFF, transport_size=7, data_length=8, data=b"B\xf6\xe6fD\x07\xcdq") + expected = DataItem( + return_code=0xFF, transport_size=enums.DataTransportSize.REAL, data_length=8, data=b"B\xf6\xe6fD\x07\xcdq" + ) result = response.data[0] assert expected.return_code == result.return_code assert expected.transport_size == result.transport_size diff --git a/tests/s1200/test_setup_communication.py b/tests/s1200/test_setup_communication.py index 42e21b9..9e92283 100644 --- a/tests/s1200/test_setup_communication.py +++ b/tests/s1200/test_setup_communication.py @@ -3,7 +3,7 @@ from python_s7comm.s7comm.transport.cotp.tpkt import TPKT -def test_create_setup_communication_packet(): +def test_create_setup_communication_packet() -> None: s1200_request = ( b"\x03\x00\x00\x19" + b"\x02\xf0\x80" @@ -16,11 +16,3 @@ def test_create_setup_communication_packet(): cotp_packet = COTP().create_data_packet(tpdu_nr=0, payload=tpdu_packet).serialize() tpkt_packet = TPKT.serialize(payload=cotp_packet) assert tpkt_packet == s1200_request - - -# def test_create_setup_communication_packet(): -# s1200_response = ( -# b"\x03\x00\x00\x1b" -# + b"\x02\xf0\x80" -# + b"\x32\x03\x00\x00\x00\x00\x00\x08\x00\x00\x00\x00\xf0\x00\x00\x01\x00\x01\x00\xf0" -# ) diff --git a/tests/test_addresses.py b/tests/test_addresses.py index dc400a0..c19c25b 100644 --- a/tests/test_addresses.py +++ b/tests/test_addresses.py @@ -53,38 +53,38 @@ ] -@pytest.mark.parametrize("address_str, parameters_dict", valid_address_cases) # type: ignore[misc] +@pytest.mark.parametrize("address_str, parameters_dict", valid_address_cases) def test_from_string_valid_address(address_str: str, parameters_dict: dict[Any, Any]) -> None: result = VariableAddress.from_string(address_str) expected = VariableAddress(**parameters_dict) assert result == expected -@pytest.mark.parametrize("address_str", invalid_address_format_cases) # type: ignore[misc] +@pytest.mark.parametrize("address_str", invalid_address_format_cases) def test_from_string_invalid_address_format(address_str: str) -> None: with pytest.raises(ValueError): VariableAddress.from_string(address_str) -@pytest.mark.parametrize("address_str", invalid_db_number_cases) # type: ignore[misc] +@pytest.mark.parametrize("address_str", invalid_db_number_cases) def test_from_string_invalid_db_number(address_str: str) -> None: with pytest.raises(ValueError): VariableAddress.from_string(address_str) -@pytest.mark.parametrize("address_str", invalid_start_address_cases) # type: ignore[misc] +@pytest.mark.parametrize("address_str", invalid_start_address_cases) def test_from_string_invalid_start_address(address_str: str) -> None: with pytest.raises(ValueError): VariableAddress.from_string(address_str) -@pytest.mark.parametrize("address_str", invalid_start_bit_cases) # type: ignore[misc] +@pytest.mark.parametrize("address_str", invalid_start_bit_cases) def test_from_string_invalid_start_bit(address_str: str) -> None: with pytest.raises(ValueError): VariableAddress.from_string(address_str) -@pytest.mark.parametrize("address_str", invalid_amount_cases) # type: ignore[misc] +@pytest.mark.parametrize("address_str", invalid_amount_cases) def test_from_string_invalid_amount(address_str: str) -> None: with pytest.raises(ValueError): VariableAddress.from_string(address_str) diff --git a/tests/test_parser.py b/tests/test_parser.py index 7e49456..3e2675d 100644 --- a/tests/test_parser.py +++ b/tests/test_parser.py @@ -81,4 +81,5 @@ def test_parse_message_type(self, message_type_byte: int, expected_type: Message packet = b"\x32\x03\x00\x00\x00\x00\x00\x08\x00\x00\x00\x00\xf0\x00\x00\x01\x00\x01\x00\xf0" response = S7PacketParser.parse(packet) + assert response.header is not None assert response.header.message_type == expected_type diff --git a/tests/test_s7comm_send.py b/tests/test_s7comm_send.py index a672e58..701f380 100644 --- a/tests/test_s7comm_send.py +++ b/tests/test_s7comm_send.py @@ -3,8 +3,7 @@ import pytest -from python_s7comm.s7comm import S7Comm -from python_s7comm.s7comm.client import PacketLostError, StalePacketError +from python_s7comm.s7comm import PacketLostError, S7Comm, StalePacketError from python_s7comm.s7comm.enums import SubfunctionCode, UserdataFunction from python_s7comm.s7comm.packets import ( SetupCommunicationParameter,