Merge 8d5feba1e0 into 5903c9a5a8

custom_components/xiaomi_home/miot/miot_cloud.py

@@ -106,7 +106,7 @@ class MIoTOauthClient:
 
     @property
     def state(self) -> str:
-        return self._state
+        return self.state
 
     async def deinit_async(self) -> None:
         if self._session and not self._session.closed:

custom_components/xiaomi_home/miot/miot_mdns.py

@@ -117,7 +117,7 @@ class MipsServiceData:
         self.type = service_info.type
         self.server = service_info.server or ''
         # Parse profile
-        self.did = str(int.from_bytes(self.profile_bin[1:9], byteorder='big'))
+        self.did = str(int.from_bytes(self.profile_bin[1:9]))
         self.group_id = binascii.hexlify(
             self.profile_bin[9:17][::-1]).decode('utf-8')
         self.role = int(self.profile_bin[20] >> 4)

test/check_rule_format.py

@@ -1,14 +1,11 @@
 # -*- 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')
@@ -30,10 +27,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:
-        _LOGGER.info('%s is not found.', file_path,)
+        print(file_path, 'is not found.')
         return None
     except json.JSONDecodeError:
-        _LOGGER.info('%s is not a valid JSON file.', file_path)
+        print(file_path, 'is not a valid JSON file.')
         return None
 
 
@@ -47,10 +44,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:
-        _LOGGER.info('%s is not found.', file_path)
+        print(file_path, 'is not found.')
         return None
     except yaml.YAMLError:
-        _LOGGER.info('%s, is not a valid YAML file.', file_path)
+        print(file_path, 'is not a valid YAML file.')
         return None
 
 
@@ -119,43 +116,37 @@ def bool_trans(d: dict) -> bool:
         return False
     default_trans: dict = d['translate'].pop('default')
     if not default_trans:
-        _LOGGER.info('default trans is empty')
+        print('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:
-            _LOGGER.info(
-                'bool trans inconsistent, %s, %s, %s',
-                key, default_keys, trans_keys)
+            print('bool trans inconsistent', 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):
-        _LOGGER.info('invalid type')
+        print('invalid type')
         return False
     if dict1.keys() != dict2.keys():
-        _LOGGER.info(
-            'inconsistent key values, %s, %s', dict1.keys(), dict2.keys())
+        print('inconsistent key values, ', 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]):
-                _LOGGER.info(
-                    'inconsistent key values, dict, %s', key)
+                print('inconsistent key values, dict, ', 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])):
-                _LOGGER.info(
-                    'inconsistent key values, list, %s', key)
+                print('inconsistent key values, list, ', key)
                 return False
         elif not isinstance(dict1[key], type(dict2[key])):
-            _LOGGER.info(
-                'inconsistent key values, type, %s', key)
+            print('inconsistent key values, type, ', key)
             return False
     return True
 
@@ -248,8 +239,7 @@ 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):
-            _LOGGER.info(
-                'compare_dict_structure failed /translations, %s', name)
+            print('compare_dict_structure failed /translations, ', name)
             assert False
     # Check i18n files structure
     default_dict = load_json_file(
@@ -258,8 +248,7 @@ 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):
-            _LOGGER.info(
-                'compare_dict_structure failed /miot/i18n, %s', name)
+            print('compare_dict_structure failed /miot/i18n, ', name)
             assert False
 
 
@@ -295,10 +284,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)
-    _LOGGER.info('%s formatted.', SPEC_BOOL_TRANS_FILE)
+    print(SPEC_BOOL_TRANS_FILE, 'formatted.')
     sort_data = sort_multi_lang(file_path=SPEC_MULTI_LANG_FILE)
     save_json_file(file_path=SPEC_MULTI_LANG_FILE, data=sort_data)
-    _LOGGER.info('%s formatted.', SPEC_MULTI_LANG_FILE)
+    print(SPEC_MULTI_LANG_FILE, 'formatted.')
     sort_data = sort_spec_filter(file_path=SPEC_FILTER_FILE)
     save_json_file(file_path=SPEC_FILTER_FILE, data=sort_data)
-    _LOGGER.info('%s formatted.', SPEC_FILTER_FILE)
+    print(SPEC_FILTER_FILE, 'formatted.')

test/conftest.py

@@ -1,6 +1,5 @@
 # -*- coding: utf-8 -*-
 """Pytest fixtures."""
-import logging
 import random
 import shutil
 import pytest
