From 2c2040437e677f87e18d88399d7bb747967253d7 Mon Sep 17 00:00:00 2001 From: Vadim Kraus <38394456+VadimKraus@users.noreply.github.com> Date: Sun, 3 Mar 2024 21:17:36 +0100 Subject: [PATCH] raise validation error for all zero response data (#68) --- solax/response_parser.py | 30 ++++++++++++++++++++++++------ solax/utils.py | 17 ++++++++++++++++- tests/conftest.py | 1 + tests/fixtures.py | 25 +++++++++++++++++++++++++ tests/test_smoke.py | 15 +++++++++++++++ tests/test_vol.py | 22 +++++++++++++++++++++- 6 files changed, 102 insertions(+), 8 deletions(-) diff --git a/solax/response_parser.py b/solax/response_parser.py index 5d6cd85..3715ee0 100644 --- a/solax/response_parser.py +++ b/solax/response_parser.py @@ -8,13 +8,30 @@ from voluptuous.humanize import humanize_error from solax.units import SensorUnit -from solax.utils import PackerBuilderResult +from solax.utils import PackerBuilderResult, contains_none_zero_value _LOGGER = logging.getLogger(__name__) _LOGGER.setLevel(logging.INFO) InverterResponse = namedtuple("InverterResponse", "data, serial_number, version, type") +_KEY_DATA = "Data" +_KEY_SERIAL = "SN" +_KEY_VERSION = "version" +_KEY_VER = "ver" +_KEY_TYPE = "type" + + +GenericResponseSchema = vol.Schema( + { + vol.Required(_KEY_DATA): vol.Schema(contains_none_zero_value), + vol.Required(vol.Or(_KEY_SERIAL, _KEY_SERIAL.lower())): vol.All(), + vol.Required(vol.Or(_KEY_VERSION, _KEY_VER)): vol.All(), + vol.Required(_KEY_TYPE): vol.All(), + }, + extra=vol.REMOVE_EXTRA, +) + SensorIndexSpec = Union[int, PackerBuilderResult] ResponseDecoder = Dict[ str, @@ -27,7 +44,8 @@ class ResponseParser: def __init__(self, schema: vol.Schema, decoder: ResponseDecoder): - self.schema = schema + self.schema = vol.And(schema, GenericResponseSchema) + self.response_decoder = decoder def _decode_map(self) -> Dict[str, SensorIndexSpec]: @@ -82,8 +100,8 @@ def handle_response(self, resp: bytearray): _ = humanize_error(json_response, ex) raise return InverterResponse( - data=self.map_response(response["Data"]), - serial_number=response.get("SN", response.get("sn")), - version=response.get("ver", response.get("version")), - type=response["type"], + data=self.map_response(response[_KEY_DATA]), + serial_number=response.get(_KEY_SERIAL, response.get(_KEY_SERIAL.lower())), + version=response.get(_KEY_VER, response.get(_KEY_VERSION)), + type=response[_KEY_TYPE], ) diff --git a/solax/utils.py b/solax/utils.py index f2a884f..de38f04 100644 --- a/solax/utils.py +++ b/solax/utils.py @@ -1,4 +1,5 @@ -from typing import Protocol, Tuple +from numbers import Number +from typing import List, Protocol, Tuple from voluptuous import Invalid @@ -88,3 +89,17 @@ def twoway_div100(val): def to_url(host, port): return f"http://{host}:{port}/" + + +def contains_none_zero_value(value: List[Number]): + """Validate that at least one element is not zero. + Args: + value (List[Number]): list to validate + Raises: + Invalid: if all elements are zero + """ + + if isinstance(value, list): + if len(value) != 0 and any((v != 0 for v in value)): + return value + raise Invalid("All elements in the list {actual} are zero") diff --git a/tests/conftest.py b/tests/conftest.py index b256dfa..7057cae 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -1,5 +1,6 @@ # pylint: disable=unused-import from tests.fixtures import inverters_fixture # noqa: F401 +from tests.fixtures import inverters_fixture_all_zero # noqa: F401 from tests.fixtures import inverters_garbage_fixture # noqa: F401 from tests.fixtures import inverters_under_test # noqa: F401 from tests.fixtures import simple_http_fixture # noqa: F401 diff --git a/tests/fixtures.py b/tests/fixtures.py index 3983bfb..70592bc 100644 --- a/tests/fixtures.py +++ b/tests/fixtures.py @@ -1,4 +1,5 @@ from collections import namedtuple +from copy import copy import pytest @@ -294,3 +295,27 @@ def inverters_garbage_fixture(httpserver, request): query_string=request.param.query_string, ).respond_with_json({"bingo": "bango"}) yield ((httpserver.host, httpserver.port), request.param.inverter) + + +@pytest.fixture(params=INVERTERS_UNDER_TEST) +def inverters_fixture_all_zero(httpserver, request): + """Use defined responses but replace the data with all zero values. + Testing incorrect responses. + """ + + response = request.param.response + response = copy(response) + response["Data"] = [0] * (len(response["Data"])) + + httpserver.expect_request( + uri=request.param.uri, + method=request.param.method, + query_string=request.param.query_string, + headers=request.param.headers, + data=request.param.data, + ).respond_with_json(response) + yield ( + (httpserver.host, httpserver.port), + request.param.inverter, + request.param.values, + ) diff --git a/tests/test_smoke.py b/tests/test_smoke.py index a3b3c0e..d26b00f 100644 --- a/tests/test_smoke.py +++ b/tests/test_smoke.py @@ -66,3 +66,18 @@ def test_inverter_sensors_define_valid_units(inverters_under_test): f"is not a proper Unit on sensor '{name}' of Inverter '{inverters_under_test}'" ) assert isinstance(unit, Measurement), msg + + +@pytest.mark.asyncio +async def test_smoke_zero(inverters_fixture_all_zero): + """Responses with all zero values should be treated as an error. + Args: + inverters_fixture_all_zero (_type_): all responses with zero value data + """ + conn, inverter_class, _ = inverters_fixture_all_zero + + # msg = 'all zero values should be discarded' + with pytest.raises(InverterError): + inv = await build_right_variant(inverter_class, conn) + rt_api = solax.RealTimeAPI(inv) + await rt_api.get_data() diff --git a/tests/test_vol.py b/tests/test_vol.py index 3d0d380..40231dc 100644 --- a/tests/test_vol.py +++ b/tests/test_vol.py @@ -1,7 +1,7 @@ import pytest from voluptuous import Invalid -from solax.utils import startswith +from solax.utils import contains_none_zero_value, startswith def test_does_start_with(): @@ -23,3 +23,23 @@ def test_is_not_str(): actual = 1 with pytest.raises(Invalid): startswith(expected)(actual) + + +def test_contains_none_zero_value(): + with pytest.raises(Invalid): + contains_none_zero_value([0]) + + with pytest.raises(Invalid): + contains_none_zero_value([0, 0]) + + not_a_list = 1 + with pytest.raises(Invalid): + contains_none_zero_value(not_a_list) + + expected = [1, 0] + actual = contains_none_zero_value(expected) + assert actual == expected + + expected = [-1, 1] + actual = contains_none_zero_value(expected) + assert actual == expected