afe9de51c5
🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude <noreply@anthropic.com>
49 lines
1.6 KiB
Python
49 lines
1.6 KiB
Python
import pytest
|
|
from mqtt_home.mqtt_client import MqttClient
|
|
from mqtt_home.config import Settings
|
|
|
|
|
|
@pytest.fixture
|
|
def mqtt_client():
|
|
settings = Settings(
|
|
mqtt_host="localhost",
|
|
mqtt_port=1883,
|
|
emqx_api_url="http://localhost:18083/api/v5",
|
|
emqx_api_key="test-key",
|
|
emqx_api_secret="test-secret",
|
|
database_url="sqlite+aiosqlite:///:memory:",
|
|
)
|
|
return MqttClient(settings)
|
|
|
|
|
|
def test_topic_matches_exact():
|
|
assert MqttClient._topic_matches("home/light", "home/light") is True
|
|
assert MqttClient._topic_matches("home/light", "home/switch") is False
|
|
|
|
|
|
def test_topic_matches_single_level_wildcard():
|
|
assert MqttClient._topic_matches("home/light/status", "home/+/status") is True
|
|
assert MqttClient._topic_matches("home/switch/status", "home/+/status") is True
|
|
assert MqttClient._topic_matches("home/light", "home/+") is True
|
|
assert MqttClient._topic_matches("home/light/extra", "home/+") is False
|
|
|
|
|
|
def test_topic_matches_multi_level_wildcard():
|
|
assert MqttClient._topic_matches("homeassistant/light/abc/config", "homeassistant/#") is True
|
|
assert MqttClient._topic_matches("homeassistant/light/abc/config", "homeassistant/light/#") is True
|
|
assert MqttClient._topic_matches("other/topic", "homeassistant/#") is False
|
|
|
|
|
|
def test_register_callback(mqtt_client):
|
|
async def cb(topic, payload):
|
|
pass
|
|
|
|
mqtt_client.on_message("home/#", cb)
|
|
assert "home/#" in mqtt_client._callbacks
|
|
assert len(mqtt_client._callbacks["home/#"]) == 1
|
|
|
|
|
|
def test_initial_state(mqtt_client):
|
|
assert mqtt_client.is_connected is False
|
|
assert len(mqtt_client._client_id) > 0
|