@@ -18,21 +17,6 @@ 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():
@@ -57,28 +41,28 @@ def load_py_file():
                 TEST_ROOT_PATH, '../custom_components/xiaomi_home/miot',
                 file_name),
             path.join(TEST_FILES_PATH, file_name))
-    _LOGGER.info('\nloaded test py files, %s', file_list)
+    print('\nloaded test py files, ', 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)
-    _LOGGER.info('loaded spec test folder, specs')
+    print('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)
-    _LOGGER.info('loaded lan test folder, lan')
+    print('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)
-    _LOGGER.info('loaded i18n test folder, i18n')
+    print('loaded i18n test folder, i18n')
 
     yield
 
@@ -143,11 +127,6 @@ 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

test/test_cloud.py

@@ -1,13 +1,11 @@
 # -*- 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
@@ -17,18 +15,18 @@ async def test_miot_oauth_async(
     test_cloud_server: str,
     test_oauth2_redirect_url: str,
     test_domain_oauth2: str,
-    test_uuid: str,
-    test_name_uuid: str
+    test_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=test_name_uuid, type_=str)
+        domain=test_domain_oauth2, name=f'{test_cloud_server}_uuid', type_=str)
     uuid = str(local_uuid or test_uuid)
-    _LOGGER.info('uuid: %s', uuid)
+    print(f'uuid: {uuid}')
     miot_oauth = MIoTOauthClient(
         client_id=OAUTH2_CLIENT_ID,
         redirect_url=test_oauth2_redirect_url,
@@ -44,13 +42,13 @@ async def test_miot_oauth_async(
         and 'expires_ts' in load_info
         and load_info['expires_ts'] > int(time.time())
     ):
-        _LOGGER.info('load oauth info, %s', load_info)
+        print(f'load oauth info, {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)
-        _LOGGER.info('auth url: %s', auth_url)
+        print('auth url: ', auth_url)
         # get code
         webbrowser.open(auth_url)
         code: str = input('input code: ')
@@ -59,24 +57,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
-        _LOGGER.info('get_access_token result: %s', res_obj)
+        print(f'get_access_token result: {res_obj}')
         rc = await miot_storage.save_async(
             test_domain_oauth2, test_cloud_server, oauth_info)
         assert rc
-        _LOGGER.info('save oauth info')
+        print('save oauth info')
         rc = await miot_storage.save_async(
-            test_domain_oauth2, test_name_uuid, uuid)
+            test_domain_oauth2, f'{test_cloud_server}_uuid', uuid)
         assert rc
-        _LOGGER.info('save uuid')
+        print('save uuid')
 
     access_token = oauth_info.get('access_token', None)
     assert isinstance(access_token, str)
-    _LOGGER.info('access_token: %s', access_token)
+    print(f'access_token: {access_token}')
     refresh_token = oauth_info.get('refresh_token', None)
     assert isinstance(refresh_token, str)
-    _LOGGER.info('refresh_token: %s', refresh_token)
-
-    await miot_oauth.deinit_async()
+    print(f'refresh_token: {refresh_token}')
     return oauth_info
 
 
@@ -86,16 +82,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_name_uuid: str
+    test_domain_oauth2: 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=test_name_uuid, type_=str)
+        domain=test_domain_oauth2, name=f'{test_cloud_server}_uuid', type_=str)
     assert isinstance(uuid, str)
     oauth_info = await miot_storage.load_async(
         domain=test_domain_oauth2, name=test_cloud_server, type_=dict)
