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()