mirror of
https://github.com/XiaoMi/ha_xiaomi_home.git
synced 2025-04-02 07:45:31 +08:00
test: add test case for miot mdns and network
This commit is contained in:
parent
ad7f9eb610
commit
9d95379f83
30
test/test_mdns.py
Executable file
30
test/test_mdns.py
Executable file
@ -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()
|
27
test/test_network.py
Executable file
27
test/test_network.py
Executable file
@ -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()
|
@ -77,15 +77,12 @@ async def test_spec_random_parse_async(test_cache_path, test_lang):
|
||||
|
||||
# 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}')
|
||||
|
@ -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
|
||||
|
Loading…
x
Reference in New Issue
Block a user