From dfc99b10f34e4a5e067e515a2c415b3b94bae9b7 Mon Sep 17 00:00:00 2001 From: Imad <118115991+Imadnajam@users.noreply.github.com> Date: Wed, 18 Dec 2024 21:04:15 +0100 Subject: [PATCH] add robust unit test for MIoTNetwork async monitoring Added event handlers for network_status and network_info changes, ensuring proper logging and validation of state transitions. --- test/test_network.py | 42 +++++++++++++++++++++++++++++------------- 1 file changed, 29 insertions(+), 13 deletions(-) diff --git a/test/test_network.py b/test/test_network.py index aa81a4e..8f6d372 100755 --- a/test/test_network.py +++ b/test/test_network.py @@ -5,23 +5,39 @@ # pylint: disable=import-outside-toplevel, unused-argument - @pytest.mark.asyncio async def test_network_monitor_loop_async(): + """ + Test asynchronous network monitoring logic of MIoTNetwork. + """ + from miot.miot_network import MIoTNetwork, InterfaceStatus, NetworkInfo + + # Initialize MIoTNetwork instance miot_net = MIoTNetwork() - async def on_network_status_changed(status: bool): - print(f'on_network_status_changed, {status}') - miot_net.sub_network_status(key='test', handler=on_network_status_changed) + # Define handlers for network events + async def handle_network_status_change(status: bool): + print(f"Network status changed: {status}") + + async def handle_network_info_change( + status: InterfaceStatus, info: NetworkInfo + ): + print(f"Network info changed: {status}, {info}") + + # Subscribe handlers to network events + miot_net.sub_network_status(key="test", handler=handle_network_status_change) + miot_net.sub_network_info(key="test", handler=handle_network_info_change) + + # Test the network monitoring functionality + try: + await miot_net.init_async(timeout=3) + await asyncio.sleep(3) - async def on_network_info_changed( - status: InterfaceStatus, info: NetworkInfo): - print(f'on_network_info_changed, {status}, {info}') - miot_net.sub_network_info(key='test', handler=on_network_info_changed) + # Log current network state + print(f"Network Status: {miot_net.network_status}") + print(f"Network Info: {miot_net.network_info}") - await miot_net.init_async(3) - await asyncio.sleep(3) - print(f'net status: {miot_net.network_status}') - print(f'net info: {miot_net.network_info}') - await miot_net.deinit_async() + finally: + # Ensure proper cleanup + await miot_net.deinit_async()