test: add test case for miot mdns and network

test/test_mdns.py

@@ -0,0 +1,30 @@
+# -*- coding: utf-8 -*-
+"""Unit test for miot_mdns.py."""
+import asyncio
+import pytest
+
+from zeroconf import IPVersion
+from zeroconf.asyncio import AsyncZeroconf
+
+# pylint: disable=import-outside-toplevel, unused-argument
+
+
+@pytest.mark.asyncio
+async def test_service_loop_async():
+    from miot.miot_mdns import MipsService, MipsServiceData, MipsServiceState
+
+    async def on_service_state_change(
+            group_id: str, state: MipsServiceState, data: MipsServiceData):
+        print(
+            'on_service_state_change, %s, %s, %s', group_id, state, data)
+
+    async with AsyncZeroconf(ip_version=IPVersion.V4Only) as aiozc:
+        mips_service = MipsService(aiozc)
+        mips_service.sub_service_change('test', '*', on_service_state_change)
+        await mips_service.init_async()
+        services_detail = mips_service.get_services()
+        print('get all service, ', services_detail.keys())
+        for name, data in services_detail.items():
+            print(
+                '\tinfo, ', name, data['did'], data['addresses'], data['port'])
+        await mips_service.deinit_async()

test/test_network.py

@@ -0,0 +1,27 @@
+# -*- coding: utf-8 -*-
+"""Unit test for miot_network.py."""
+import pytest
+import asyncio
+
+# pylint: disable=import-outside-toplevel, unused-argument
+
+
+@pytest.mark.asyncio
+async def test_network_monitor_loop_async():
+    from miot.miot_network import MIoTNetwork, InterfaceStatus, NetworkInfo
+    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)
+
+    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)
+
+    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()

test/test_spec.py

@@ -77,15 +77,12 @@ def get_release_instance() -> list[str]:
 
     # get local cache
     storage = MIoTStorage(test_cache_path)
-    # local_urns = storage.get_names(domain='miot_specs', type_=dict)
-    # print('local urns, %s', local_urns)
     spec_parser = MIoTSpecParser(lang=test_lang, storage=storage)
     await spec_parser.init_async()
-    start: int = time.time()*1000
+    start_ts: int = time.time()*1000
     for index in test_urn_index:
         urn: str = test_urns[int(index)]
         result = await spec_parser.parse(urn=urn, skip_cache=True)
         assert result is not None
-        # print('parse, %s, %s\n', urn, json.dumps(result.dump()))
-    end: int = time.time()*1000
-    print(f'takes time, {test_count}, {end-start}')
+    end_ts: int = time.time()*1000
+    print(f'takes time, {test_count}, {end_ts-start_ts}')

test/test_storage.py

@@ -69,9 +69,6 @@ async def test_variable_async(test_cache_path):
     assert len(storage.get_names(
         domain=test_domain, type_=dict)) == test_count/2
 
-    # Delete domain path
-    # assert await storage.remove_domain_async(test_domain)
-
 
 @pytest.mark.asyncio
 @pytest.mark.github