additional clean up of samples and validation

samples/client_keepalive.py

@@ -32,21 +32,8 @@ def __main__():
 
     formatter = "[%(asctime)s] :: %(levelname)s :: %(name)s :: %(message)s"
     logging.basicConfig(level=logging.INFO, format=formatter)
+    asyncio.run(main())
 
-    loop = asyncio.new_event_loop()
-    asyncio.set_event_loop(loop)
-
-    task = loop.create_task(main())
-
-    try:
-        loop.run_until_complete(task)
-    except KeyboardInterrupt:
-        logger.info("KeyboardInterrupt received. Stopping client...")
-        task.cancel()
-        loop.run_until_complete(task)  # Ensure task finishes cleanup
-    finally:
-
-        loop.close()
 
 if __name__ == "__main__":
     __main__()

samples/client_publish.py

@@ -14,7 +14,7 @@ config = {
     "will": {
         "topic": "/will/client",
         "message": b"Dead or alive",
-        "qos": 0x01,
+        "qos": QOS_1,
         "retain": True,
     },
 }
@@ -22,7 +22,7 @@ config = {
 
 async def test_coro1() -> None:
     client = MQTTClient()
-    await client.connect("mqtt://test.mosquitto.org/")
+    await client.connect("mqtt://localhost:1883/")
     tasks = [
         asyncio.ensure_future(client.publish("a/b", b"TEST MESSAGE WITH QOS_0")),
         asyncio.ensure_future(client.publish("a/b", b"TEST MESSAGE WITH QOS_1", qos=QOS_1)),
@@ -35,8 +35,8 @@ async def test_coro1() -> None:
 
 async def test_coro2() -> None:
     try:
-        client = MQTTClient()
-        await client.connect("mqtt://test.mosquitto.org:1883/")
+        client = MQTTClient(config={'auto_connect': False})
+        await client.connect("mqtt://localhost:1884/")
         await client.publish("a/b", b"TEST MESSAGE WITH QOS_0", qos=0x00)
         await client.publish("a/b", b"TEST MESSAGE WITH QOS_1", qos=0x01)
         await client.publish("a/b", b"TEST MESSAGE WITH QOS_2", qos=0x02)

tests/plugins/test_plugins.py

@@ -132,7 +132,7 @@ async def test_plugin_exception_while_loading() -> None:
             _ = Broker(plugin_namespace='tests.mock_plugins', config=config)
 
 
-class TestAllEventsPlugin(BasePlugin[BaseContext]):
+class AllEventsPlugin(BasePlugin[BaseContext]):
     """A plugin to verify all events get sent to plugins."""
     def __init__(self, context: BaseContext) -> None:
         super().__init__(context)
@@ -161,7 +161,7 @@ async def test_all_plugin_events():
             match group:
                 case 'tests.mock_plugins':
                     return [
-                            EntryPoint(name='TestAllEventsPlugin', group='tests.mock_plugins', value='tests.plugins.test_plugins:TestAllEventsPlugin'),
+                            EntryPoint(name='AllEventsPlugin', group='tests.mock_plugins', value='tests.plugins.test_plugins:AllEventsPlugin'),
                         ]
                 case _:
                     return list()
@@ -192,7 +192,7 @@ async def test_all_plugin_events():
         await asyncio.sleep(1)
 
         # get the plugin so it doesn't get gc on shutdown
-        test_plugin = broker.plugins_manager.get_plugin('TestAllEventsPlugin')
+        test_plugin = broker.plugins_manager.get_plugin('AllEventsPlugin')
         await broker.shutdown()
         await asyncio.sleep(1)
 

tests/test_samples.py

@@ -75,13 +75,40 @@ async def test_broker_taboo():
     assert "Exception" not in stderr.decode("utf-8")
 
 
-@pytest.mark.timeout(20)
-def test_client_keepalive():
-    client_keepalive_main()
+@pytest.mark.timeout(25)
+@pytest.mark.asyncio
+async def test_client_keepalive():
+
+    broker = Broker()
+    await broker.start()
+    await asyncio.sleep(2)
+
+    keep_alive_script = Path(__file__).parent.parent / "samples/client_keepalive.py"
+    process = subprocess.Popen(["python", keep_alive_script], stdout=subprocess.PIPE, stderr=subprocess.PIPE)
+    await asyncio.sleep(2)
+
+    stdout, stderr = process.communicate()
+    assert "ERROR" not in stderr.decode("utf-8")
+    assert "Exception" not in stderr.decode("utf-8")
+
+    await broker.shutdown()
 
 
-def test_client_publish():
-    client_publish_main()
+@pytest.mark.asyncio
+async def test_client_publish():
+    broker = Broker()
+    await broker.start()
+    await asyncio.sleep(2)
+
+    client_publish = Path(__file__).parent.parent / "samples/client_publish.py"
+    process = subprocess.Popen(["python", client_publish], stdout=subprocess.PIPE, stderr=subprocess.PIPE)
+    await asyncio.sleep(2)
+
+    stdout, stderr = process.communicate()
+    assert "ERROR" not in stderr.decode("utf-8")
+    assert "Exception" not in stderr.decode("utf-8")
+
+    await broker.shutdown()
 
 broker_ssl_config = {
     "listeners": {