mirror of
https://github.com/XiaoMi/ha_xiaomi_home.git
synced 2025-04-22 09:32:45 +08:00
test: add test case for miot_common.py
This commit is contained in:
parent
687198469f
commit
db3834793c
32
test/test_common.py
Normal file
32
test/test_common.py
Normal file
@ -0,0 +1,32 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
"""Unit test for miot_common.py."""
|
||||
import pytest
|
||||
|
||||
# pylint: disable=import-outside-toplevel, unused-argument
|
||||
|
||||
|
||||
@pytest.mark.github
|
||||
def test_miot_matcher():
|
||||
from miot.common import MIoTMatcher
|
||||
|
||||
matcher: MIoTMatcher = MIoTMatcher()
|
||||
# Add
|
||||
for l1 in range(1, 11):
|
||||
matcher[f'test/{l1}/#'] = f'test/{l1}/#'
|
||||
for l2 in range(1, 11):
|
||||
matcher[f'test/{l1}/{l2}'] = f'test/{l1}/{l2}'
|
||||
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())
|
||||
assert len(match_result) == 120
|
||||
match_result: list[str] = list(matcher.iter_match(topic='test/1/1'))
|
||||
assert len(match_result) == 3
|
||||
assert set(match_result) == set(['test/1/1', 'test/+/1', 'test/1/#'])
|
||||
# Delete
|
||||
if matcher.get(topic='test/1/1'):
|
||||
del matcher['test/1/1']
|
||||
assert len(list(matcher.iter_all_nodes())) == 119
|
||||
match_result: list[str] = list(matcher.iter_match(topic='test/1/1'))
|
||||
assert len(match_result) == 2
|
||||
assert set(match_result) == set(['test/+/1', 'test/1/#'])
|
@ -158,10 +158,10 @@ async def test_user_config_async(
|
||||
config = config_base.copy()
|
||||
assert await storage.update_user_config_async(
|
||||
uid=test_uid, cloud_server=test_cloud_server, config=config)
|
||||
# test load all
|
||||
# Test load all
|
||||
assert (await storage.load_user_config_async(
|
||||
uid=test_uid, cloud_server=test_cloud_server)) == config
|
||||
# test update
|
||||
# Test update
|
||||
config_update = {
|
||||
'test_str': 'test str',
|
||||
'number_float': 456.123
|
||||
@ -171,7 +171,7 @@ async def test_user_config_async(
|
||||
config.update(config_update)
|
||||
assert (await storage.load_user_config_async(
|
||||
uid=test_uid, cloud_server=test_cloud_server)) == config
|
||||
# test replace
|
||||
# Test replace
|
||||
config_replace = None
|
||||
assert await storage.update_user_config_async(
|
||||
uid=test_uid, cloud_server=test_cloud_server,
|
||||
@ -179,9 +179,9 @@ async def test_user_config_async(
|
||||
assert (config_replace := await storage.load_user_config_async(
|
||||
uid=test_uid, cloud_server=test_cloud_server)) == config_update
|
||||
print('replace result, ', config_replace)
|
||||
# test query
|
||||
# Test query
|
||||
query_keys = list(config_base.keys())
|
||||
print('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)
|
||||
print('query result 1, ', query_result)
|
||||
@ -194,18 +194,19 @@ async def test_user_config_async(
|
||||
query_result = await storage.load_user_config_async(
|
||||
uid=test_uid, cloud_server=test_cloud_server)
|
||||
print('query result all, ', query_result)
|
||||
# remove config
|
||||
# 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)
|
||||
# remove domain
|
||||
# Remove domain
|
||||
assert await storage.remove_domain_async(domain='miot_config')
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@pytest.mark.skip(reason='clean')
|
||||
@pytest.mark.dependency()
|
||||
async def test_clear_async(test_cache_path):
|
||||
from miot.miot_storage import MIoTStorage
|
||||
|
||||
|
Loading…
x
Reference in New Issue
Block a user