diff --git a/test/conftest.py b/test/conftest.py index f6ee926..04d2a48 100644 --- a/test/conftest.py +++ b/test/conftest.py @@ -5,6 +5,7 @@ import pytest from os import path, makedirs TEST_ROOT_PATH: str = path.dirname(path.abspath(__file__)) +TEST_CACHE_PATH: str = path.join(TEST_ROOT_PATH, 'test_cache') TEST_LANG: str = 'zh-Hans' TEST_UID: str = '123456789' TEST_CLOUD_SERVER: str = 'cn' @@ -32,28 +33,29 @@ def load_py_file(): TEST_ROOT_PATH, '../custom_components/xiaomi_home/miot', file_name), path.join(TEST_ROOT_PATH, 'miot', file_name)) - print('\nloaded test py file, %s' % file_list) + print('loaded test py file, ', 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_ROOT_PATH, 'miot/specs'), dirs_exist_ok=True) - print('\nloaded spec test folder, miot/specs') + print('loaded spec test folder, miot/specs') # Copy i18n files to test folder shutil.copytree( src=path.join( TEST_ROOT_PATH, '../custom_components/xiaomi_home/miot/i18n'), dst=path.join(TEST_ROOT_PATH, 'miot/i18n'), dirs_exist_ok=True) - print('\nloaded i18n test folder, miot/i18n') + print('loaded i18n test folder, miot/i18n') + yield - # shutil.rmtree(path.join(TEST_ROOT_PATH, 'miot')) - # print('\nremoved test file, %s', file_list) - # shutil.rmtree(path.join(TEST_ROOT_PATH, 'miot/specs')) - # print('\nremoved test folder, %s', 'miot/specs') - # shutil.rmtree(path.join(TEST_ROOT_PATH, 'miot/i18n')) - # print('\nremoved test folder, %s', 'miot/i18n') + + shutil.rmtree(path.join(TEST_ROOT_PATH, 'miot')) + print('removed test file, ', file_list) + + shutil.rmtree(TEST_CACHE_PATH) + print('removed test file, ', TEST_CACHE_PATH) @pytest.fixture(scope='session') @@ -63,9 +65,8 @@ def test_root_path() -> str: @pytest.fixture(scope='session') def test_cache_path() -> str: - cache_path: str = path.join(TEST_ROOT_PATH, 'test_cache') - makedirs(cache_path, exist_ok=True) - return cache_path + makedirs(TEST_CACHE_PATH, exist_ok=True) + return TEST_CACHE_PATH @pytest.fixture(scope='session') diff --git a/test/test_spec.py b/test/test_spec.py new file mode 100755 index 0000000..8115f77 --- /dev/null +++ b/test/test_spec.py @@ -0,0 +1,91 @@ +# -*- coding: utf-8 -*- +"""Unit test for miot_spec.py.""" +import json +import random +import time +from urllib.request import Request, urlopen +import pytest + +# pylint: disable=import-outside-toplevel, unused-argument + + +@pytest.mark.parametrize('urn', [ + 'urn:miot-spec-v2:device:gateway:0000A019:xiaomi-hub1:3', + 'urn:miot-spec-v2:device:light:0000A001:mijia-group3:3:0000C802', + 'urn:miot-spec-v2:device:air-conditioner:0000A004:xiaomi-ar03r1:1', + 'urn:miot-spec-v2:device:air-purifier:0000A007:xiaomi-va5:1:0000D050', + 'urn:miot-spec-v2:device:humidifier:0000A00E:xiaomi-p800:1', + 'urn:miot-spec-v2:device:curtain:0000A00C:xiaomi-acn010:1:0000D031', + 'urn:miot-spec-v2:device:motion-sensor:0000A014:xiaomi-pir1:2', + 'urn:miot-spec-v2:device:light:0000A001:philips-strip3:2']) +@pytest.mark.asyncio +@pytest.mark.dependency() +async def test_spec_parse_async(test_cache_path, test_lang, urn): + from miot.miot_spec import MIoTSpecParser + from miot.miot_storage import MIoTStorage + + storage = MIoTStorage(test_cache_path) + spec_parser = MIoTSpecParser(lang=test_lang, storage=storage) + await spec_parser.init_async() + assert await spec_parser.parse(urn=urn) + + +@pytest.mark.parametrize('urn_list', [[ + 'urn:miot-spec-v2:device:gateway:0000A019:xiaomi-hub1:3', + 'urn:miot-spec-v2:device:light:0000A001:mijia-group3:3:0000C802', + 'urn:miot-spec-v2:device:air-conditioner:0000A004:xiaomi-ar03r1:1', + 'urn:miot-spec-v2:device:air-purifier:0000A007:xiaomi-va5:1:0000D050', + 'urn:miot-spec-v2:device:humidifier:0000A00E:xiaomi-p800:1', + 'urn:miot-spec-v2:device:curtain:0000A00C:xiaomi-acn010:1:0000D031', + 'urn:miot-spec-v2:device:motion-sensor:0000A014:xiaomi-pir1:2', + 'urn:miot-spec-v2:device:light:0000A001:philips-strip3:2']]) +@pytest.mark.asyncio +@pytest.mark.dependency() +async def test_spec_refresh_async(test_cache_path, test_lang, urn_list): + from miot.miot_spec import MIoTSpecParser + from miot.miot_storage import MIoTStorage + + storage = MIoTStorage(test_cache_path) + spec_parser = MIoTSpecParser(lang=test_lang, storage=storage) + await spec_parser.init_async() + assert await spec_parser.refresh_async(urn_list=urn_list) == len(urn_list) + + +@pytest.mark.asyncio +@pytest.mark.dependency() +async def test_spec_random_parse_async(test_cache_path, test_lang): + from miot.miot_spec import MIoTSpecParser + from miot.miot_storage import MIoTStorage + + test_count = 10 + # get test data + + def get_release_instance() -> list[str]: + request = Request( + 'https://miot-spec.org/miot-spec-v2/instances?status=released', + method='GET') + with urlopen(request) as response: + content = response.read() + res_obj = json.loads(str(content, 'utf-8')) + result: list[str] = [] + for item in res_obj['instances']: + result.append(item['type']) + return result + test_urns: list[str] = get_release_instance() + test_urn_index: list[int] = random.sample( + list(range(len(test_urns))), test_count) + + # 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 + 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}') diff --git a/test/test_storage.py b/test/test_storage.py index 9fbb068..22e5f67 100755 --- a/test/test_storage.py +++ b/test/test_storage.py @@ -75,7 +75,7 @@ async def test_variable_async(test_cache_path): @pytest.mark.asyncio @pytest.mark.github -@pytest.mark.dependency(depends=['test_variable_async']) +@pytest.mark.dependency() async def test_load_domain_async(test_cache_path): from miot.miot_storage import MIoTStorage @@ -90,7 +90,7 @@ async def test_load_domain_async(test_cache_path): @pytest.mark.asyncio @pytest.mark.github -@pytest.mark.dependency(depends=['test_variable_async']) +@pytest.mark.dependency() async def test_multi_task_load_async(test_cache_path): from miot.miot_storage import MIoTStorage @@ -181,40 +181,34 @@ 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('\nreplace result, ', config_replace) + print('replace result, ', config_replace) # test query query_keys = list(config_base.keys()) - print('\nquery keys, %s', query_keys) + print('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('\nquery result 1, ', 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) - print('\nquery result 2, ', query_result) + print('query result 2, ', query_result) query_result = await storage.load_user_config_async( uid=test_uid, cloud_server=test_cloud_server) - print('\nquery result all, ', 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) - print('\nremove result, ', query_result) + print('remove result, ', query_result) # remove domain assert await storage.remove_domain_async(domain='miot_config') @pytest.mark.asyncio -@pytest.mark.github -@pytest.mark.dependency(depends=[ - 'test_variable_async', - 'test_load_domain_async', - 'test_multi_task_load_async', - 'test_file_save_load_async', - 'test_user_config_async']) +@pytest.mark.skip(reason='clean') async def test_clear_async(test_cache_path): from miot.miot_storage import MIoTStorage