Skip to content

Commit

Permalink
Remove list() from TCP tests, simplify more array expands (#362)
Browse files Browse the repository at this point in the history
  • Loading branch information
MarijnS95 authored Mar 29, 2023
1 parent 51d65a2 commit 265a4ac
Showing 1 changed file with 32 additions and 27 deletions.
59 changes: 32 additions & 27 deletions tests/test_tcp_models.py
Original file line number Diff line number Diff line change
Expand Up @@ -36,25 +36,29 @@ async def test_inverter_tcp_checksum_correct() -> None:
"""Ensure the implementation validates CRC before accepting data."""
serial_number = 1
# Serial number, twice
serial_number_bytes = list(serial_number.to_bytes(4, "little") * 2)
serial_number_bytes = serial_number.to_bytes(4, "little") * 2

test_message_contents = list(b"foo")
test_message = (
test_message_contents = b"foo"
test_message = [
len(test_message_contents), # Length
tcp.MESSAGE_RECV_SEP,
tcp.MESSAGE_TYPE_STRING,
*serial_number_bytes,
*test_message_contents,
)
]
false_checksum = 0xFE
checksum = sum(test_message) & 0xFF

with pytest.raises(OmnikInverterPacketInvalidError) as excinfo:
assert tcp.parse_messages(
serial_number,
bytearray(
[tcp.MESSAGE_START, *list(test_message)]
+ [false_checksum, tcp.MESSAGE_END],
[
tcp.MESSAGE_START,
*test_message,
false_checksum,
tcp.MESSAGE_END,
],
),
)

Expand All @@ -68,21 +72,21 @@ async def test_inverter_tcp_recv_sep() -> None:
"""Require RECV_SEP "separator" between length and message_type."""
serial_number = 1
# Serial number, twice
serial_number_bytes = list(serial_number.to_bytes(4, "little") * 2)
serial_number_bytes = serial_number.to_bytes(4, "little") * 2

test_message = (
test_message = [
0, # Length
123, # Invalid separator
tcp.MESSAGE_TYPE_STRING,
*serial_number_bytes,
)
]
checksum = sum(test_message) & 0xFF

with pytest.raises(OmnikInverterPacketInvalidError) as excinfo:
assert tcp.parse_messages(
serial_number,
bytearray(
[tcp.MESSAGE_START, *list(test_message)] + [checksum, tcp.MESSAGE_END],
[tcp.MESSAGE_START, *test_message, checksum, tcp.MESSAGE_END],
),
)

Expand All @@ -91,20 +95,20 @@ async def test_inverter_tcp_recv_sep() -> None:

async def test_inverter_tcp_double_serial_match() -> None:
"""Require both serial numbers in the received buffer to be identical."""
test_message = (
test_message = [
0, # Length
tcp.MESSAGE_RECV_SEP,
tcp.MESSAGE_TYPE_STRING,
*list((1).to_bytes(4, "little")),
*list((2).to_bytes(4, "little")),
)
*(1).to_bytes(4, "little"),
*(2).to_bytes(4, "little"),
]
checksum = sum(test_message) & 0xFF

with pytest.raises(OmnikInverterPacketInvalidError) as excinfo:
assert tcp.parse_messages(
3,
bytearray(
[tcp.MESSAGE_START, *list(test_message)] + [checksum, tcp.MESSAGE_END],
[tcp.MESSAGE_START, *test_message, checksum, tcp.MESSAGE_END],
),
)

Expand All @@ -115,22 +119,23 @@ async def test_inverter_tcp_end_marker() -> None:
"""Require end marker."""
serial_number = 1
# Serial number, twice
serial_number_bytes = list(serial_number.to_bytes(4, "little") * 2)
serial_number_bytes = serial_number.to_bytes(4, "little") * 2

test_message = (
test_message = [
0, # Length
tcp.MESSAGE_RECV_SEP,
tcp.MESSAGE_TYPE_STRING,
*serial_number_bytes,
)
]
checksum = sum(test_message) & 0xFF

with pytest.raises(OmnikInverterPacketInvalidError) as excinfo:
assert tcp.parse_messages(
serial_number,
bytearray(
[tcp.MESSAGE_START, *list(test_message)]
+ [
[
tcp.MESSAGE_START,
*test_message,
checksum,
123, # Invalid end byte
],
Expand All @@ -144,21 +149,21 @@ async def test_inverter_tcp_known_message_type() -> None:
"""Require message type to be known."""
serial_number = 1
# Serial number, twice
serial_number_bytes = list(serial_number.to_bytes(4, "little") * 2)
serial_number_bytes = serial_number.to_bytes(4, "little") * 2

test_message = (
test_message = [
0, # Length
tcp.MESSAGE_RECV_SEP,
0, # Unknown message type
*serial_number_bytes,
)
]
checksum = sum(test_message) & 0xFF

with pytest.raises(OmnikInverterPacketInvalidError) as excinfo:
assert tcp.parse_messages(
serial_number,
bytearray(
[tcp.MESSAGE_START, *list(test_message)] + [checksum, tcp.MESSAGE_END],
[tcp.MESSAGE_START, *test_message, checksum, tcp.MESSAGE_END],
),
)

Expand All @@ -172,14 +177,14 @@ async def test_inverter_tcp_require_information_reply() -> None:
"""Require at least one of the messages to be an information reply."""
serial_number = 1
# Serial number, twice
serial_number_bytes = list(serial_number.to_bytes(4, "little") * 2)
serial_number_bytes = serial_number.to_bytes(4, "little") * 2

test_message = (
test_message = [
0, # Length
tcp.MESSAGE_RECV_SEP,
tcp.MESSAGE_TYPE_STRING,
*serial_number_bytes,
)
]
checksum = sum(test_message) & 0xFF

with pytest.raises(OmnikInverterPacketInvalidError) as excinfo:
Expand Down

0 comments on commit 265a4ac

Please sign in to comment.