Sonar qube 6 (#174)

* test class ModbusConn

app/src/async_stream.py

@@ -3,10 +3,15 @@ import logging
 import traceback
 import time
 from asyncio import StreamReader, StreamWriter
-from messages import hex_dump_memory, State
 from typing import Self
 from itertools import count
 
+if __name__ == "app.src.async_stream":
+    from app.src.messages import hex_dump_memory, State
+else:  # pragma: no cover
+    from messages import hex_dump_memory, State
+
+
 import gc
 logger = logging.getLogger('conn')
 

app/src/gen3plus/connection_g3p.py

@@ -1,7 +1,12 @@
 import logging
 from asyncio import StreamReader, StreamWriter
-from async_stream import AsyncStream
+
-from gen3plus.solarman_v5 import SolarmanV5
+if __name__ == "app.src.gen3plus.connection_g3p":
+    from app.src.async_stream import AsyncStream
+    from app.src.gen3plus.solarman_v5 import SolarmanV5
+else:  # pragma: no cover
+    from async_stream import AsyncStream
+    from gen3plus.solarman_v5 import SolarmanV5
 
 logger = logging.getLogger('conn')
 

app/src/gen3plus/inverter_g3p.py

@@ -3,11 +3,18 @@ import traceback
 import json
 import asyncio
 from asyncio import StreamReader, StreamWriter
-from config import Config
-from inverter import Inverter
-from gen3plus.connection_g3p import ConnectionG3P
 from aiomqtt import MqttCodeError
-from infos import Infos
+
+if __name__ == "app.src.gen3plus.inverter_g3p":
+    from app.src.config import Config
+    from app.src.inverter import Inverter
+    from app.src.gen3plus.connection_g3p import ConnectionG3P
+    from app.src.infos import Infos
+else:  # pragma: no cover
+    from config import Config
+    from inverter import Inverter
+    from gen3plus.connection_g3p import ConnectionG3P
+    from infos import Infos
 
 
 logger_mqtt = logging.getLogger('mqtt')

app/src/inverter.py

@@ -28,7 +28,6 @@ class Inverter():
         cls.proxy_unique_id = ha['proxy_unique_id']
 
         # call Mqtt singleton to establisch the connection to the mqtt broker
-        print('call Mqtt.init')
         cls.mqtt = Mqtt(cls._cb_mqtt_is_up)
 
         # register all counters which should be reset at midnight.

app/src/modbus_tcp.py

@@ -1,9 +1,13 @@
 import logging
 import traceback
 import asyncio
-from config import Config
 
-from gen3plus.inverter_g3p import InverterG3P
+if __name__ == "app.src.modbus_tcp":
+    from app.src.config import Config
+    from app.src.gen3plus.inverter_g3p import InverterG3P
+else:  # pragma: no cover
+    from config import Config
+    from gen3plus.inverter_g3p import InverterG3P
 
 logger = logging.getLogger('conn')
 

app/tests/test_modbus_tcp.py

@@ -0,0 +1,78 @@
+# test_with_pytest.py
+import pytest
+import asyncio
+
+from mock import patch
+from app.src.singleton import Singleton
+from app.src.config import Config
+from app.src.infos import Infos
+from app.src.modbus_tcp import ModbusConn
+
+
+pytest_plugins = ('pytest_asyncio',)
+
+# initialize the proxy statistics
+Infos.static_init()
+
+@pytest.fixture(scope="module", autouse=True)
+def module_init():
+    Singleton._instances.clear()
+    yield
+
+@pytest.fixture(scope="module")
+def test_port():
+    return 1883
+
+@pytest.fixture(scope="module")
+def test_hostname():
+    # if getenv("GITHUB_ACTIONS") == "true":
+    #     return 'mqtt'
+    # else:
+        return 'test.mosquitto.org'
+
+@pytest.fixture
+def config_mqtt_conn(test_hostname, test_port):
+    Config.act_config = {'mqtt':{'host': test_hostname, 'port': test_port, 'user': '', 'passwd': ''},
+                         'ha':{'auto_conf_prefix': 'homeassistant','discovery_prefix': 'homeassistant', 'entity_prefix': 'tsun'}
+                        }
+
+@pytest.fixture
+def config_no_conn(test_port):
+    Config.act_config = {'mqtt':{'host': "", 'port': test_port, 'user': '', 'passwd': ''},
+                         'ha':{'auto_conf_prefix': 'homeassistant','discovery_prefix': 'homeassistant', 'entity_prefix': 'tsun'}
+                        }
+
+class FakeReader():
+    pass
+
+
+class FakeWriter():
+    pass
+
+
+@pytest.fixture
+def patch_open():
+    async def new_conn(conn):
+        await asyncio.sleep(0.01)
+        return FakeReader(), FakeWriter()
+    
+    def new_open(host: str, port: int):
+        return new_conn(None)
+
+    with patch.object(asyncio, 'open_connection', new_open) as conn:
+        yield conn
+
+
+@pytest.mark.asyncio
+async def test_modbus_conn(patch_open):
+    _ = patch_open
+    assert Infos.stat['proxy']['Inverter_Cnt'] == 0
+
+    async with ModbusConn('test.local', 1234) as stream:
+        assert stream.node_id == 'G3P'
+        assert stream.addr == ('test.local', 1234)
+        assert type(stream.reader) is FakeReader
+        assert type(stream.writer) is FakeWriter
+        assert Infos.stat['proxy']['Inverter_Cnt'] == 1
+    
+    assert Infos.stat['proxy']['Inverter_Cnt'] == 0