79b2878b3d
🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude <noreply@anthropic.com>
56 lines
1.7 KiB
Python
56 lines
1.7 KiB
Python
import pytest
|
|
import httpx
|
|
from unittest.mock import AsyncMock, patch
|
|
|
|
from mqtt_home.emqx_api import EmqxApiClient
|
|
from mqtt_home.config import Settings
|
|
|
|
|
|
@pytest.fixture
|
|
def settings():
|
|
return Settings(
|
|
mqtt_host="localhost",
|
|
emqx_api_url="http://localhost:18083/api/v5",
|
|
emqx_api_key="test-key",
|
|
emqx_api_secret="test-secret",
|
|
database_url="sqlite+aiosqlite:///:memory:",
|
|
)
|
|
|
|
|
|
@pytest.fixture
|
|
def emqx_client(settings):
|
|
return EmqxApiClient(settings)
|
|
|
|
|
|
def _make_response(status_code: int, json_data: dict) -> httpx.Response:
|
|
request = httpx.Request("GET", "http://localhost:18083/api/v5/test")
|
|
return httpx.Response(status_code, json=json_data, request=request)
|
|
|
|
|
|
async def test_get_broker_status(emqx_client):
|
|
with patch.object(emqx_client._client, "get", new_callable=AsyncMock) as mock_get:
|
|
mock_get.return_value = _make_response(
|
|
200, {"data": {"uptime": 12345, "version": "5.0.0"}}
|
|
)
|
|
result = await emqx_client.get_broker_status()
|
|
assert result["uptime"] == 12345
|
|
|
|
|
|
async def test_get_clients(emqx_client):
|
|
with patch.object(emqx_client._client, "get", new_callable=AsyncMock) as mock_get:
|
|
mock_get.return_value = _make_response(
|
|
200, {"data": [{"clientid": "dev1", "connected": True}]}
|
|
)
|
|
result = await emqx_client.get_clients()
|
|
assert len(result) == 1
|
|
assert result[0]["clientid"] == "dev1"
|
|
|
|
|
|
async def test_get_topics(emqx_client):
|
|
with patch.object(emqx_client._client, "get", new_callable=AsyncMock) as mock_get:
|
|
mock_get.return_value = _make_response(
|
|
200, {"data": [{"topic": "home/light"}]}
|
|
)
|
|
result = await emqx_client.get_topics()
|
|
assert len(result) == 1
|