Skip to content

Commit

Permalink
add robust unit test for MIoTNetwork async monitoring
Browse files Browse the repository at this point in the history
Added event handlers for network_status and network_info changes, ensuring proper logging and validation of state transitions.
  • Loading branch information
Imadnajam authored Dec 18, 2024
1 parent 54e2637 commit dfc99b1
Showing 1 changed file with 29 additions and 13 deletions.
42 changes: 29 additions & 13 deletions test/test_network.py
Original file line number Diff line number Diff line change
Expand Up @@ -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()

0 comments on commit dfc99b1

Please sign in to comment.