images/ha_xiaomi_home
2
0
Fork 0
mirror of https://github.com/XiaoMi/ha_xiaomi_home.git synced 2025-11-04 11:22:08 +08:00
Code Issues Actions Packages Projects Releases Wiki Activity

style: using logging for test case log print (#636)

Browse Source 
* style: using logging for test case log print

* fix: fix miot cloud test case resource error
This commit is contained in:
Paul Shawn
2025-01-13 10:54:18 +08:00
committed by GitHub
parent 5903c9a5a8
commit 045528fbf2
9 changed files with 153 additions and 88 deletions
Download Patch File Download Diff File Expand all files Collapse all files

43
test/check_rule_format.py
View File

@@ -1,11 +1,14 @@
# -*- coding: utf-8 -*-
"""Test rule format."""
import json
import logging
from os import listdir, path
from typing import Optional
import pytest
import yaml
_LOGGER = logging.getLogger(__name__)
ROOT_PATH: str = path.dirname(path.abspath(__file__))
TRANS_RELATIVE_PATH: str = path.join(
ROOT_PATH, '../custom_components/xiaomi_home/translations')
@@ -27,10 +30,10 @@ def load_json_file(file_path: str) -> Optional[dict]:
with open(file_path, 'r', encoding='utf-8') as file:
return json.load(file)
except FileNotFoundError:
print(file_path, 'is not found.')
_LOGGER.info('%s is not found.', file_path,)
return None
except json.JSONDecodeError:
print(file_path, 'is not a valid JSON file.')
_LOGGER.info('%s is not a valid JSON file.', file_path)
return None
@@ -44,10 +47,10 @@ def load_yaml_file(file_path: str) -> Optional[dict]:
with open(file_path, 'r', encoding='utf-8') as file:
return yaml.safe_load(file)
except FileNotFoundError:
print(file_path, 'is not found.')
_LOGGER.info('%s is not found.', file_path)
return None
except yaml.YAMLError:
print(file_path, 'is not a valid YAML file.')
_LOGGER.info('%s, is not a valid YAML file.', file_path)
return None
@@ -116,37 +119,43 @@ def bool_trans(d: dict) -> bool:
return False
default_trans: dict = d['translate'].pop('default')
if not default_trans:
print('default trans is empty')
_LOGGER.info('default trans is empty')
return False
default_keys: set[str] = set(default_trans.keys())
for key, trans in d['translate'].items():
trans_keys: set[str] = set(trans.keys())
if set(trans.keys()) != default_keys:
print('bool trans inconsistent', key, default_keys, trans_keys)
_LOGGER.info(
'bool trans inconsistent, %s, %s, %s',
key, default_keys, trans_keys)
return False
return True
def compare_dict_structure(dict1: dict, dict2: dict) -> bool:
if not isinstance(dict1, dict) or not isinstance(dict2, dict):
print('invalid type')
_LOGGER.info('invalid type')
return False
if dict1.keys() != dict2.keys():
print('inconsistent key values, ', dict1.keys(), dict2.keys())
_LOGGER.info(
'inconsistent key values, %s, %s', dict1.keys(), dict2.keys())
return False
for key in dict1:
if isinstance(dict1[key], dict) and isinstance(dict2[key], dict):
if not compare_dict_structure(dict1[key], dict2[key]):
print('inconsistent key values, dict, ', key)
_LOGGER.info(
'inconsistent key values, dict, %s', key)
return False
elif isinstance(dict1[key], list) and isinstance(dict2[key], list):
if not all(
isinstance(i, type(j))
for i, j in zip(dict1[key], dict2[key])):
print('inconsistent key values, list, ', key)
_LOGGER.info(
'inconsistent key values, list, %s', key)
return False
elif not isinstance(dict1[key], type(dict2[key])):
print('inconsistent key values, type, ', key)
_LOGGER.info(
'inconsistent key values, type, %s', key)
return False
return True
@@ -239,7 +248,8 @@ def test_miot_lang_integrity():
compare_dict: dict = load_json_file(
path.join(TRANS_RELATIVE_PATH, name))
if not compare_dict_structure(default_dict, compare_dict):
print('compare_dict_structure failed /translations, ', name)
_LOGGER.info(
'compare_dict_structure failed /translations, %s', name)
assert False
# Check i18n files structure
default_dict = load_json_file(
@@ -248,7 +258,8 @@ def test_miot_lang_integrity():
compare_dict: dict = load_json_file(
path.join(MIOT_I18N_RELATIVE_PATH, name))
if not compare_dict_structure(default_dict, compare_dict):
print('compare_dict_structure failed /miot/i18n, ', name)
_LOGGER.info(
'compare_dict_structure failed /miot/i18n, %s', name)
assert False
@@ -284,10 +295,10 @@ def test_miot_data_sort():
def test_sort_spec_data():
sort_data: dict = sort_bool_trans(file_path=SPEC_BOOL_TRANS_FILE)
save_json_file(file_path=SPEC_BOOL_TRANS_FILE, data=sort_data)
print(SPEC_BOOL_TRANS_FILE, 'formatted.')
_LOGGER.info('%s formatted.', SPEC_BOOL_TRANS_FILE)
sort_data = sort_multi_lang(file_path=SPEC_MULTI_LANG_FILE)
save_json_file(file_path=SPEC_MULTI_LANG_FILE, data=sort_data)
print(SPEC_MULTI_LANG_FILE, 'formatted.')
_LOGGER.info('%s formatted.', SPEC_MULTI_LANG_FILE)
sort_data = sort_spec_filter(file_path=SPEC_FILTER_FILE)
save_json_file(file_path=SPEC_FILTER_FILE, data=sort_data)
print(SPEC_FILTER_FILE, 'formatted.')
_LOGGER.info('%s formatted.', SPEC_FILTER_FILE)

29
test/conftest.py
View File

@@ -1,5 +1,6 @@
# -*- coding: utf-8 -*-
"""Pytest fixtures."""
import logging
import random
import shutil
import pytest
@@ -17,6 +18,21 @@ TEST_CLOUD_SERVER: str = 'cn'
DOMAIN_OAUTH2: str = 'oauth2_info'
DOMAIN_USER_INFO: str = 'user_info'
_LOGGER = logging.getLogger(__name__)
@pytest.fixture(scope='session', autouse=True)
def set_logger():
logger = logging.getLogger()
logger.setLevel(logging.INFO)
console_handler = logging.StreamHandler()
console_handler.setLevel(logging.INFO)
formatter = logging.Formatter(
'%(asctime)s - %(name)s - %(levelname)s - %(message)s')
console_handler.setFormatter(formatter)
logger.addHandler(console_handler)
_LOGGER.info('set logger, %s', logger)
@pytest.fixture(scope='session', autouse=True)
def load_py_file():
@@ -41,28 +57,28 @@ def load_py_file():
TEST_ROOT_PATH, '../custom_components/xiaomi_home/miot',
file_name),
path.join(TEST_FILES_PATH, file_name))
print('\nloaded test py files, ', file_list)
_LOGGER.info('\nloaded test py files, %s', file_list)
# Copy spec files to test folder
shutil.copytree(
src=path.join(
TEST_ROOT_PATH, '../custom_components/xiaomi_home/miot/specs'),
dst=path.join(TEST_FILES_PATH, 'specs'),
dirs_exist_ok=True)
print('loaded spec test folder, specs')
_LOGGER.info('loaded spec test folder, specs')
# Copy lan files to test folder
shutil.copytree(
src=path.join(
TEST_ROOT_PATH, '../custom_components/xiaomi_home/miot/lan'),
dst=path.join(TEST_FILES_PATH, 'lan'),
dirs_exist_ok=True)
print('loaded lan test folder, lan')
_LOGGER.info('loaded lan test folder, lan')
# Copy i18n files to test folder
shutil.copytree(
src=path.join(
TEST_ROOT_PATH, '../custom_components/xiaomi_home/miot/i18n'),
dst=path.join(TEST_FILES_PATH, 'i18n'),
dirs_exist_ok=True)
print('loaded i18n test folder, i18n')
_LOGGER.info('loaded i18n test folder, i18n')
yield
@@ -127,6 +143,11 @@ def test_domain_oauth2() -> str:
return DOMAIN_OAUTH2
@pytest.fixture(scope='session')
def test_name_uuid() -> str:
return f'{TEST_CLOUD_SERVER}_uuid'
@pytest.fixture(scope='session')
def test_domain_user_info() -> str:
return DOMAIN_USER_INFO