@@ -104,7 +100,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())
-    _LOGGER.info('token remaining valid time: %ss', remaining_time)
+    print(f'token remaining valid time: {remaining_time}s')
     # Refresh token
     miot_oauth = MIoTOauthClient(
         client_id=OAUTH2_CLIENT_ID,
@@ -121,14 +117,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
-    _LOGGER.info('refresh token, remaining valid time: %ss', remaining_time)
+    print(f'refresh token, remaining valid time: {remaining_time}s')
     # Save token
     rc = await miot_storage.save_async(
         test_domain_oauth2, test_cloud_server, update_info)
     assert rc
-    _LOGGER.info('refresh token success, %s', update_info)
-
-    await miot_oauth.deinit_async()
+    print(f'refresh token success, {update_info}')
 
 
 @pytest.mark.asyncio
@@ -141,6 +135,7 @@ 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(
@@ -154,9 +149,7 @@ 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']
-    _LOGGER.info('your nickname: %s', nickname)
-
-    await miot_http.deinit_async()
+    print(f'your nickname: {nickname}\n')
 
 
 @pytest.mark.asyncio
@@ -170,6 +163,7 @@ 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(
@@ -181,15 +175,13 @@ async def test_miot_cloud_get_uid_async(
 
     uid = await miot_http.get_uid_async()
     assert isinstance(uid, str)
-    _LOGGER.info('your uid: %s', uid)
+    print(f'your uid: {uid}\n')
     # 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()
@@ -202,6 +194,7 @@ 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(
@@ -226,15 +219,13 @@ async def test_miot_cloud_get_homeinfos_async(
         domain=test_domain_user_info,
         name=f'uid_{test_cloud_server}', type_=str)
     assert uid == uid2
-    _LOGGER.info('your uid: %s', uid)
+    print(f'your uid: {uid}\n')
     # Get homes
     home_list = homeinfos.get('home_list', {})
-    _LOGGER.info('your home_list: ,%s', home_list)
+    print(f'your home_list: {home_list}\n')
     # Get share homes
     share_home_list = homeinfos.get('share_home_list', {})
-    _LOGGER.info('your share_home_list: %s', share_home_list)
-
-    await miot_http.deinit_async()
+    print(f'your share_home_list: {share_home_list}\n')
 
 
 @pytest.mark.asyncio
@@ -248,6 +239,7 @@ 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(
@@ -269,13 +261,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
-    _LOGGER.info('your uid: %s', uid)
+    print(f'your uid: {uid}\n')
     # Get homes
     homes = devices['homes']
-    _LOGGER.info('your homes: %s', homes)
+    print(f'your homes: {homes}\n')
     # Get devices
     devices = devices['devices']
-    _LOGGER.info('your devices count: %s', len(devices))
+    print(f'your devices count: {len(devices)}\n')
     # Storage homes and devices
     rc = await miot_storage.save_async(
         domain=test_domain_user_info,
@@ -286,8 +278,6 @@ 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()
@@ -300,6 +290,7 @@ 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(
@@ -321,11 +312,8 @@ 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)
-    _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()
+    print(f'test did list, {len(test_list)}, {test_list}\n')
+    print(f'test result: {len(devices_info)}, {list(devices_info.keys())}\n')
 
 
 @pytest.mark.asyncio
@@ -339,6 +327,7 @@ 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(
@@ -360,9 +349,7 @@ 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']
-        _LOGGER.info('%s(%s), prop.2.1: %s', device_name, did, prop_value)
-
-    await miot_http.deinit_async()
+        print(f'{device_name}({did}), prop.2.1: {prop_value}\n')
 
 
 @pytest.mark.asyncio
@@ -376,6 +363,7 @@ 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(
@@ -396,11 +384,8 @@ 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])
-
-    _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()
+    print(f'test did list, {len(test_list)}, {test_list}\n')
+    print(f'test result: {len(prop_values)}, {prop_values}\n')
 
 
 @pytest.mark.skip(reason='skip danger operation')
@@ -419,6 +404,7 @@ 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(
@@ -445,13 +431,11 @@ 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}])
-    _LOGGER.info('test did, %s, prop.3.1=False -> %s', test_did, result)
+    print(f'test did, {test_did}, prop.3.1=False -> {result}\n')
     await asyncio.sleep(1)
     result = await miot_http.set_prop_async(params=[{
         'did': test_did, 'siid': 3, 'piid': 1, 'value': True}])
-    _LOGGER.info('test did, %s, prop.3.1=True -> %s', test_did, result)
-
-    await miot_http.deinit_async()
+    print(f'test did, {test_did}, prop.3.1=True -> {result}\n')
 
 
 @pytest.mark.skip(reason='skip danger operation')
@@ -470,6 +454,7 @@ 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(
@@ -497,6 +482,4 @@ 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.'}])
-    _LOGGER.info('test did, %s, action.4.1 -> %s', test_did, result)
-
-    await miot_http.deinit_async()
+    print(f'test did, {test_did}, action.4.1 -> {result}\n')

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] = list(matcher.iter_all_nodes())
+    match_result: list[(str, dict)] = 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

test/test_lan.py

@@ -1,14 +1,11 @@
 # -*- 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
 
 
@@ -70,7 +67,7 @@ async def test_lan_async(test_devices: dict):
 
     miot_network = MIoTNetwork()
     await miot_network.init_async()
-    _LOGGER.info('miot_network, %s', miot_network.network_info)
+    print('miot_network, ', miot_network.network_info)
     mips_service = MipsService(
         aiozc=AsyncZeroconf(ip_version=IPVersion.V4Only))
     await mips_service.init_async()
@@ -84,7 +81,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):
-        _LOGGER.info('device state change, %s, %s', did, state)
+        print('device state change, ', did, state)
         if did != test_did:
             return
         if (
@@ -94,10 +91,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:
-                    _LOGGER.info('sub prop.3.1 msg, %s=%s', did, msg))
+                    print(f'sub prop.3.1 msg, {did}={msg}'))
             miot_lan.sub_prop(
                 did=did, handler=lambda msg, ctx:
-                    _LOGGER.info('sub all device msg, %s=%s', did, msg))
+                    print(f'sub all device msg, {did}={msg}'))
             evt_push_available.set()
         else:
             # miot_lan.unsub_prop(did=did, siid=3, piid=1)
@@ -105,7 +102,7 @@ async def test_lan_async(test_devices: dict):
             evt_push_unavailable.set()
 
     async def lan_state_change(state: bool):
-        _LOGGER.info('lan state change, %s', state)
+        print('lan state change, ', state)
         if not state:
             return
         miot_lan.update_devices(devices={

test/test_mdns.py

@@ -1,12 +1,9 @@
 # -*- 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
 
 
@@ -16,7 +13,7 @@ async def test_service_loop_async():
 
     async def on_service_state_change(
             group_id: str, state: MipsServiceState, data: MipsServiceData):
-        _LOGGER.info(
+        print(
             'on_service_state_change, %s, %s, %s', group_id, state, data)
 
     async with AsyncZeroconf(ip_version=IPVersion.V4Only) as aiozc:
@@ -24,9 +21,8 @@ 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()
-        _LOGGER.info('get all service, %s', services_detail.keys())
+        print('get all service, ', services_detail.keys())
         for name, data in services_detail.items():
-            _LOGGER.info(
-                '\tinfo, %s, %s, %s, %s',
-                name, data['did'], data['addresses'], data['port'])
+            print(
+                '\tinfo, ', name, data['did'], data['addresses'], data['port'])
         await mips_service.deinit_async()

test/test_network.py

@@ -1,11 +1,8 @@
 # -*- 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
 
 
@@ -15,16 +12,16 @@ async def test_network_monitor_loop_async():
     miot_net = MIoTNetwork()
 
     async def on_network_status_changed(status: bool):
-        _LOGGER.info('on_network_status_changed, %s', status)
+        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):
-        _LOGGER.info('on_network_info_changed, %s, %s', status, info)
+        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()
+    await miot_net.init_async(3)
     await asyncio.sleep(3)
-    _LOGGER.info('net status: %s', miot_net.network_status)
-    _LOGGER.info('net info: %s', miot_net.network_info)
+    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

@@ -1,14 +1,11 @@
 # -*- 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
 
 
@@ -82,10 +79,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 = 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
-    end_ts = time.time()*1000
-    _LOGGER.info('takes time, %s, %s', test_count, end_ts-start_ts)
+    end_ts: int = time.time()*1000
+    print(f'takes time, {test_count}, {end_ts-start_ts}')

test/test_storage.py

@@ -1,12 +1,9 @@
 # -*- 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
 
 
@@ -104,7 +101,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)))
-    _LOGGER.info('task count, %s', len(task_list))
+    print(f'\ntask count, {len(task_list)}')
     result: list = await asyncio.gather(*task_list)
     assert None not in result
 
@@ -181,28 +178,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
-    _LOGGER.info('replace result, %s', config_replace)
+    print('replace result, ', config_replace)
     # Test query
     query_keys = list(config_base.keys())
-    _LOGGER.info('query keys, %s', query_keys)
+    print('query keys, ', query_keys)
     query_result = await storage.load_user_config_async(
         uid=test_uid, cloud_server=test_cloud_server, keys=query_keys)
-    _LOGGER.info('query result 1, %s', query_result)
+    print('query result 1, ', 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)
-    _LOGGER.info('query result 2, %s', query_result)
+    print('query result 2, ', query_result)
     query_result = await storage.load_user_config_async(
         uid=test_uid, cloud_server=test_cloud_server)
-    _LOGGER.info('query result all, %s', query_result)
+    print('query result all, ', 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)
-    _LOGGER.info('remove result, %s', query_result)
+    print('remove result, ', query_result)
     # Remove domain
     assert await storage.remove_domain_async(domain='miot_config')