From 315c17286a6eeb96e56fb6ef45bf034b7293bc1d Mon Sep 17 00:00:00 2001 From: topsworld Date: Mon, 13 Jan 2025 10:41:39 +0800 Subject: [PATCH] style: using logging for test case log print --- test/check_rule_format.py | 43 ++++++++++------- test/conftest.py | 29 ++++++++++-- test/test_cloud.py | 99 ++++++++++++++++++++++----------------- test/test_common.py | 2 +- test/test_lan.py | 13 +++-- test/test_mdns.py | 12 +++-- test/test_network.py | 13 +++-- test/test_spec.py | 9 ++-- test/test_storage.py | 17 ++++--- 9 files changed, 149 insertions(+), 88 deletions(-) diff --git a/test/check_rule_format.py b/test/check_rule_format.py index 3c20afae..5075367a 100644 --- a/test/check_rule_format.py +++ b/test/check_rule_format.py @@ -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) diff --git a/test/conftest.py b/test/conftest.py index 63464cd9..48f07942 100644 --- a/test/conftest.py +++ b/test/conftest.py @@ -1,5 +1,6 @@ # -*- coding: utf-8 -*- """Pytest fixtures.""" +import logging import random import shutil import pytest @@ -17,6 +18,21 @@ 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 diff --git a/test/test_cloud.py b/test/test_cloud.py index acece121..6d5ef87d 100755 --- a/test/test_cloud.py +++ b/test/test_cloud.py @@ -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,22 @@ 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) return oauth_info @@ -82,16 +84,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 +102,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 +119,12 @@ 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) @pytest.mark.asyncio @@ -135,7 +137,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 +150,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 +166,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 +177,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 +198,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 +222,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 +244,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 +265,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 +282,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 +296,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 +317,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 +335,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 +356,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 +372,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 +392,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 +415,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 +441,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 +466,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 +493,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() diff --git a/test/test_common.py b/test/test_common.py index a6d68bcb..18a4736a 100644 --- a/test/test_common.py +++ b/test/test_common.py @@ -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 diff --git a/test/test_lan.py b/test/test_lan.py index a6051c07..a2861cca 100755 --- a/test/test_lan.py +++ b/test/test_lan.py @@ -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 device_state_change(did: str, state: dict, ctx: Any): # 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 device_state_change(did: str, state: dict, ctx: Any): 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={ diff --git a/test/test_mdns.py b/test/test_mdns.py index ddf6a10a..82cf477a 100755 --- a/test/test_mdns.py +++ b/test/test_mdns.py @@ -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 on_service_state_change( 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() diff --git a/test/test_network.py b/test/test_network.py index aa81a4e4..f59ddb20 100755 --- a/test/test_network.py +++ b/test/test_network.py @@ -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() diff --git a/test/test_spec.py b/test/test_spec.py index 57ccbb67..248e9d83 100755 --- a/test/test_spec.py +++ b/test/test_spec.py @@ -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 @@ def get_release_instance() -> list[str]: 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) diff --git a/test/test_storage.py b/test/test_storage.py index 76ec5100..ace0c532 100755 --- a/test/test_storage.py +++ b/test/test_storage.py @@ -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')