103
test/test_cloud.py
View File

@@ -1,11 +1,13 @@
# -*- coding: utf-8 -*-
"""Unit test for miot_cloud.py."""
import asyncio
import logging
import time
import webbrowser
import pytest
# pylint: disable=import-outside-toplevel, unused-argument
_LOGGER = logging.getLogger(__name__)
@pytest.mark.asyncio
@@ -15,18 +17,18 @@ async def test_miot_oauth_async(
test_cloud_server: str,
test_oauth2_redirect_url: str,
test_domain_oauth2: str,
test_uuid: str
test_uuid: str,
test_name_uuid: str
) -> dict:
from miot.const import OAUTH2_CLIENT_ID
from miot.miot_cloud import MIoTOauthClient
from miot.miot_storage import MIoTStorage
print('') # separate from previous output
miot_storage = MIoTStorage(test_cache_path)
local_uuid = await miot_storage.load_async(
domain=test_domain_oauth2, name=f'{test_cloud_server}_uuid', type_=str)
domain=test_domain_oauth2, name=test_name_uuid, type_=str)
uuid = str(local_uuid or test_uuid)
print(f'uuid: {uuid}')
_LOGGER.info('uuid: %s', uuid)
miot_oauth = MIoTOauthClient(
client_id=OAUTH2_CLIENT_ID,
redirect_url=test_oauth2_redirect_url,
@@ -42,13 +44,13 @@ async def test_miot_oauth_async(
and 'expires_ts' in load_info
and load_info['expires_ts'] > int(time.time())
):
print(f'load oauth info, {load_info}')
_LOGGER.info('load oauth info, %s', load_info)
oauth_info = load_info
if oauth_info is None:
# gen oauth url
auth_url: str = miot_oauth.gen_auth_url()
assert isinstance(auth_url, str)
print('auth url: ', auth_url)
_LOGGER.info('auth url: %s', auth_url)
# get code
webbrowser.open(auth_url)
code: str = input('input code: ')
@@ -57,22 +59,24 @@ async def test_miot_oauth_async(
res_obj = await miot_oauth.get_access_token_async(code=code)
assert res_obj is not None
oauth_info = res_obj
print(f'get_access_token result: {res_obj}')
_LOGGER.info('get_access_token result: %s', res_obj)
rc = await miot_storage.save_async(
test_domain_oauth2, test_cloud_server, oauth_info)
assert rc
print('save oauth info')
_LOGGER.info('save oauth info')
rc = await miot_storage.save_async(
test_domain_oauth2, f'{test_cloud_server}_uuid', uuid)
test_domain_oauth2, test_name_uuid, uuid)
assert rc
print('save uuid')
_LOGGER.info('save uuid')
access_token = oauth_info.get('access_token', None)
assert isinstance(access_token, str)
print(f'access_token: {access_token}')
_LOGGER.info('access_token: %s', access_token)
refresh_token = oauth_info.get('refresh_token', None)
assert isinstance(refresh_token, str)
print(f'refresh_token: {refresh_token}')
_LOGGER.info('refresh_token: %s', refresh_token)
await miot_oauth.deinit_async()
return oauth_info
@@ -82,16 +86,16 @@ async def test_miot_oauth_refresh_token(
test_cache_path: str,
test_cloud_server: str,
test_oauth2_redirect_url: str,
test_domain_oauth2: str
test_domain_oauth2: str,
test_name_uuid: str
):
from miot.const import OAUTH2_CLIENT_ID
from miot.miot_cloud import MIoTOauthClient
from miot.miot_storage import MIoTStorage
print('') # separate from previous output
miot_storage = MIoTStorage(test_cache_path)
uuid = await miot_storage.load_async(
domain=test_domain_oauth2, name=f'{test_cloud_server}_uuid', type_=str)
domain=test_domain_oauth2, name=test_name_uuid, type_=str)
assert isinstance(uuid, str)
oauth_info = await miot_storage.load_async(
domain=test_domain_oauth2, name=test_cloud_server, type_=dict)
@@ -100,7 +104,7 @@ async def test_miot_oauth_refresh_token(
assert 'refresh_token' in oauth_info
assert 'expires_ts' in oauth_info
remaining_time = oauth_info['expires_ts'] - int(time.time())
print(f'token remaining valid time: {remaining_time}s')
_LOGGER.info('token remaining valid time: %ss', remaining_time)
# Refresh token
miot_oauth = MIoTOauthClient(
client_id=OAUTH2_CLIENT_ID,
@@ -117,12 +121,14 @@ async def test_miot_oauth_refresh_token(
assert 'expires_ts' in update_info
remaining_time = update_info['expires_ts'] - int(time.time())
assert remaining_time > 0
print(f'refresh token, remaining valid time: {remaining_time}s')
_LOGGER.info('refresh token, remaining valid time: %ss', remaining_time)
# Save token
rc = await miot_storage.save_async(
test_domain_oauth2, test_cloud_server, update_info)
assert rc
print(f'refresh token success, {update_info}')
_LOGGER.info('refresh token success, %s', update_info)
await miot_oauth.deinit_async()
@pytest.mark.asyncio
@@ -135,7 +141,6 @@ async def test_miot_cloud_get_nickname_async(
from miot.const import OAUTH2_CLIENT_ID
from miot.miot_cloud import MIoTHttpClient
from miot.miot_storage import MIoTStorage
print('') # separate from previous output
miot_storage = MIoTStorage(test_cache_path)
oauth_info = await miot_storage.load_async(
@@ -149,7 +154,9 @@ async def test_miot_cloud_get_nickname_async(
user_info = await miot_http.get_user_info_async()
assert isinstance(user_info, dict) and 'miliaoNick' in user_info
nickname = user_info['miliaoNick']
print(f'your nickname: {nickname}\n')
_LOGGER.info('your nickname: %s', nickname)
await miot_http.deinit_async()
@pytest.mark.asyncio
@@ -163,7 +170,6 @@ async def test_miot_cloud_get_uid_async(
from miot.const import OAUTH2_CLIENT_ID
from miot.miot_cloud import MIoTHttpClient
from miot.miot_storage import MIoTStorage
print('') # separate from previous output
miot_storage = MIoTStorage(test_cache_path)
oauth_info = await miot_storage.load_async(
@@ -175,13 +181,15 @@ async def test_miot_cloud_get_uid_async(
uid = await miot_http.get_uid_async()
assert isinstance(uid, str)
print(f'your uid: {uid}\n')
_LOGGER.info('your uid: %s', uid)
# Save uid
rc = await miot_storage.save_async(
domain=test_domain_user_info,
name=f'uid_{test_cloud_server}', data=uid)
assert rc
await miot_http.deinit_async()
@pytest.mark.asyncio
@pytest.mark.dependency()
@@ -194,7 +202,6 @@ async def test_miot_cloud_get_homeinfos_async(
from miot.const import OAUTH2_CLIENT_ID
from miot.miot_cloud import MIoTHttpClient
from miot.miot_storage import MIoTStorage
print('') # separate from previous output
miot_storage = MIoTStorage(test_cache_path)
oauth_info = await miot_storage.load_async(
@@ -219,13 +226,15 @@ async def test_miot_cloud_get_homeinfos_async(
domain=test_domain_user_info,
name=f'uid_{test_cloud_server}', type_=str)
assert uid == uid2
print(f'your uid: {uid}\n')
_LOGGER.info('your uid: %s', uid)
# Get homes
home_list = homeinfos.get('home_list', {})
print(f'your home_list: {home_list}\n')
_LOGGER.info('your home_list: ,%s', home_list)
# Get share homes
share_home_list = homeinfos.get('share_home_list', {})
print(f'your share_home_list: {share_home_list}\n')
_LOGGER.info('your share_home_list: %s', share_home_list)
await miot_http.deinit_async()
@pytest.mark.asyncio
@@ -239,7 +248,6 @@ async def test_miot_cloud_get_devices_async(
from miot.const import OAUTH2_CLIENT_ID
from miot.miot_cloud import MIoTHttpClient
from miot.miot_storage import MIoTStorage
print('') # separate from previous output
miot_storage = MIoTStorage(test_cache_path)
oauth_info = await miot_storage.load_async(
@@ -261,13 +269,13 @@ async def test_miot_cloud_get_devices_async(
domain=test_domain_user_info,
name=f'uid_{test_cloud_server}', type_=str)
assert uid == uid2
print(f'your uid: {uid}\n')
_LOGGER.info('your uid: %s', uid)
# Get homes
homes = devices['homes']
print(f'your homes: {homes}\n')
_LOGGER.info('your homes: %s', homes)
# Get devices
devices = devices['devices']
print(f'your devices count: {len(devices)}\n')
_LOGGER.info('your devices count: %s', len(devices))
# Storage homes and devices
rc = await miot_storage.save_async(
domain=test_domain_user_info,
@@ -278,6 +286,8 @@ async def test_miot_cloud_get_devices_async(
name=f'devices_{test_cloud_server}', data=devices)
assert rc
await miot_http.deinit_async()
@pytest.mark.asyncio
@pytest.mark.dependency()
@@ -290,7 +300,6 @@ async def test_miot_cloud_get_devices_with_dids_async(
from miot.const import OAUTH2_CLIENT_ID
from miot.miot_cloud import MIoTHttpClient
from miot.miot_storage import MIoTStorage
print('') # separate from previous output
miot_storage = MIoTStorage(test_cache_path)
oauth_info = await miot_storage.load_async(
@@ -312,8 +321,11 @@ async def test_miot_cloud_get_devices_with_dids_async(
devices_info = await miot_http.get_devices_with_dids_async(
dids=test_list)
assert isinstance(devices_info, dict)
print(f'test did list, {len(test_list)}, {test_list}\n')
print(f'test result: {len(devices_info)}, {list(devices_info.keys())}\n')
_LOGGER.info('test did list, %s, %s', len(test_list), test_list)
_LOGGER.info(
'test result: %s, %s', len(devices_info), list(devices_info.keys()))
await miot_http.deinit_async()
@pytest.mark.asyncio
@@ -327,7 +339,6 @@ async def test_miot_cloud_get_prop_async(
from miot.const import OAUTH2_CLIENT_ID
from miot.miot_cloud import MIoTHttpClient
from miot.miot_storage import MIoTStorage
print('') # separate from previous output
miot_storage = MIoTStorage(test_cache_path)
oauth_info = await miot_storage.load_async(
@@ -349,7 +360,9 @@ async def test_miot_cloud_get_prop_async(
for did in test_list:
prop_value = await miot_http.get_prop_async(did=did, siid=2, piid=1)
device_name = local_devices[did]['name']
print(f'{device_name}({did}), prop.2.1: {prop_value}\n')
_LOGGER.info('%s(%s), prop.2.1: %s', device_name, did, prop_value)
await miot_http.deinit_async()
@pytest.mark.asyncio
@@ -363,7 +376,6 @@ async def test_miot_cloud_get_props_async(
from miot.const import OAUTH2_CLIENT_ID
from miot.miot_cloud import MIoTHttpClient
from miot.miot_storage import MIoTStorage
print('') # separate from previous output
miot_storage = MIoTStorage(test_cache_path)
oauth_info = await miot_storage.load_async(
@@ -384,8 +396,11 @@ async def test_miot_cloud_get_props_async(
test_list = did_list[:6]
prop_values = await miot_http.get_props_async(params=[
{'did': did, 'siid': 2, 'piid': 1} for did in test_list])
print(f'test did list, {len(test_list)}, {test_list}\n')
print(f'test result: {len(prop_values)}, {prop_values}\n')
_LOGGER.info('test did list, %s, %s', len(test_list), test_list)
_LOGGER.info('test result, %s, %s', len(prop_values), prop_values)
await miot_http.deinit_async()
@pytest.mark.skip(reason='skip danger operation')
@@ -404,7 +419,6 @@ async def test_miot_cloud_set_prop_async(
from miot.const import OAUTH2_CLIENT_ID
from miot.miot_cloud import MIoTHttpClient
from miot.miot_storage import MIoTStorage
print('') # separate from previous output
miot_storage = MIoTStorage(test_cache_path)
oauth_info = await miot_storage.load_async(
@@ -431,11 +445,13 @@ async def test_miot_cloud_set_prop_async(
assert test_did != '', 'no central hub gateway found'
result = await miot_http.set_prop_async(params=[{
'did': test_did, 'siid': 3, 'piid': 1, 'value': False}])
print(f'test did, {test_did}, prop.3.1=False -> {result}\n')
_LOGGER.info('test did, %s, prop.3.1=False -> %s', test_did, result)
await asyncio.sleep(1)
result = await miot_http.set_prop_async(params=[{
'did': test_did, 'siid': 3, 'piid': 1, 'value': True}])
print(f'test did, {test_did}, prop.3.1=True -> {result}\n')
_LOGGER.info('test did, %s, prop.3.1=True -> %s', test_did, result)
await miot_http.deinit_async()
@pytest.mark.skip(reason='skip danger operation')
@@ -454,7 +470,6 @@ async def test_miot_cloud_action_async(
from miot.const import OAUTH2_CLIENT_ID
from miot.miot_cloud import MIoTHttpClient
from miot.miot_storage import MIoTStorage
print('') # separate from previous output
miot_storage = MIoTStorage(test_cache_path)
oauth_info = await miot_storage.load_async(
@@ -482,4 +497,6 @@ async def test_miot_cloud_action_async(
result = await miot_http.action_async(
did=test_did, siid=4, aiid=1,
in_list=[{'piid': 1, 'value': 'hello world.'}])
print(f'test did, {test_did}, action.4.1 -> {result}\n')
_LOGGER.info('test did, %s, action.4.1 -> %s', test_did, result)
await miot_http.deinit_async()

2
test/test_common.py
View File

@@ -18,7 +18,7 @@ def test_miot_matcher():
if not matcher.get(topic=f'test/+/{l2}'):
matcher[f'test/+/{l2}'] = f'test/+/{l2}'
# Match
match_result: list[(str, dict)] = list(matcher.iter_all_nodes())
match_result: list[str] = list(matcher.iter_all_nodes())
assert len(match_result) == 120
match_result: list[str] = list(matcher.iter_match(topic='test/1/1'))
assert len(match_result) == 3

13
test/test_lan.py
View File

@@ -1,11 +1,14 @@
# -*- coding: utf-8 -*-
"""Unit test for miot_lan.py."""
import logging
from typing import Any
import pytest
import asyncio
from zeroconf import IPVersion
from zeroconf.asyncio import AsyncZeroconf
_LOGGER = logging.getLogger(__name__)
# pylint: disable=import-outside-toplevel, unused-argument
@@ -67,7 +70,7 @@ async def test_lan_async(test_devices: dict):
miot_network = MIoTNetwork()
await miot_network.init_async()
print('miot_network, ', miot_network.network_info)
_LOGGER.info('miot_network, %s', miot_network.network_info)
mips_service = MipsService(
aiozc=AsyncZeroconf(ip_version=IPVersion.V4Only))
await mips_service.init_async()
@@ -81,7 +84,7 @@ async def test_lan_async(test_devices: dict):
await miot_lan.vote_for_lan_ctrl_async(key='test', vote=True)
async def device_state_change(did: str, state: dict, ctx: Any):
print('device state change, ', did, state)
_LOGGER.info('device state change, %s, %s', did, state)
if did != test_did:
return
if (
@@ -91,10 +94,10 @@ async def test_lan_async(test_devices: dict):
# Test sub prop
miot_lan.sub_prop(
did=did, siid=3, piid=1, handler=lambda msg, ctx:
print(f'sub prop.3.1 msg, {did}={msg}'))
_LOGGER.info('sub prop.3.1 msg, %s=%s', did, msg))
miot_lan.sub_prop(
did=did, handler=lambda msg, ctx:
print(f'sub all device msg, {did}={msg}'))
_LOGGER.info('sub all device msg, %s=%s', did, msg))
evt_push_available.set()
else:
# miot_lan.unsub_prop(did=did, siid=3, piid=1)
@@ -102,7 +105,7 @@ async def test_lan_async(test_devices: dict):
evt_push_unavailable.set()
async def lan_state_change(state: bool):
print('lan state change, ', state)
_LOGGER.info('lan state change, %s', state)
if not state:
return
miot_lan.update_devices(devices={

12
test/test_mdns.py
View File

@@ -1,9 +1,12 @@
# -*- coding: utf-8 -*-
"""Unit test for miot_mdns.py."""
import logging
import pytest
from zeroconf import IPVersion
from zeroconf.asyncio import AsyncZeroconf
_LOGGER = logging.getLogger(__name__)
# pylint: disable=import-outside-toplevel, unused-argument
@@ -13,7 +16,7 @@ async def test_service_loop_async():
async def on_service_state_change(
group_id: str, state: MipsServiceState, data: MipsServiceData):
print(
_LOGGER.info(
'on_service_state_change, %s, %s, %s', group_id, state, data)
async with AsyncZeroconf(ip_version=IPVersion.V4Only) as aiozc:
@@ -21,8 +24,9 @@ async def test_service_loop_async():
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())
_LOGGER.info('get all service, %s', services_detail.keys())
for name, data in services_detail.items():
print(
'\tinfo, ', name, data['did'], data['addresses'], data['port'])
_LOGGER.info(
'\tinfo, %s, %s, %s, %s',
name, data['did'], data['addresses'], data['port'])
await mips_service.deinit_async()

13
test/test_network.py
View File

@@ -1,8 +1,11 @@
# -*- coding: utf-8 -*-
"""Unit test for miot_network.py."""
import logging
import pytest
import asyncio
_LOGGER = logging.getLogger(__name__)
# pylint: disable=import-outside-toplevel, unused-argument
@@ -12,16 +15,16 @@ async def test_network_monitor_loop_async():
miot_net = MIoTNetwork()
async def on_network_status_changed(status: bool):
print(f'on_network_status_changed, {status}')
_LOGGER.info('on_network_status_changed, %s', 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}')
_LOGGER.info('on_network_info_changed, %s, %s', status, info)
miot_net.sub_network_info(key='test', handler=on_network_info_changed)
await miot_net.init_async(3)
await miot_net.init_async()
await asyncio.sleep(3)
print(f'net status: {miot_net.network_status}')
print(f'net info: {miot_net.network_info}')
_LOGGER.info('net status: %s', miot_net.network_status)
_LOGGER.info('net info: %s', miot_net.network_info)
await miot_net.deinit_async()

9
test/test_spec.py
View File

@@ -1,11 +1,14 @@
# -*- coding: utf-8 -*-
"""Unit test for miot_spec.py."""
import json
import logging
import random
import time
from urllib.request import Request, urlopen
import pytest
_LOGGER = logging.getLogger(__name__)
# pylint: disable=import-outside-toplevel, unused-argument
@@ -79,10 +82,10 @@ async def test_spec_random_parse_async(test_cache_path, test_lang):
storage = MIoTStorage(test_cache_path)
spec_parser = MIoTSpecParser(lang=test_lang, storage=storage)
await spec_parser.init_async()
start_ts: int = time.time()*1000
start_ts = 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
end_ts: int = time.time()*1000
print(f'takes time, {test_count}, {end_ts-start_ts}')
end_ts = time.time()*1000
_LOGGER.info('takes time, %s, %s', test_count, end_ts-start_ts)

17
test/test_storage.py
View File

@@ -1,9 +1,12 @@
# -*- coding: utf-8 -*-
"""Unit test for miot_storage.py."""
import asyncio
import logging
from os import path
import pytest
_LOGGER = logging.getLogger(__name__)
# pylint: disable=import-outside-toplevel, unused-argument
@@ -101,7 +104,7 @@ async def test_multi_task_load_async(test_cache_path):
for _ in range(task_count):
task_list.append(asyncio.create_task(storage.load_async(
domain=test_domain, name=name, type_=dict)))
print(f'\ntask count, {len(task_list)}')
_LOGGER.info('task count, %s', len(task_list))
result: list = await asyncio.gather(*task_list)
assert None not in result
@@ -178,28 +181,28 @@ async def test_user_config_async(
config=config_update, replace=True)
assert (config_replace := await storage.load_user_config_async(
uid=test_uid, cloud_server=test_cloud_server)) == config_update
print('replace result, ', config_replace)
_LOGGER.info('replace result, %s', config_replace)
# Test query
query_keys = list(config_base.keys())
print('query keys, ', query_keys)
_LOGGER.info('query keys, %s', query_keys)
query_result = await storage.load_user_config_async(
uid=test_uid, cloud_server=test_cloud_server, keys=query_keys)
print('query result 1, ', query_result)
_LOGGER.info('query result 1, %s', query_result)
assert await storage.update_user_config_async(
uid=test_uid, cloud_server=test_cloud_server,
config=config_base, replace=True)
query_result = await storage.load_user_config_async(
uid=test_uid, cloud_server=test_cloud_server, keys=query_keys)
print('query result 2, ', query_result)
_LOGGER.info('query result 2, %s', query_result)
query_result = await storage.load_user_config_async(
uid=test_uid, cloud_server=test_cloud_server)
print('query result all, ', query_result)
_LOGGER.info('query result all, %s', query_result)
# Remove config
assert await storage.update_user_config_async(
uid=test_uid, cloud_server=test_cloud_server, config=None)
query_result = await storage.load_user_config_async(
uid=test_uid, cloud_server=test_cloud_server)
print('remove result, ', query_result)
_LOGGER.info('remove result, %s', query_result)
# Remove domain
assert await storage.remove_domain_async(domain='miot_config')