Added additional tests

Author: imbeacon

tb_mqtt_client/common/config_loader.py

@@ -24,6 +24,17 @@ class DeviceConfig:
     """
 
     def __init__(self, config=None):
+        self.host = None
+        self.port = 1883
+        self.access_token: Optional[str] = None
+        self.username: Optional[str] = None
+        self.password: Optional[str] = None
+        self.client_id: Optional[str] = None
+        self.ca_cert: Optional[str] = None
+        self.client_cert: Optional[str] = None
+        self.private_key: Optional[str] = None
+        self.qos: int = 1
+
         if config is not None:
             self.host: str = config.get("host", "localhost")
             self.port: int = config.get("port", 1883)
@@ -35,24 +46,24 @@ def __init__(self, config=None):
             self.client_cert: Optional[str] = config.get("client_cert")
             self.private_key: Optional[str] = config.get("private_key")
 
-        self.host: str = os.getenv("TB_HOST")
-        self.port: int = int(os.getenv("TB_PORT", 1883))
+        self.host: str = os.getenv("TB_HOST", self.host)
+        self.port: int = int(os.getenv("TB_PORT", self.port))
 
         # Authentication options
-        self.access_token: Optional[str] = os.getenv("TB_ACCESS_TOKEN")
-        self.username: Optional[str] = os.getenv("TB_USERNAME")
-        self.password: Optional[str] = os.getenv("TB_PASSWORD")
+        self.access_token: Optional[str] = os.getenv("TB_ACCESS_TOKEN", self.access_token)
+        self.username: Optional[str] = os.getenv("TB_USERNAME", self.username)
+        self.password: Optional[str] = os.getenv("TB_PASSWORD", self.password)
 
         # Optional
-        self.client_id: Optional[str] = os.getenv("TB_CLIENT_ID")
+        self.client_id: Optional[str] = os.getenv("TB_CLIENT_ID", self.client_id)
 
         # TLS options
-        self.ca_cert: Optional[str] = os.getenv("TB_CA_CERT")
-        self.client_cert: Optional[str] = os.getenv("TB_CLIENT_CERT")
-        self.private_key: Optional[str] = os.getenv("TB_PRIVATE_KEY")
+        self.ca_cert: Optional[str] = os.getenv("TB_CA_CERT", self.ca_cert)
+        self.client_cert: Optional[str] = os.getenv("TB_CLIENT_CERT", self.client_cert)
+        self.private_key: Optional[str] = os.getenv("TB_PRIVATE_KEY", self.private_key)
 
         # Default values
-        self.qos: int = int(os.getenv("TB_QOS", 1))
+        self.qos: int = int(os.getenv("TB_QOS", self.qos))
 
     def use_tls_auth(self) -> bool:
         return all([self.ca_cert, self.client_cert, self.private_key])

tb_mqtt_client/common/gmqtt_patch.py

@@ -122,8 +122,8 @@ def parse_mqtt_properties(packet: bytes) -> dict:
         properties_dict = defaultdict(list)
 
         try:
-            properties_len, _ = unpack_variable_byte_integer(packet)
-            props = packet[:properties_len]
+            properties_len, rest = unpack_variable_byte_integer(packet)
+            props = rest[:properties_len]  # slice out exactly the properties section
 
             while props:
                 property_identifier = props[0]

tb_mqtt_client/common/rate_limit/backpressure_controller.py

@@ -30,7 +30,6 @@ def __init__(self, main_stop_event: Event):
         self._consecutive_quota_exceeded = 0
         self._last_quota_exceeded = datetime.now(UTC)
         self._max_backoff_seconds = 3600  # 1 hour
-        self._can_process_messages_events: List[asyncio.Event] = []
         logger.debug("BackpressureController initialized with default pause duration of %s seconds",
                      self._default_pause_duration.total_seconds())
 
@@ -90,25 +89,10 @@ def should_pause(self) -> bool:
         # Reset pause state
         self._pause_until = None
         logger.info("Backpressure released, resuming publishing")
-        for event in self._can_process_messages_events:
-            if not event.is_set():
-                event.set()
-                logger.debug("Set can-process event %s", event)
         return False
 
     def clear(self):
         if self._pause_until is not None:
             logger.info("Clearing backpressure pause")
         self._pause_until = None
         self._consecutive_quota_exceeded = 0
-
-    def register_can_process_event(self, event: Event):
-        """
-        Register an event that will be set when the controller can process messages again.
-        This is useful for other components to wait until the backpressure is lifted.
-        """
-        if not isinstance(event, Event):
-            raise ValueError("Expected an asyncio.Event instance")
-        self._can_process_messages_events.append(event)
-        logger.debug("Registered a new can-process event, total events: %d",
-                     len(self._can_process_messages_events))

tests/common/test_async_utils.py

@@ -13,13 +13,36 @@
 #  limitations under the License.
 
 import asyncio
+import threading
+import time
 
 import pytest
 
-from tb_mqtt_client.common.async_utils import FutureMap, await_or_stop
+import tb_mqtt_client.common.async_utils as async_utils_mod
+from tb_mqtt_client.common.async_utils import FutureMap, await_or_stop, run_coroutine_sync
 from tb_mqtt_client.common.publish_result import PublishResult
 
 
+@pytest.fixture
+def fake_loop(monkeypatch):
+    class _FakeTask:
+        def __init__(self, thread: threading.Thread):
+            self._thread = thread
+
+        def done(self):  # optional helpers if ever needed
+            return not self._thread.is_alive()
+
+    class _FakeLoop:
+        def create_task(self, coro):
+            t = threading.Thread(target=lambda: asyncio.run(coro), daemon=True)
+            t.start()
+            return _FakeTask(t)
+
+    monkeypatch.setattr(async_utils_mod.asyncio, "get_running_loop", lambda: _FakeLoop())
+
+    return _FakeLoop()
+
+
 @pytest.mark.asyncio
 async def test_future_map_register_and_get_parents():
     fm = FutureMap()
@@ -150,5 +173,36 @@ async def coro():
     assert result is None
 
 
+def test_returns_result_when_coroutine_completes(fake_loop):
+    async def ok_coro():
+        await asyncio.sleep(0.01)
+        return "done"
+
+    result = run_coroutine_sync(lambda: ok_coro(), timeout=0.5)
+    assert result == "done"
+
+
+def test_raises_original_exception_from_coroutine(fake_loop):
+    class CustomError(RuntimeError):
+        pass
+
+    async def bad_coro():
+        await asyncio.sleep(0.01)
+        raise CustomError("boom")
+
+    with pytest.raises(CustomError, match="boom"):
+        run_coroutine_sync(lambda: bad_coro(), timeout=0.5)
+
+
+def test_timeout_raises_timeout_error(fake_loop):
+    async def slow_coro():
+        await asyncio.sleep(0.2)
+        return "too late"
+
+    with pytest.raises(TimeoutError, match=r"did not complete in 0\.05 seconds"):
+        run_coroutine_sync(lambda: slow_coro(), timeout=0.05, raise_on_timeout=True)
+    time.sleep(0.25)
+
+
 if __name__ == '__main__':
     pytest.main([__file__, "--tb=short", "-v"])

tests/common/test_config_loader.py

@@ -20,7 +20,32 @@
 
 class TestDeviceConfig(unittest.TestCase):
 
-    def loads_default_values_when_env_vars_missing(self):
+    def test_config_creation_from_dict(self):
+        config_dict = {
+            "host": "localhost",
+            "port": 1883,
+            "access_token": "test_token",
+            "username": "test_user",
+            "password": "test_pass",
+            "client_id": "test_client",
+            "ca_cert": "test_ca",
+            "client_cert": "test_cert",
+            "private_key": "test_key",
+            "qos": 1
+        }
+        config = DeviceConfig(config_dict)
+        self.assertEqual(config.host, "localhost")
+        self.assertEqual(config.port, 1883)
+        self.assertEqual(config.access_token, "test_token")
+        self.assertEqual(config.username, "test_user")
+        self.assertEqual(config.password, "test_pass")
+        self.assertEqual(config.client_id, "test_client")
+        self.assertEqual(config.ca_cert, "test_ca")
+        self.assertEqual(config.client_cert, "test_cert")
+        self.assertEqual(config.private_key, "test_key")
+        self.assertEqual(config.qos, 1)
+
+    def test_loads_default_values_when_env_vars_missing(self):
         os.environ.clear()
         config = DeviceConfig()
         self.assertEqual(config.host, None)
@@ -34,7 +59,7 @@ def loads_default_values_when_env_vars_missing(self):
         self.assertEqual(config.private_key, None)
         self.assertEqual(config.qos, 1)
 
-    def loads_values_from_env_vars(self):
+    def test_loads_values_from_env_vars(self):
         os.environ["TB_HOST"] = "test_host"
         os.environ["TB_PORT"] = "8883"
         os.environ["TB_ACCESS_TOKEN"] = "test_token"
@@ -58,22 +83,22 @@ def loads_values_from_env_vars(self):
         self.assertEqual(config.private_key, "test_key")
         self.assertEqual(config.qos, 2)
 
-    def detects_tls_auth_correctly(self):
+    def test_detects_tls_auth_correctly(self):
         os.environ["TB_CA_CERT"] = "test_ca"
         os.environ["TB_CLIENT_CERT"] = "test_cert"
         os.environ["TB_PRIVATE_KEY"] = "test_key"
         config = DeviceConfig()
         self.assertTrue(config.use_tls_auth())
 
-    def detects_tls_correctly(self):
+    def test_detects_tls_correctly(self):
         os.environ["TB_CA_CERT"] = "test_ca"
         config = DeviceConfig()
         self.assertTrue(config.use_tls())
 
 
 class TestGatewayConfig(unittest.TestCase):
 
-    def loads_gateway_specific_env_vars(self):
+    def test_loads_gateway_specific_env_vars(self):
         os.environ["TB_GW_HOST"] = "gw_host"
         os.environ["TB_GW_PORT"] = "8884"
         os.environ["TB_GW_ACCESS_TOKEN"] = "gw_token"
@@ -97,10 +122,14 @@ def loads_gateway_specific_env_vars(self):
         self.assertEqual(config.private_key, "gw_key")
         self.assertEqual(config.qos, 0)
 
-    def falls_back_to_device_config_when_gateway_env_vars_missing(self):
+    def test_falls_back_to_device_config_when_gateway_env_vars_missing(self):
         os.environ.clear()
         os.environ["TB_HOST"] = "device_host"
         os.environ["TB_PORT"] = "1884"
         config = GatewayConfig()
         self.assertEqual(config.host, "device_host")
         self.assertEqual(config.port, 1884)
+
+
+if __name__ == "__main__":
+    unittest.main()