mirror of
https://github.com/maziggy/bambuddy.git
synced 2026-10-09 07:25:44 +02:00
232 lines
8.0 KiB
Python
232 lines
8.0 KiB
Python
"""Unit tests for REST smart plug service."""
|
|
|
|
import json
|
|
from unittest.mock import AsyncMock, MagicMock, patch
|
|
|
|
import httpx
|
|
import pytest
|
|
|
|
from backend.app.services.rest_smart_plug import RESTSmartPlugService
|
|
|
|
|
|
@pytest.fixture
|
|
def service():
|
|
return RESTSmartPlugService(timeout=5.0)
|
|
|
|
|
|
@pytest.fixture
|
|
def mock_plug():
|
|
plug = MagicMock()
|
|
plug.name = "Test REST Plug"
|
|
plug.plug_type = "rest"
|
|
plug.rest_on_url = "http://192.168.1.50:8080/api/plug/on"
|
|
plug.rest_on_body = '{"state": "on"}'
|
|
plug.rest_off_url = "http://192.168.1.50:8080/api/plug/off"
|
|
plug.rest_off_body = '{"state": "off"}'
|
|
plug.rest_method = "POST"
|
|
plug.rest_headers = '{"Authorization": "Bearer test-token"}'
|
|
plug.rest_status_url = "http://192.168.1.50:8080/api/plug/status"
|
|
plug.rest_status_path = "state"
|
|
plug.rest_status_on_value = "ON"
|
|
plug.rest_power_path = "power"
|
|
plug.rest_energy_path = "energy.today"
|
|
return plug
|
|
|
|
|
|
class TestURLValidation:
|
|
def test_valid_ip_url(self, service):
|
|
assert service._validate_url("http://192.168.1.50:8080/api") is True
|
|
|
|
def test_hostname_url(self, service):
|
|
assert service._validate_url("http://openhab.local:8080/api") is True
|
|
|
|
def test_loopback_blocked(self, service):
|
|
assert service._validate_url("http://127.0.0.1/api") is False
|
|
|
|
def test_link_local_blocked(self, service):
|
|
assert service._validate_url("http://169.254.1.1/api") is False
|
|
|
|
def test_empty_hostname(self, service):
|
|
assert service._validate_url("http:///api") is False
|
|
|
|
|
|
class TestParseHeaders:
|
|
def test_valid_json(self, service):
|
|
headers = service._parse_headers('{"Authorization": "Bearer abc", "X-Custom": "val"}')
|
|
assert headers == {"Authorization": "Bearer abc", "X-Custom": "val"}
|
|
|
|
def test_none_headers(self, service):
|
|
assert service._parse_headers(None) == {}
|
|
|
|
def test_empty_string(self, service):
|
|
assert service._parse_headers("") == {}
|
|
|
|
def test_invalid_json(self, service):
|
|
assert service._parse_headers("not json") == {}
|
|
|
|
|
|
class TestExtractJsonPath:
|
|
def test_simple_path(self, service):
|
|
data = {"state": "ON"}
|
|
assert service._extract_json_path(data, "state") == "ON"
|
|
|
|
def test_nested_path(self, service):
|
|
data = {"data": {"power": {"current": 42.5}}}
|
|
assert service._extract_json_path(data, "data.power.current") == 42.5
|
|
|
|
def test_missing_path(self, service):
|
|
data = {"state": "ON"}
|
|
assert service._extract_json_path(data, "missing") is None
|
|
|
|
def test_empty_path(self, service):
|
|
assert service._extract_json_path({"a": 1}, "") is None
|
|
|
|
def test_none_path(self, service):
|
|
assert service._extract_json_path({"a": 1}, None) is None
|
|
|
|
|
|
class TestTurnOn:
|
|
@pytest.mark.asyncio
|
|
async def test_turn_on_success(self, service, mock_plug):
|
|
mock_response = MagicMock()
|
|
mock_response.status_code = 200
|
|
|
|
with patch.object(service, "_send_request", new_callable=AsyncMock, return_value=mock_response):
|
|
result = await service.turn_on(mock_plug)
|
|
|
|
assert result is True
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_turn_on_failure(self, service, mock_plug):
|
|
with patch.object(service, "_send_request", new_callable=AsyncMock, return_value=None):
|
|
result = await service.turn_on(mock_plug)
|
|
|
|
assert result is False
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_turn_on_no_url(self, service, mock_plug):
|
|
mock_plug.rest_on_url = None
|
|
result = await service.turn_on(mock_plug)
|
|
assert result is False
|
|
|
|
|
|
class TestTurnOff:
|
|
@pytest.mark.asyncio
|
|
async def test_turn_off_success(self, service, mock_plug):
|
|
mock_response = MagicMock()
|
|
mock_response.status_code = 200
|
|
|
|
with patch.object(service, "_send_request", new_callable=AsyncMock, return_value=mock_response):
|
|
result = await service.turn_off(mock_plug)
|
|
|
|
assert result is True
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_turn_off_no_url(self, service, mock_plug):
|
|
mock_plug.rest_off_url = None
|
|
result = await service.turn_off(mock_plug)
|
|
assert result is False
|
|
|
|
|
|
class TestGetStatus:
|
|
@pytest.mark.asyncio
|
|
async def test_status_on(self, service, mock_plug):
|
|
mock_response = MagicMock()
|
|
mock_response.json.return_value = {"state": "ON"}
|
|
|
|
with patch.object(service, "_send_request", new_callable=AsyncMock, return_value=mock_response):
|
|
result = await service.get_status(mock_plug)
|
|
|
|
assert result["state"] == "ON"
|
|
assert result["reachable"] is True
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_status_off(self, service, mock_plug):
|
|
mock_response = MagicMock()
|
|
mock_response.json.return_value = {"state": "OFF"}
|
|
|
|
with patch.object(service, "_send_request", new_callable=AsyncMock, return_value=mock_response):
|
|
result = await service.get_status(mock_plug)
|
|
|
|
assert result["state"] == "OFF"
|
|
assert result["reachable"] is True
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_status_unreachable(self, service, mock_plug):
|
|
with patch.object(service, "_send_request", new_callable=AsyncMock, return_value=None):
|
|
result = await service.get_status(mock_plug)
|
|
|
|
assert result["state"] is None
|
|
assert result["reachable"] is False
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_status_no_url(self, service, mock_plug):
|
|
mock_plug.rest_status_url = None
|
|
result = await service.get_status(mock_plug)
|
|
|
|
assert result["state"] is None
|
|
assert result["reachable"] is True # No URL = assume reachable
|
|
|
|
|
|
class TestGetEnergy:
|
|
@pytest.mark.asyncio
|
|
async def test_energy_with_paths(self, service, mock_plug):
|
|
mock_response = MagicMock()
|
|
mock_response.json.return_value = {"power": 42.5, "energy": {"today": 1.23}}
|
|
|
|
with patch.object(service, "_send_request", new_callable=AsyncMock, return_value=mock_response):
|
|
result = await service.get_energy(mock_plug)
|
|
|
|
assert result["power"] == 42.5
|
|
assert result["today"] == 1.23
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_energy_no_status_url(self, service, mock_plug):
|
|
mock_plug.rest_status_url = None
|
|
result = await service.get_energy(mock_plug)
|
|
assert result is None
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_energy_no_paths(self, service, mock_plug):
|
|
mock_plug.rest_power_path = None
|
|
mock_plug.rest_energy_path = None
|
|
result = await service.get_energy(mock_plug)
|
|
assert result is None
|
|
|
|
|
|
class TestTestConnection:
|
|
@pytest.mark.asyncio
|
|
async def test_connection_success(self, service):
|
|
with patch("httpx.AsyncClient") as mock_client_cls:
|
|
mock_client = AsyncMock()
|
|
mock_response = MagicMock()
|
|
mock_response.raise_for_status = MagicMock()
|
|
mock_client.request = AsyncMock(return_value=mock_response)
|
|
mock_client.__aenter__ = AsyncMock(return_value=mock_client)
|
|
mock_client.__aexit__ = AsyncMock(return_value=None)
|
|
mock_client_cls.return_value = mock_client
|
|
|
|
result = await service.test_connection("http://192.168.1.50:8080/api")
|
|
|
|
assert result["success"] is True
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_connection_timeout(self, service):
|
|
with patch("httpx.AsyncClient") as mock_client_cls:
|
|
mock_client = AsyncMock()
|
|
mock_client.request = AsyncMock(side_effect=httpx.TimeoutException("timeout"))
|
|
mock_client.__aenter__ = AsyncMock(return_value=mock_client)
|
|
mock_client.__aexit__ = AsyncMock(return_value=None)
|
|
mock_client_cls.return_value = mock_client
|
|
|
|
result = await service.test_connection("http://192.168.1.50:8080/api")
|
|
|
|
assert result["success"] is False
|
|
assert "timed out" in result["error"]
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_connection_invalid_url(self, service):
|
|
result = await service.test_connection("http://127.0.0.1/api")
|
|
assert result["success"] is False
|
|
assert "blocked" in result["error"].lower()
|