3ea47d471e
实现设备自动发现规则管理系统，包含以下主要功能： 1. 新增规则管理页面和API，支持创建、编辑和删除主题匹配规则 2. 添加规则匹配引擎，支持+和#通配符匹配设备主题 3. 实现Broker设备注册表，自动发现并管理符合规则的设备 4. 扩展仪表盘显示Broker信息和活跃主题 5. 修改设备卡片和详情页以区分规则发现的设备 6. 添加相关测试用例确保功能稳定性
67 lines
2.2 KiB
Python
67 lines
2.2 KiB
Python
import pytest
|
|
from mqtt_home.rule_registry import list_rules, get_rule, create_rule, update_rule, delete_rule
|
|
from mqtt_home.schemas import RuleCreate, RuleUpdate
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_rule(db_session):
|
|
rule = await create_rule(db_session, RuleCreate(
|
|
name="Test Rule",
|
|
topic_pattern="home/+/state",
|
|
device_type="switch",
|
|
command_template="home/{device_id}/set",
|
|
state_value_path="state",
|
|
))
|
|
assert rule.id is not None
|
|
assert rule.name == "Test Rule"
|
|
assert rule.topic_pattern == "home/+/state"
|
|
assert rule.is_enabled is True
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_list_rules(db_session):
|
|
await create_rule(db_session, RuleCreate(name="Rule 1", topic_pattern="a/+"))
|
|
await create_rule(db_session, RuleCreate(name="Rule 2", topic_pattern="b/+"))
|
|
rules = await list_rules(db_session)
|
|
assert len(rules) == 2
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_rule(db_session):
|
|
created = await create_rule(db_session, RuleCreate(name="Test", topic_pattern="x/+"))
|
|
rule = await get_rule(db_session, created.id)
|
|
assert rule is not None
|
|
assert rule.name == "Test"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_rule_not_found(db_session):
|
|
rule = await get_rule(db_session, 99999)
|
|
assert rule is None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_update_rule(db_session):
|
|
created = await create_rule(db_session, RuleCreate(name="Original", topic_pattern="x/+"))
|
|
updated = await update_rule(db_session, created.id, RuleUpdate(name="Updated"))
|
|
assert updated.name == "Updated"
|
|
assert updated.topic_pattern == "x/+" # unchanged
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_update_rule_not_found(db_session):
|
|
result = await update_rule(db_session, 99999, RuleUpdate(name="X"))
|
|
assert result is None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_delete_rule(db_session):
|
|
created = await create_rule(db_session, RuleCreate(name="To Delete", topic_pattern="x/+"))
|
|
assert await delete_rule(db_session, created.id) is True
|
|
assert await get_rule(db_session, created.id) is None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_delete_rule_not_found(db_session):
|
|
assert await delete_rule(db_session, 99999) is False
|