From 7134aeb1c938be132a0089a55ce4829d7c91e0ea Mon Sep 17 00:00:00 2001 From: Guzz-T Date: Tue, 3 Mar 2026 20:40:58 +0100 Subject: [PATCH] Increase the test-coverage --- tests/fake/fake_socket.py | 11 +++++++- tests/shi/test_shi_contiguous.py | 10 +++++++ tests/shi/test_shi_modbus.py | 1 - tests/shi/test_shi_vector.py | 6 ++++ tests/test_collections.py | 7 +++-- tests/test_datatypes.py | 12 ++++++++ tests/test_definitions.py | 1 + tests/test_socket_interaction.py | 48 ++++++++++++++++++++++++++++++-- 8 files changed, 90 insertions(+), 6 deletions(-) diff --git a/tests/fake/fake_socket.py b/tests/fake/fake_socket.py index 56d926cc..be3979fc 100644 --- a/tests/fake/fake_socket.py +++ b/tests/fake/fake_socket.py @@ -19,6 +19,8 @@ def fake_visibility_value(i): class FakeSocket: last_instance = None prev_instance = None + create_connection_exception = None + force_recv_result = None # These code are hard coded here in order to prevent # accidential changes in constants.py @@ -125,6 +127,9 @@ def sendall(self, data): def recv(self, cnt, flag=0): assert self._connected + if FakeSocket.force_recv_result is not None: + return FakeSocket.force_recv_result + if (not self._blocking) and len(self._buffer) < cnt: raise BlockingIOError("Not enough bytes in buffer.") @@ -147,5 +152,9 @@ def __exit__(self, exc_type, exc, tb): self.close() return False + def fake_create_connection(info): - return FakeSocket() \ No newline at end of file + if FakeSocket.create_connection_exception is not None: + raise FakeSocket.create_connection_exception + else: + return FakeSocket() \ No newline at end of file diff --git a/tests/shi/test_shi_contiguous.py b/tests/shi/test_shi_contiguous.py index 62a27d50..9dc7a72d 100644 --- a/tests/shi/test_shi_contiguous.py +++ b/tests/shi/test_shi_contiguous.py @@ -278,6 +278,16 @@ def test_get_data(self): data_arr = block.get_data_arr() assert data_arr is None + # Missing data (via gaps) + block = ContiguousDataBlock() + field_a.raw = [56, 57] + field_c.raw = [21, 22, 23] + block.add(def_a, field_a) + block.add(def_c, field_c) + + data_arr = block.get_data_arr() + assert data_arr is None + def test_repr(self): block = ContiguousDataBlock() text_empty = repr(block) diff --git a/tests/shi/test_shi_modbus.py b/tests/shi/test_shi_modbus.py index ba432418..1d0145ee 100644 --- a/tests/shi/test_shi_modbus.py +++ b/tests/shi/test_shi_modbus.py @@ -95,7 +95,6 @@ def test_no_connection(self): # Cannot connect to read holdings data = LuxtronikSmartHomeReadHoldingsTelegram(0, 1) result = self.modbus_interface.send(data) - print(data.data) assert not result # Cannot connect to write holdings diff --git a/tests/shi/test_shi_vector.py b/tests/shi/test_shi_vector.py index 90e4f59e..8537f965 100644 --- a/tests/shi/test_shi_vector.py +++ b/tests/shi/test_shi_vector.py @@ -88,6 +88,12 @@ def test_create(self): field = DataVectorTest.create_any_field('BAR') assert field is None + # create field by def + field = DataVectorTest.create_any_field(TEST_DEFINITIONS._definitions[2]) + assert field.name == 'field_9a' + assert field.writeable + assert type(field) is Base + # create versioned data vector data_vector = DataVectorTest(parse_version("1.2")) assert data_vector.version == (1, 2, 0, 0) diff --git a/tests/test_collections.py b/tests/test_collections.py index 39a3688a..21392835 100644 --- a/tests/test_collections.py +++ b/tests/test_collections.py @@ -226,10 +226,13 @@ def test_init(self): d = LuxtronikFieldsDictionary() assert type(d._def_lookup) is LuxtronikDefinitionsDictionary + assert d.def_dict is d._def_lookup assert type(d._field_lookup) is dict - assert len(d._field_lookup.values()) == 0 + assert d.field_dict is d._field_lookup assert type(d._pairs) is list - assert len(d._pairs) == 0 + assert d.pairs is d._pairs + assert len(d.field_dict) == 0 + assert len(d.pairs) == 0 def test_add(self): d = LuxtronikFieldsDictionary() diff --git a/tests/test_datatypes.py b/tests/test_datatypes.py index 81038f7d..4bc2c3ec 100644 --- a/tests/test_datatypes.py +++ b/tests/test_datatypes.py @@ -77,6 +77,7 @@ def test_init(self): a = Base("base") assert a.name == "base" assert a.writeable is False + assert a.unit is None b = Base("base", writeable=True) assert b.name == "base" @@ -174,6 +175,8 @@ def test_eq(self): d = Bool("bool") assert c != d + assert a != "b" + def test_lt(self): """Test cases for __lt__ function""" @@ -198,6 +201,7 @@ def test_init(self): assert a.name == "selection_base" assert not a.codes assert len(a.codes) == 0 + assert a.unit is None def test_options(self): """Test cases for options property""" @@ -246,6 +250,7 @@ def test_init(self): assert a.name == "selection_base_child" assert a.codes assert len(a.codes) == 3 + assert a.unit is None def test_options(self): """Test cases for options property""" @@ -287,6 +292,7 @@ def test_init(self): assert a.name == "bitmask_base" assert not a.bit_values assert len(a.bit_values) == 0 + assert a.unit is None def test_bits(self): """Test cases for bits property""" @@ -483,6 +489,7 @@ def test_init(self): assert a.name == "celsius" assert a.datatype_class == "temperature" assert a.datatype_unit == "°C" + assert a.unit == "°C" def test_from_heatpump(self): """Test cases for from_heatpump function""" @@ -515,6 +522,7 @@ def test_init(self): assert a.name == "bool" assert a.datatype_class == "boolean" assert a.datatype_unit is None + assert a.unit is None def test_from_heatpump(self): """Test cases for from_heatpump function""" @@ -539,6 +547,7 @@ def test_init(self): assert a.name == "frequency" assert a.datatype_class == "frequency" assert a.datatype_unit == "Hz" + assert a.unit == "Hz" class TestSeconds: @@ -551,6 +560,7 @@ def test_init(self): assert a.name == "seconds" assert a.datatype_class == "timespan" assert a.datatype_unit == "s" + assert a.unit == "s" class TestIPv4Address: @@ -563,6 +573,7 @@ def test_init(self): assert a.name == "ipv4_address" assert a.datatype_class == "ipv4_address" assert a.datatype_unit is None + assert a.unit is None def test_from_heatpump(self): """Test cases for from_heatpump function""" @@ -593,6 +604,7 @@ def test_init(self): assert a.name == "timestamp" assert a.datatype_class == "timestamp" assert a.datatype_unit is None + assert a.unit is None def test_from_heatpump(self): """Test cases for from_heatpump function""" diff --git a/tests/test_definitions.py b/tests/test_definitions.py index bcf3d79e..43a16f22 100644 --- a/tests/test_definitions.py +++ b/tests/test_definitions.py @@ -36,6 +36,7 @@ def test_init(self): assert definition.writeable == self.TEST_DATA['writeable'] assert definition.names == names assert definition.name == names[0] + assert definition.data_type == 'INT16' assert definition.valid assert definition assert definition.type_name == 'bar' diff --git a/tests/test_socket_interaction.py b/tests/test_socket_interaction.py index 34e8ebdf..4a5227e4 100644 --- a/tests/test_socket_interaction.py +++ b/tests/test_socket_interaction.py @@ -1,6 +1,7 @@ """Test suite for the socket interaction of LuxtronikSocketInterface and Luxtronik""" import unittest.mock as mock +import socket from luxtronik import Luxtronik, LuxtronikSocketInterface, Parameters, Calculations, Visibilities from luxtronik.collections import integrate_data @@ -68,7 +69,7 @@ def test_luxtronik_socket_interface(self): self.clear_data_vector(p) assert not self.check_data_vector(p) - # Read parameters + # Read calculations c = lux.read_calculations() s = FakeSocket.last_instance assert type(c) is Calculations @@ -78,7 +79,7 @@ def test_luxtronik_socket_interface(self): self.clear_data_vector(c) assert not self.check_data_vector(c) - # Read parameters + # Read visibilities v = lux.read_visibilities() s = FakeSocket.last_instance assert type(v) is Visibilities @@ -121,6 +122,14 @@ def test_luxtronik_socket_interface(self): assert not p[4].write_pending assert self.check_luxtronik_data(d) + # erroneous read + FakeSocket.force_recv_result = b'' + + p = lux.read_parameters() + assert p is None + + FakeSocket.force_recv_result = None + def test_luxtronik(self): host = "my_heatpump" port = 4711 @@ -189,3 +198,38 @@ def test_luxtronik(self): # Now, the values should be read assert self.check_luxtronik_data(lux) + + def test_connect(self): + host = "my_heatpump" + port = 4711 + lux = LuxtronikSocketInterface(host, port) + + p = lux.read_parameters() + assert p is not None + + FakeSocket.create_connection_exception = socket.gaierror + + p = lux.read_parameters() + assert p is None + + FakeSocket.create_connection_exception = socket.timeout + + p = lux.read_parameters() + assert p is None + + FakeSocket.create_connection_exception = ConnectionRefusedError + + p = lux.read_parameters() + assert p is None + + FakeSocket.create_connection_exception = OSError + + p = lux.read_parameters() + assert p is None + + FakeSocket.create_connection_exception = ValueError + + p = lux.read_parameters() + assert p is None + + FakeSocket.create_connection_exception = None