public_git/bambuddy
1
0
Fork 0
mirror of https://github.com/maziggy/bambuddy.git synced 2026-09-30 03:01:21 +02:00
Code Issues Packages Projects Releases Wiki Activity
Files
8d3abf414be73eeaead228fcf04e11c65390484c
bambuddy/backend/tests/integration/test_smart_plugs_api.py
T

960 lines
37 KiB
Python
Raw Blame History

"""Integration tests for Smart Plugs API endpoints.
Tests the full request/response cycle for /api/v1/smart-plugs/ endpoints.
"""
import pytest
from httpx import AsyncClient
class TestSmartPlugsAPI:
"""Integration tests for /api/v1/smart-plugs/ endpoints."""
# ========================================================================
# List endpoints
# ========================================================================
@pytest.mark.asyncio
@pytest.mark.integration
async def test_list_smart_plugs_empty(self, async_client: AsyncClient):
"""Verify empty list is returned when no plugs exist."""
response = await async_client.get("/api/v1/smart-plugs/")
assert response.status_code == 200
assert response.json() == []
@pytest.mark.asyncio
@pytest.mark.integration
async def test_list_smart_plugs_with_data(self, async_client: AsyncClient, smart_plug_factory, db_session):
"""Verify list returns existing plugs."""
await smart_plug_factory(name="Test Plug 1")
response = await async_client.get("/api/v1/smart-plugs/")
assert response.status_code == 200
data = response.json()
assert len(data) >= 1
assert any(p["name"] == "Test Plug 1" for p in data)
# ========================================================================
# Create endpoints
# ========================================================================
@pytest.mark.asyncio
@pytest.mark.integration
async def test_create_smart_plug(self, async_client: AsyncClient):
"""Verify smart plug can be created."""
data = {
"name": "New Plug",
"ip_address": "192.168.1.100",
"enabled": True,
"auto_on": True,
"auto_off": False,
}
response = await async_client.post("/api/v1/smart-plugs/", json=data)
assert response.status_code == 200
result = response.json()
assert result["name"] == "New Plug"
assert result["ip_address"] == "192.168.1.100"
assert result["auto_off"] is False
@pytest.mark.asyncio
@pytest.mark.integration
async def test_create_smart_plug_with_printer(self, async_client: AsyncClient, printer_factory, db_session):
"""Verify smart plug can be linked to a printer."""
printer = await printer_factory(name="Test Printer")
data = {
"name": "Printer Plug",
"ip_address": "192.168.1.101",
"printer_id": printer.id,
}
response = await async_client.post("/api/v1/smart-plugs/", json=data)
assert response.status_code == 200
result = response.json()
assert result["printer_id"] == printer.id
@pytest.mark.asyncio
@pytest.mark.integration
async def test_create_plug_with_invalid_printer_id(self, async_client: AsyncClient):
"""Verify creating plug with non-existent printer fails."""
data = {
"name": "Test Plug",
"ip_address": "192.168.1.100",
"printer_id": 9999,
}
response = await async_client.post("/api/v1/smart-plugs/", json=data)
assert response.status_code == 400
assert "Printer not found" in response.json()["detail"]
# ========================================================================
# Get single endpoint
# ========================================================================
@pytest.mark.asyncio
@pytest.mark.integration
async def test_get_smart_plug(self, async_client: AsyncClient, smart_plug_factory, db_session):
"""Verify single plug can be retrieved."""
plug = await smart_plug_factory(name="Get Test Plug")
response = await async_client.get(f"/api/v1/smart-plugs/{plug.id}")
assert response.status_code == 200
result = response.json()
assert result["id"] == plug.id
assert result["name"] == "Get Test Plug"
@pytest.mark.asyncio
@pytest.mark.integration
async def test_get_smart_plug_not_found(self, async_client: AsyncClient):
"""Verify 404 for non-existent plug."""
response = await async_client.get("/api/v1/smart-plugs/9999")
assert response.status_code == 404
# ========================================================================
# Update endpoints (CRITICAL - toggle persistence)
# ========================================================================
@pytest.mark.asyncio
@pytest.mark.integration
async def test_update_auto_off_toggle(self, async_client: AsyncClient, smart_plug_factory, db_session):
"""CRITICAL: Verify auto_off toggle persists correctly.
This tests the regression scenario where toggling auto_off
wasn't being saved properly.
"""
# Create plug with auto_off=True
plug = await smart_plug_factory(auto_off=True)
# Verify initial state
response = await async_client.get(f"/api/v1/smart-plugs/{plug.id}")
assert response.status_code == 200
assert response.json()["auto_off"] is True
# Toggle auto_off to False
response = await async_client.patch(f"/api/v1/smart-plugs/{plug.id}", json={"auto_off": False})
assert response.status_code == 200
assert response.json()["auto_off"] is False
# Verify change persisted by fetching again
response = await async_client.get(f"/api/v1/smart-plugs/{plug.id}")
assert response.json()["auto_off"] is False
@pytest.mark.asyncio
@pytest.mark.integration
async def test_update_auto_on_toggle(self, async_client: AsyncClient, smart_plug_factory, db_session):
"""Verify auto_on toggle persists correctly."""
plug = await smart_plug_factory(auto_on=True)
response = await async_client.patch(f"/api/v1/smart-plugs/{plug.id}", json={"auto_on": False})
assert response.status_code == 200
assert response.json()["auto_on"] is False
# Verify persistence
response = await async_client.get(f"/api/v1/smart-plugs/{plug.id}")
assert response.json()["auto_on"] is False
@pytest.mark.asyncio
@pytest.mark.integration
async def test_update_enabled_toggle(self, async_client: AsyncClient, smart_plug_factory, db_session):
"""Verify enabled toggle persists correctly."""
plug = await smart_plug_factory(enabled=True)
response = await async_client.patch(f"/api/v1/smart-plugs/{plug.id}", json={"enabled": False})
assert response.status_code == 200
assert response.json()["enabled"] is False
@pytest.mark.asyncio
@pytest.mark.integration
async def test_update_off_delay_mode(self, async_client: AsyncClient, smart_plug_factory, db_session):
"""Verify off_delay_mode can be changed."""
plug = await smart_plug_factory(off_delay_mode="time")
response = await async_client.patch(
f"/api/v1/smart-plugs/{plug.id}", json={"off_delay_mode": "temperature", "off_temp_threshold": 50}
)
assert response.status_code == 200
result = response.json()
assert result["off_delay_mode"] == "temperature"
assert result["off_temp_threshold"] == 50
@pytest.mark.asyncio
@pytest.mark.integration
async def test_update_schedule_settings(self, async_client: AsyncClient, smart_plug_factory, db_session):
"""Verify schedule settings can be updated."""
plug = await smart_plug_factory(schedule_enabled=False)
response = await async_client.patch(
f"/api/v1/smart-plugs/{plug.id}",
json={
"schedule_enabled": True,
"schedule_on_time": "08:00",
"schedule_off_time": "22:00",
},
)
assert response.status_code == 200
result = response.json()
assert result["schedule_enabled"] is True
assert result["schedule_on_time"] == "08:00"
assert result["schedule_off_time"] == "22:00"
@pytest.mark.asyncio
@pytest.mark.integration
async def test_update_multiple_fields(self, async_client: AsyncClient, smart_plug_factory, db_session):
"""Verify multiple fields can be updated at once."""
plug = await smart_plug_factory(
name="Old Name",
auto_on=True,
auto_off=True,
)
response = await async_client.patch(
f"/api/v1/smart-plugs/{plug.id}",
json={
"name": "New Name",
"auto_on": False,
"auto_off": False,
},
)
assert response.status_code == 200
result = response.json()
assert result["name"] == "New Name"
assert result["auto_on"] is False
assert result["auto_off"] is False
# ========================================================================
# Control endpoints
# ========================================================================
@pytest.mark.asyncio
@pytest.mark.integration
async def test_control_smart_plug_on(
self, async_client: AsyncClient, smart_plug_factory, mock_tasmota_service, db_session
):
"""Verify smart plug can be turned on."""
plug = await smart_plug_factory()
response = await async_client.post(f"/api/v1/smart-plugs/{plug.id}/control", json={"action": "on"})
assert response.status_code == 200
result = response.json()
assert result["success"] is True
assert result["action"] == "on"
@pytest.mark.asyncio
@pytest.mark.integration
async def test_control_smart_plug_off(
self, async_client: AsyncClient, smart_plug_factory, mock_tasmota_service, db_session
):
"""Verify smart plug can be turned off."""
plug = await smart_plug_factory()
response = await async_client.post(f"/api/v1/smart-plugs/{plug.id}/control", json={"action": "off"})
assert response.status_code == 200
result = response.json()
assert result["success"] is True
assert result["action"] == "off"
@pytest.mark.asyncio
@pytest.mark.integration
async def test_control_smart_plug_toggle(
self, async_client: AsyncClient, smart_plug_factory, mock_tasmota_service, db_session
):
"""Verify smart plug can be toggled."""
plug = await smart_plug_factory()
response = await async_client.post(f"/api/v1/smart-plugs/{plug.id}/control", json={"action": "toggle"})
assert response.status_code == 200
result = response.json()
assert result["success"] is True
assert result["action"] == "toggle"
@pytest.mark.asyncio
@pytest.mark.integration
async def test_control_invalid_action(self, async_client: AsyncClient, smart_plug_factory, db_session):
"""Verify invalid action returns error."""
plug = await smart_plug_factory()
response = await async_client.post(f"/api/v1/smart-plugs/{plug.id}/control", json={"action": "invalid"})
# FastAPI returns 422 for pydantic validation errors
assert response.status_code == 422
# ========================================================================
# Status endpoint
# ========================================================================
@pytest.mark.asyncio
@pytest.mark.integration
async def test_get_smart_plug_status(
self, async_client: AsyncClient, smart_plug_factory, mock_tasmota_service, db_session
):
"""Verify smart plug status can be retrieved."""
plug = await smart_plug_factory()
response = await async_client.get(f"/api/v1/smart-plugs/{plug.id}/status")
assert response.status_code == 200
result = response.json()
assert result["state"] == "ON"
assert result["reachable"] is True
# ========================================================================
# Delete endpoint
# ========================================================================
@pytest.mark.asyncio
@pytest.mark.integration
async def test_delete_smart_plug(self, async_client: AsyncClient, smart_plug_factory, db_session):
"""Verify smart plug can be deleted."""
plug = await smart_plug_factory()
plug_id = plug.id
response = await async_client.delete(f"/api/v1/smart-plugs/{plug_id}")
assert response.status_code == 200
# Verify deleted
response = await async_client.get(f"/api/v1/smart-plugs/{plug_id}")
assert response.status_code == 404
@pytest.mark.asyncio
@pytest.mark.integration
async def test_delete_nonexistent_plug(self, async_client: AsyncClient):
"""Verify deleting non-existent plug returns 404."""
response = await async_client.delete("/api/v1/smart-plugs/9999")
assert response.status_code == 404
# ========================================================================
# Switchbar visibility
# ========================================================================
@pytest.mark.asyncio
@pytest.mark.integration
async def test_update_show_in_switchbar(self, async_client: AsyncClient, smart_plug_factory, db_session):
"""Verify show_in_switchbar toggle persists correctly."""
plug = await smart_plug_factory(show_in_switchbar=False)
response = await async_client.patch(f"/api/v1/smart-plugs/{plug.id}", json={"show_in_switchbar": True})
assert response.status_code == 200
assert response.json()["show_in_switchbar"] is True
# Verify persistence
response = await async_client.get(f"/api/v1/smart-plugs/{plug.id}")
assert response.json()["show_in_switchbar"] is True
# ========================================================================
# Tasmota Discovery endpoints
# ========================================================================
@pytest.mark.asyncio
@pytest.mark.integration
async def test_tasmota_discovery_scan(self, async_client: AsyncClient):
"""Verify Tasmota discovery scan can be started."""
response = await async_client.post("/api/v1/smart-plugs/discover/scan")
assert response.status_code == 200
data = response.json()
assert "running" in data
assert "scanned" in data
assert "total" in data
@pytest.mark.asyncio
@pytest.mark.integration
async def test_tasmota_discovery_status(self, async_client: AsyncClient):
"""Verify Tasmota discovery status endpoint works."""
response = await async_client.get("/api/v1/smart-plugs/discover/status")
assert response.status_code == 200
data = response.json()
assert "running" in data
assert "scanned" in data
assert "total" in data
@pytest.mark.asyncio
@pytest.mark.integration
async def test_tasmota_discovery_devices(self, async_client: AsyncClient):
"""Verify Tasmota discovered devices endpoint works."""
response = await async_client.get("/api/v1/smart-plugs/discover/devices")
assert response.status_code == 200
data = response.json()
assert isinstance(data, list)
@pytest.mark.asyncio
@pytest.mark.integration
async def test_tasmota_discovery_stop(self, async_client: AsyncClient):
"""Verify Tasmota discovery can be stopped."""
response = await async_client.post("/api/v1/smart-plugs/discover/stop")
assert response.status_code == 200
data = response.json()
assert "running" in data
# ========================================================================
# Home Assistant Integration tests
# ========================================================================
@pytest.mark.asyncio
@pytest.mark.integration
async def test_create_homeassistant_plug(self, async_client: AsyncClient):
"""Verify Home Assistant plug can be created."""
data = {
"name": "HA Plug",
"plug_type": "homeassistant",
"ha_entity_id": "switch.printer_plug",
"enabled": True,
"auto_on": True,
"auto_off": False,
}
response = await async_client.post("/api/v1/smart-plugs/", json=data)
assert response.status_code == 200
result = response.json()
assert result["name"] == "HA Plug"
assert result["plug_type"] == "homeassistant"
assert result["ha_entity_id"] == "switch.printer_plug"
assert result["ip_address"] is None
@pytest.mark.asyncio
@pytest.mark.integration
async def test_create_homeassistant_plug_missing_entity_id(self, async_client: AsyncClient):
"""Verify creating HA plug without entity_id fails."""
data = {
"name": "HA Plug",
"plug_type": "homeassistant",
# Missing ha_entity_id
"enabled": True,
}
response = await async_client.post("/api/v1/smart-plugs/", json=data)
assert response.status_code == 422 # Validation error
@pytest.mark.asyncio
@pytest.mark.integration
async def test_create_tasmota_plug_missing_ip(self, async_client: AsyncClient):
"""Verify creating Tasmota plug without IP fails."""
data = {
"name": "Tasmota Plug",
"plug_type": "tasmota",
# Missing ip_address
"enabled": True,
}
response = await async_client.post("/api/v1/smart-plugs/", json=data)
assert response.status_code == 422 # Validation error
@pytest.mark.asyncio
@pytest.mark.integration
async def test_ha_entities_endpoint_not_configured(self, async_client: AsyncClient):
"""Verify HA entities endpoint returns error when not configured."""
response = await async_client.get("/api/v1/smart-plugs/ha/entities")
assert response.status_code == 400
assert "not configured" in response.json()["detail"].lower()
@pytest.mark.asyncio
@pytest.mark.integration
async def test_update_plug_type(self, async_client: AsyncClient, smart_plug_factory, db_session):
"""Verify plug_type can be updated."""
plug = await smart_plug_factory(plug_type="tasmota", ip_address="192.168.1.100")
response = await async_client.patch(
f"/api/v1/smart-plugs/{plug.id}",
json={
"plug_type": "homeassistant",
"ha_entity_id": "switch.test",
},
)
assert response.status_code == 200
result = response.json()
assert result["plug_type"] == "homeassistant"
assert result["ha_entity_id"] == "switch.test"
@pytest.mark.asyncio
@pytest.mark.integration
async def test_control_homeassistant_plug(
self, async_client: AsyncClient, smart_plug_factory, mock_homeassistant_service, db_session
):
"""Verify HA smart plug can be controlled."""
plug = await smart_plug_factory(plug_type="homeassistant", ha_entity_id="switch.test")
response = await async_client.post(f"/api/v1/smart-plugs/{plug.id}/control", json={"action": "on"})
assert response.status_code == 200
result = response.json()
assert result["success"] is True
assert result["action"] == "on"
@pytest.mark.asyncio
@pytest.mark.integration
async def test_get_homeassistant_plug_status(
self, async_client: AsyncClient, smart_plug_factory, mock_homeassistant_service, db_session
):
"""Verify HA smart plug status can be retrieved."""
plug = await smart_plug_factory(plug_type="homeassistant", ha_entity_id="switch.test")
response = await async_client.get(f"/api/v1/smart-plugs/{plug.id}/status")
assert response.status_code == 200
result = response.json()
assert result["state"] == "ON"
assert result["reachable"] is True
@pytest.mark.asyncio
@pytest.mark.integration
async def test_create_homeassistant_plug_with_energy_sensors(self, async_client: AsyncClient):
"""Verify HA plug can be created with energy sensor entities."""
data = {
"name": "HA Plug with Energy",
"plug_type": "homeassistant",
"ha_entity_id": "switch.printer_plug",
"ha_power_entity": "sensor.printer_power",
"ha_energy_today_entity": "sensor.printer_energy_today",
"ha_energy_total_entity": "sensor.printer_energy_total",
"enabled": True,
}
response = await async_client.post("/api/v1/smart-plugs/", json=data)
assert response.status_code == 200
result = response.json()
assert result["ha_power_entity"] == "sensor.printer_power"
assert result["ha_energy_today_entity"] == "sensor.printer_energy_today"
assert result["ha_energy_total_entity"] == "sensor.printer_energy_total"
@pytest.mark.asyncio
@pytest.mark.integration
async def test_update_ha_energy_sensor_entities(self, async_client: AsyncClient, smart_plug_factory, db_session):
"""Verify HA energy sensor entities can be updated."""
plug = await smart_plug_factory(plug_type="homeassistant", ha_entity_id="switch.test")
response = await async_client.patch(
f"/api/v1/smart-plugs/{plug.id}",
json={
"ha_power_entity": "sensor.new_power",
"ha_energy_today_entity": "sensor.new_today",
"ha_energy_total_entity": "sensor.new_total",
},
)
assert response.status_code == 200
result = response.json()
assert result["ha_power_entity"] == "sensor.new_power"
assert result["ha_energy_today_entity"] == "sensor.new_today"
assert result["ha_energy_total_entity"] == "sensor.new_total"
@pytest.mark.asyncio
@pytest.mark.integration
async def test_ha_sensors_endpoint_not_configured(self, async_client: AsyncClient):
"""Verify HA sensors endpoint returns error when not configured."""
response = await async_client.get("/api/v1/smart-plugs/ha/sensors")
assert response.status_code == 400
assert "not configured" in response.json()["detail"].lower()
# ========================================================================
# MQTT Integration tests
# ========================================================================
@pytest.mark.asyncio
@pytest.mark.integration
async def test_create_mqtt_plug(self, async_client: AsyncClient, mock_mqtt_smart_plug_service):
"""Verify MQTT plug can be created with topic and JSON paths."""
data = {
"name": "MQTT Energy Monitor",
"plug_type": "mqtt",
"mqtt_topic": "zigbee2mqtt/shelly-working-room",
"mqtt_power_path": "power_l1",
"mqtt_energy_path": "energy_l1",
"mqtt_state_path": "state_l1",
"mqtt_multiplier": 1.0,
"enabled": True,
}
response = await async_client.post("/api/v1/smart-plugs/", json=data)
assert response.status_code == 200
result = response.json()
assert result["name"] == "MQTT Energy Monitor"
assert result["plug_type"] == "mqtt"
assert result["mqtt_topic"] == "zigbee2mqtt/shelly-working-room"
assert result["mqtt_power_path"] == "power_l1"
assert result["mqtt_energy_path"] == "energy_l1"
assert result["mqtt_state_path"] == "state_l1"
assert result["mqtt_multiplier"] == 1.0
assert result["ip_address"] is None
@pytest.mark.asyncio
@pytest.mark.integration
async def test_create_mqtt_plug_missing_topic(self, async_client: AsyncClient):
"""Verify creating MQTT plug without topic fails."""
data = {
"name": "MQTT Plug",
"plug_type": "mqtt",
# Missing mqtt_topic
"mqtt_power_path": "power",
"enabled": True,
}
response = await async_client.post("/api/v1/smart-plugs/", json=data)
assert response.status_code == 422 # Validation error
@pytest.mark.asyncio
@pytest.mark.integration
async def test_create_mqtt_plug_missing_topic(self, async_client: AsyncClient):
"""Verify creating MQTT plug without any topic fails."""
data = {
"name": "MQTT Plug",
"plug_type": "mqtt",
# No topic configured at all
"enabled": True,
}
response = await async_client.post("/api/v1/smart-plugs/", json=data)
assert response.status_code == 422 # Validation error
@pytest.mark.asyncio
@pytest.mark.integration
async def test_create_mqtt_plug_with_multiplier(self, async_client: AsyncClient, mock_mqtt_smart_plug_service):
"""Verify MQTT plug can use multiplier for unit conversion."""
data = {
"name": "MQTT mW to W",
"plug_type": "mqtt",
"mqtt_topic": "sensors/power",
"mqtt_power_path": "power_mw",
"mqtt_multiplier": 0.001, # Convert mW to W
"enabled": True,
}
response = await async_client.post("/api/v1/smart-plugs/", json=data)
assert response.status_code == 200
result = response.json()
assert result["mqtt_multiplier"] == 0.001
@pytest.mark.asyncio
@pytest.mark.integration
async def test_control_mqtt_plug_returns_error(self, async_client: AsyncClient, smart_plug_factory, db_session):
"""Verify MQTT plugs cannot be controlled (monitor-only)."""
plug = await smart_plug_factory(
plug_type="mqtt",
mqtt_topic="test/topic",
mqtt_power_path="power",
)
response = await async_client.post(f"/api/v1/smart-plugs/{plug.id}/control", json={"action": "on"})
assert response.status_code == 400
assert "monitor-only" in response.json()["detail"].lower()
@pytest.mark.asyncio
@pytest.mark.integration
async def test_update_mqtt_plug_topic(self, async_client: AsyncClient, smart_plug_factory, db_session):
"""Verify MQTT plug topic can be updated."""
plug = await smart_plug_factory(
plug_type="mqtt",
mqtt_topic="old/topic",
mqtt_power_path="power",
)
response = await async_client.patch(
f"/api/v1/smart-plugs/{plug.id}",
json={
"mqtt_topic": "new/topic",
"mqtt_power_path": "new_power",
},
)
assert response.status_code == 200
result = response.json()
assert result["mqtt_topic"] == "new/topic"
assert result["mqtt_power_path"] == "new_power"
# ========================================================================
# Enhanced MQTT Integration tests (separate topics per data type)
# ========================================================================
@pytest.mark.asyncio
@pytest.mark.integration
async def test_create_mqtt_plug_with_separate_topics(self, async_client: AsyncClient, mock_mqtt_smart_plug_service):
"""Verify MQTT plug can be created with separate topics for power, energy, and state."""
data = {
"name": "MQTT Separate Topics",
"plug_type": "mqtt",
"mqtt_power_topic": "zigbee/power",
"mqtt_power_path": "power_l1",
"mqtt_power_multiplier": 0.001,
"mqtt_energy_topic": "zigbee/energy",
"mqtt_energy_path": "energy_total",
"mqtt_energy_multiplier": 1.0,
"mqtt_state_topic": "zigbee/state",
"mqtt_state_path": "state",
"mqtt_state_on_value": "ON",
"enabled": True,
}
response = await async_client.post("/api/v1/smart-plugs/", json=data)
assert response.status_code == 200
result = response.json()
assert result["name"] == "MQTT Separate Topics"
assert result["plug_type"] == "mqtt"
# Power fields
assert result["mqtt_power_topic"] == "zigbee/power"
assert result["mqtt_power_path"] == "power_l1"
assert result["mqtt_power_multiplier"] == 0.001
# Energy fields
assert result["mqtt_energy_topic"] == "zigbee/energy"
assert result["mqtt_energy_path"] == "energy_total"
assert result["mqtt_energy_multiplier"] == 1.0
# State fields
assert result["mqtt_state_topic"] == "zigbee/state"
assert result["mqtt_state_path"] == "state"
assert result["mqtt_state_on_value"] == "ON"
@pytest.mark.asyncio
@pytest.mark.integration
async def test_create_mqtt_plug_energy_only(self, async_client: AsyncClient, mock_mqtt_smart_plug_service):
"""Verify MQTT plug can be created with only energy monitoring."""
data = {
"name": "Energy Only Monitor",
"plug_type": "mqtt",
"mqtt_energy_topic": "sensors/energy",
"mqtt_energy_path": "kwh",
"mqtt_energy_multiplier": 0.001, # Wh to kWh
"enabled": True,
}
response = await async_client.post("/api/v1/smart-plugs/", json=data)
assert response.status_code == 200
result = response.json()
assert result["mqtt_energy_topic"] == "sensors/energy"
assert result["mqtt_energy_path"] == "kwh"
assert result["mqtt_energy_multiplier"] == 0.001
@pytest.mark.asyncio
@pytest.mark.integration
async def test_create_mqtt_plug_state_only(self, async_client: AsyncClient, mock_mqtt_smart_plug_service):
"""Verify MQTT plug can be created with only state monitoring."""
data = {
"name": "State Only Monitor",
"plug_type": "mqtt",
"mqtt_state_topic": "switches/outlet",
"mqtt_state_path": "state",
"mqtt_state_on_value": "true",
"enabled": True,
}
response = await async_client.post("/api/v1/smart-plugs/", json=data)
assert response.status_code == 200
result = response.json()
assert result["mqtt_state_topic"] == "switches/outlet"
assert result["mqtt_state_path"] == "state"
assert result["mqtt_state_on_value"] == "true"
@pytest.mark.asyncio
@pytest.mark.integration
async def test_create_mqtt_plug_topic_only_succeeds(self, async_client: AsyncClient, mock_mqtt_smart_plug_service):
"""Verify creating MQTT plug with topic only (no path) succeeds for raw values."""
data = {
"name": "Raw MQTT Plug",
"plug_type": "mqtt",
# Topic only, no path - valid for raw numeric MQTT values
"mqtt_power_topic": "zigbee/power",
"enabled": True,
}
response = await async_client.post("/api/v1/smart-plugs/", json=data)
assert response.status_code == 200 # Should succeed
result = response.json()
assert result["mqtt_power_topic"] == "zigbee/power"
assert result["mqtt_power_path"] is None
@pytest.mark.asyncio
@pytest.mark.integration
async def test_update_mqtt_plug_separate_multipliers(
self, async_client: AsyncClient, smart_plug_factory, db_session, mock_mqtt_smart_plug_service
):
"""Verify MQTT plug multipliers can be updated separately."""
plug = await smart_plug_factory(
plug_type="mqtt",
mqtt_power_topic="test/power",
mqtt_power_path="power",
mqtt_power_multiplier=1.0,
mqtt_energy_topic="test/energy",
mqtt_energy_path="energy",
mqtt_energy_multiplier=1.0,
)
response = await async_client.patch(
f"/api/v1/smart-plugs/{plug.id}",
json={
"mqtt_power_multiplier": 0.001, # Change power multiplier only
"mqtt_energy_multiplier": 0.001, # Change energy multiplier only
},
)
assert response.status_code == 200
result = response.json()
assert result["mqtt_power_multiplier"] == 0.001
assert result["mqtt_energy_multiplier"] == 0.001
# ========================================================================
# REST Smart Plug Integration tests
# ========================================================================
@pytest.mark.asyncio
@pytest.mark.integration
async def test_create_rest_plug(self, async_client: AsyncClient):
"""Verify REST plug can be created with ON/OFF URLs."""
data = {
"name": "REST Plug",
"plug_type": "rest",
"rest_on_url": "http://openhab:8080/rest/items/MyPlug",
"rest_on_body": "ON",
"rest_off_url": "http://openhab:8080/rest/items/MyPlug",
"rest_off_body": "OFF",
"rest_method": "POST",
"enabled": True,
}
response = await async_client.post("/api/v1/smart-plugs/", json=data)
assert response.status_code == 200
result = response.json()
assert result["name"] == "REST Plug"
assert result["plug_type"] == "rest"
assert result["rest_on_url"] == "http://openhab:8080/rest/items/MyPlug"
assert result["rest_on_body"] == "ON"
assert result["rest_off_url"] == "http://openhab:8080/rest/items/MyPlug"
assert result["rest_off_body"] == "OFF"
assert result["rest_method"] == "POST"
assert result["ip_address"] is None
assert result["ha_entity_id"] is None
@pytest.mark.asyncio
@pytest.mark.integration
async def test_create_rest_plug_on_url_only(self, async_client: AsyncClient):
"""Verify REST plug can be created with only ON URL."""
data = {
"name": "REST ON Only",
"plug_type": "rest",
"rest_on_url": "http://iobroker:8087/set/plug?value=true",
"rest_method": "GET",
"enabled": True,
}
response = await async_client.post("/api/v1/smart-plugs/", json=data)
assert response.status_code == 200
result = response.json()
assert result["rest_on_url"] == "http://iobroker:8087/set/plug?value=true"
assert result["rest_off_url"] is None
assert result["rest_method"] == "GET"
@pytest.mark.asyncio
@pytest.mark.integration
async def test_create_rest_plug_missing_urls_fails(self, async_client: AsyncClient):
"""Verify creating REST plug without any URL fails."""
data = {
"name": "REST Plug",
"plug_type": "rest",
"enabled": True,
}
response = await async_client.post("/api/v1/smart-plugs/", json=data)
assert response.status_code == 422 # Validation error
@pytest.mark.asyncio
@pytest.mark.integration
async def test_create_rest_plug_with_status_and_energy(self, async_client: AsyncClient):
"""Verify REST plug with status polling and energy monitoring."""
data = {
"name": "REST Full",
"plug_type": "rest",
"rest_on_url": "http://ha:8080/api/plug/on",
"rest_off_url": "http://ha:8080/api/plug/off",
"rest_method": "POST",
"rest_headers": '{"Authorization": "Bearer test-token"}',
"rest_status_url": "http://ha:8080/api/plug/status",
"rest_status_path": "state",
"rest_status_on_value": "ON",
"rest_power_path": "power.current",
"rest_energy_path": "energy.today",
"enabled": True,
}
response = await async_client.post("/api/v1/smart-plugs/", json=data)
assert response.status_code == 200
result = response.json()
assert result["rest_headers"] == '{"Authorization": "Bearer test-token"}'
assert result["rest_status_url"] == "http://ha:8080/api/plug/status"
assert result["rest_status_path"] == "state"
assert result["rest_status_on_value"] == "ON"
assert result["rest_power_path"] == "power.current"
assert result["rest_energy_path"] == "energy.today"
@pytest.mark.asyncio
@pytest.mark.integration
async def test_update_rest_plug(self, async_client: AsyncClient, smart_plug_factory, db_session):
"""Verify REST plug fields can be updated."""
plug = await smart_plug_factory(plug_type="rest")
response = await async_client.patch(
f"/api/v1/smart-plugs/{plug.id}",
json={
"rest_on_url": "http://new-host:8080/on",
"rest_method": "PUT",
},
)
assert response.status_code == 200
result = response.json()
assert result["rest_on_url"] == "http://new-host:8080/on"
assert result["rest_method"] == "PUT"
@pytest.mark.asyncio
@pytest.mark.integration
async def test_rest_plug_is_controllable(self, async_client: AsyncClient, smart_plug_factory, db_session):
"""Verify REST plugs can be controlled (not monitor-only like MQTT)."""
plug = await smart_plug_factory(plug_type="rest")
# REST plugs should NOT return 400 like MQTT plugs
response = await async_client.post(
f"/api/v1/smart-plugs/{plug.id}/control",
json={"action": "on"},
)
# Should attempt to send the request (may fail with 503 since URL is not real,
# but should NOT return 400 "monitor-only")
assert response.status_code != 400
Reference in New Issue View Git Blame Copy Permalink