b934600380
- Sync and async HTTP clients for Ignis Core WiZ server - 23 endpoints: auth, devices, groups, control, schedules, stats, API keys - Pydantic models with client-side validation - 108 unit tests - README with role table and usage examples
923 lines
33 KiB
Python
923 lines
33 KiB
Python
import json
|
|
import sys
|
|
import os
|
|
import unittest
|
|
from datetime import datetime, timedelta, timezone
|
|
|
|
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
|
|
|
from unittest.mock import AsyncMock, MagicMock, patch
|
|
|
|
from ignis_client.models import (
|
|
IgnisError,
|
|
CommandRequest,
|
|
ScheduleOnceRequest,
|
|
ScheduleCronRequest,
|
|
DeviceControlResponse,
|
|
GroupControlResponse,
|
|
GroupCommandResult,
|
|
RescanResponse,
|
|
ScheduleCreateResponse,
|
|
ScheduleTasksResponse,
|
|
ScheduleTaskItem,
|
|
ServerInfoResponse,
|
|
KeyActionRequest,
|
|
SCENES,
|
|
)
|
|
from ignis_client.sync import IgnisClient
|
|
from ignis_client.async_client import AsyncIgnisClient
|
|
|
|
|
|
class _FakeResponse:
|
|
def __init__(self, status_code, json_body):
|
|
self.status_code = status_code
|
|
self._json = json_body
|
|
|
|
def json(self):
|
|
return self._json
|
|
|
|
@property
|
|
def text(self):
|
|
return str(self._json)
|
|
|
|
|
|
def _make_response(status_code, json_body, headers=None):
|
|
return _FakeResponse(status_code, json_body)
|
|
resp = MagicMock()
|
|
resp.status_code = status_code
|
|
resp.json.return_value = json_body
|
|
resp.headers = headers or {}
|
|
return resp
|
|
|
|
|
|
class CommandRequestTests(unittest.TestCase):
|
|
def test_state_only(self):
|
|
cmd = CommandRequest(state=True)
|
|
self.assertEqual(cmd.to_wiz_params(), {"state": True})
|
|
|
|
def test_state_off(self):
|
|
cmd = CommandRequest(state=False)
|
|
self.assertEqual(cmd.to_wiz_params(), {"state": False})
|
|
|
|
def test_brightness(self):
|
|
cmd = CommandRequest(brightness=50)
|
|
self.assertEqual(cmd.to_wiz_params(), {"dimming": 50})
|
|
|
|
def test_state_and_brightness(self):
|
|
cmd = CommandRequest(state=True, brightness=80)
|
|
self.assertEqual(cmd.to_wiz_params(), {"state": True, "dimming": 80})
|
|
|
|
def test_scene(self):
|
|
cmd = CommandRequest(scene="cozy")
|
|
self.assertEqual(cmd.to_wiz_params(), {"sceneId": 6})
|
|
|
|
def test_scene_fireplace(self):
|
|
cmd = CommandRequest(scene="fireplace")
|
|
self.assertEqual(cmd.to_wiz_params(), {"sceneId": 5})
|
|
|
|
def test_temp(self):
|
|
cmd = CommandRequest(temp=3200)
|
|
self.assertEqual(cmd.to_wiz_params(), {"temp": 3200})
|
|
|
|
def test_temp_min(self):
|
|
cmd = CommandRequest(temp=2200)
|
|
self.assertEqual(cmd.to_wiz_params(), {"temp": 2200})
|
|
|
|
def test_temp_max(self):
|
|
cmd = CommandRequest(temp=6500)
|
|
self.assertEqual(cmd.to_wiz_params(), {"temp": 6500})
|
|
|
|
def test_rgb(self):
|
|
cmd = CommandRequest(r=255, g=180, b=120)
|
|
self.assertEqual(cmd.to_wiz_params(), {"r": 255, "g": 180, "b": 120})
|
|
|
|
def test_rgb_zero(self):
|
|
cmd = CommandRequest(r=0, g=0, b=0)
|
|
self.assertEqual(cmd.to_wiz_params(), {"r": 0, "g": 0, "b": 0})
|
|
|
|
def test_empty_rejected(self):
|
|
with self.assertRaises(Exception) as ctx:
|
|
CommandRequest()
|
|
self.assertIn("Никаких команд не передано", str(ctx.exception))
|
|
|
|
def test_partial_rgb_rejected(self):
|
|
with self.assertRaises(Exception) as ctx:
|
|
CommandRequest(r=255, g=128)
|
|
self.assertIn("r, g и b нужно передавать вместе", str(ctx.exception))
|
|
|
|
def test_partial_rgb_single_channel_rejected(self):
|
|
with self.assertRaises(Exception):
|
|
CommandRequest(r=255)
|
|
|
|
def test_scene_temp_conflict(self):
|
|
with self.assertRaises(Exception) as ctx:
|
|
CommandRequest(scene="party", temp=3200)
|
|
self.assertIn("только один режим", str(ctx.exception))
|
|
|
|
def test_scene_rgb_conflict(self):
|
|
with self.assertRaises(Exception):
|
|
CommandRequest(scene="ocean", r=0, g=0, b=0)
|
|
|
|
def test_temp_rgb_conflict(self):
|
|
with self.assertRaises(Exception):
|
|
CommandRequest(temp=3000, r=255, g=0, b=0)
|
|
|
|
def test_all_three_conflict(self):
|
|
with self.assertRaises(Exception):
|
|
CommandRequest(scene="ocean", temp=3000, r=255, g=0, b=0)
|
|
|
|
def test_brightness_at_min(self):
|
|
cmd = CommandRequest(brightness=10)
|
|
self.assertEqual(cmd.to_wiz_params(), {"dimming": 10})
|
|
|
|
def test_brightness_at_max(self):
|
|
cmd = CommandRequest(brightness=100)
|
|
self.assertEqual(cmd.to_wiz_params(), {"dimming": 100})
|
|
|
|
def test_brightness_below_min(self):
|
|
with self.assertRaises(Exception):
|
|
CommandRequest(brightness=9)
|
|
|
|
def test_brightness_above_max(self):
|
|
with self.assertRaises(Exception):
|
|
CommandRequest(brightness=101)
|
|
|
|
def test_temp_below_min(self):
|
|
with self.assertRaises(Exception):
|
|
CommandRequest(temp=2199)
|
|
|
|
def test_temp_above_max(self):
|
|
with self.assertRaises(Exception):
|
|
CommandRequest(temp=6501)
|
|
|
|
def test_r_channel_below_min(self):
|
|
with self.assertRaises(Exception):
|
|
CommandRequest(r=-1, g=0, b=0)
|
|
|
|
def test_r_channel_above_max(self):
|
|
with self.assertRaises(Exception):
|
|
CommandRequest(r=256, g=0, b=0)
|
|
|
|
def test_extra_field_forbidden(self):
|
|
with self.assertRaises(Exception):
|
|
CommandRequest(state=True, bogus=42)
|
|
|
|
def test_unknown_scene_in_to_wiz_params(self):
|
|
cmd = CommandRequest(scene="bullshit")
|
|
with self.assertRaises(ValueError) as ctx:
|
|
cmd.to_wiz_params()
|
|
self.assertIn("Неизвестная сцена", str(ctx.exception))
|
|
|
|
def test_has_rgb_property_true(self):
|
|
self.assertTrue(CommandRequest(r=1, g=2, b=3).has_rgb)
|
|
|
|
def test_has_rgb_property_false(self):
|
|
self.assertFalse(CommandRequest(state=True).has_rgb)
|
|
|
|
def test_model_dump_excludes_none(self):
|
|
cmd = CommandRequest(state=True)
|
|
self.assertEqual(cmd.model_dump(exclude_none=True), {"state": True})
|
|
|
|
def test_model_dump_excludes_none_with_brightness(self):
|
|
cmd = CommandRequest(state=False, brightness=40)
|
|
self.assertEqual(
|
|
cmd.model_dump(exclude_none=True), {"state": False, "brightness": 40}
|
|
)
|
|
|
|
def test_all_scenes_have_valid_ids(self):
|
|
self.assertEqual(len(SCENES), 31)
|
|
for name, sid in SCENES.items():
|
|
self.assertIsInstance(name, str)
|
|
self.assertIsInstance(sid, int)
|
|
self.assertGreater(sid, 0)
|
|
self.assertLessEqual(sid, 35)
|
|
cmd = CommandRequest(scene=name)
|
|
self.assertEqual(cmd.to_wiz_params(), {"sceneId": sid})
|
|
|
|
|
|
class ScheduleOnceRequestTests(unittest.TestCase):
|
|
def test_both_run_at_and_hours_from_now(self):
|
|
with self.assertRaises(Exception) as ctx:
|
|
ScheduleOnceRequest(
|
|
target_id="grp-1",
|
|
run_at=datetime.now(timezone.utc),
|
|
hours_from_now=2,
|
|
state=True,
|
|
)
|
|
self.assertIn("ровно одно", str(ctx.exception))
|
|
|
|
def test_neither(self):
|
|
with self.assertRaises(Exception) as ctx:
|
|
ScheduleOnceRequest(target_id="grp-1", state=True)
|
|
self.assertIn("ровно одно", str(ctx.exception))
|
|
|
|
def test_with_run_at(self):
|
|
ts = datetime.now(timezone.utc) + timedelta(hours=1)
|
|
req = ScheduleOnceRequest(target_id="grp-1", run_at=ts, state=True)
|
|
self.assertEqual(req.target_id, "grp-1")
|
|
self.assertEqual(req.run_at, ts)
|
|
self.assertIsNone(req.hours_from_now)
|
|
|
|
def test_with_hours_from_now(self):
|
|
req = ScheduleOnceRequest(
|
|
target_id="dev-1", hours_from_now=3, is_group=False, brightness=50
|
|
)
|
|
self.assertEqual(req.target_id, "dev-1")
|
|
self.assertEqual(req.hours_from_now, 3)
|
|
self.assertFalse(req.is_group)
|
|
|
|
def test_hours_from_now_negative(self):
|
|
with self.assertRaises(Exception):
|
|
ScheduleOnceRequest(target_id="grp-1", hours_from_now=-1, state=True)
|
|
|
|
def test_empty_target_id_rejected(self):
|
|
with self.assertRaises(Exception):
|
|
ScheduleOnceRequest(
|
|
target_id="",
|
|
run_at=datetime.now(timezone.utc) + timedelta(hours=1),
|
|
state=True,
|
|
)
|
|
|
|
def test_inherits_command_validation(self):
|
|
with self.assertRaises(Exception):
|
|
ScheduleOnceRequest(
|
|
target_id="grp-1", hours_from_now=1, scene="party", temp=3000
|
|
)
|
|
|
|
def test_model_dump_includes_schedule_fields(self):
|
|
ts = datetime(2026, 6, 1, 12, 0, 0, tzinfo=timezone.utc)
|
|
req = ScheduleOnceRequest(target_id="grp-1", run_at=ts, state=True)
|
|
dumped = req.model_dump(exclude_none=True)
|
|
self.assertEqual(dumped["target_id"], "grp-1")
|
|
self.assertEqual(dumped["state"], True)
|
|
|
|
|
|
class ScheduleCronRequestTests(unittest.TestCase):
|
|
def test_basic_cron(self):
|
|
req = ScheduleCronRequest(target_id="grp-1", hour="8", minute="30", state=True)
|
|
self.assertEqual(req.hour, "8")
|
|
self.assertEqual(req.minute, "30")
|
|
self.assertEqual(req.day_of_week, "*")
|
|
self.assertTrue(req.is_group)
|
|
|
|
def test_cron_with_day_of_week(self):
|
|
req = ScheduleCronRequest(
|
|
target_id="grp-1",
|
|
hour="18",
|
|
minute="0",
|
|
day_of_week="mon-fri",
|
|
scene="cozy",
|
|
)
|
|
self.assertEqual(req.day_of_week, "mon-fri")
|
|
|
|
def test_empty_hour_rejected(self):
|
|
with self.assertRaises(Exception):
|
|
ScheduleCronRequest(target_id="grp-1", hour="", minute="30", state=True)
|
|
|
|
def test_empty_minute_rejected(self):
|
|
with self.assertRaises(Exception):
|
|
ScheduleCronRequest(target_id="grp-1", hour="8", minute="", state=True)
|
|
|
|
def test_empty_target_id_rejected(self):
|
|
with self.assertRaises(Exception):
|
|
ScheduleCronRequest(target_id="", hour="8", minute="30", state=True)
|
|
|
|
|
|
class IgnisErrorTests(unittest.TestCase):
|
|
def test_construction(self):
|
|
err = IgnisError(503, "Сервер не настроен")
|
|
self.assertEqual(err.status_code, 503)
|
|
self.assertEqual(err.detail, "Сервер не настроен")
|
|
|
|
def test_string_representation(self):
|
|
err = IgnisError(404, "Лампа не в сети")
|
|
self.assertIn("404", str(err))
|
|
self.assertIn("Лампа не в сети", str(err))
|
|
|
|
def test_is_exception(self):
|
|
err = IgnisError(403, "bad key")
|
|
self.assertIsInstance(err, Exception)
|
|
|
|
|
|
class ResponseModelTests(unittest.TestCase):
|
|
def test_device_control_response(self):
|
|
data = {"device_id": "aa:bb:cc:dd", "applied": {"state": True}, "status": "ok"}
|
|
resp = DeviceControlResponse(**data)
|
|
self.assertEqual(resp.device_id, "aa:bb:cc:dd")
|
|
self.assertIsNone(resp.result)
|
|
|
|
def test_device_control_response_with_result(self):
|
|
data = {
|
|
"device_id": "aa:bb:cc:dd",
|
|
"applied": {"state": True},
|
|
"result": {"mac": "aa:bb:cc:dd", "rssi": -42},
|
|
"status": "ok",
|
|
}
|
|
resp = DeviceControlResponse(**data)
|
|
self.assertEqual(resp.result["mac"], "aa:bb:cc:dd")
|
|
|
|
def test_group_control_response_ok(self):
|
|
data = {
|
|
"status": "ok",
|
|
"applied": {"state": True},
|
|
"sent_to": ["10.0.0.1", "10.0.0.2"],
|
|
"success_count": 2,
|
|
"failure_count": 0,
|
|
"results": [
|
|
{"ip": "10.0.0.1", "ok": True, "kind": "ok", "result": {"mac": "aa"}},
|
|
{"ip": "10.0.0.2", "ok": True, "kind": "ok", "result": {"mac": "bb"}},
|
|
],
|
|
}
|
|
resp = GroupControlResponse(**data)
|
|
self.assertEqual(resp.status, "ok")
|
|
self.assertEqual(resp.success_count, 2)
|
|
self.assertEqual(len(resp.results), 2)
|
|
|
|
def test_group_control_response_partial(self):
|
|
data = {
|
|
"status": "partial",
|
|
"applied": {"temp": 3200},
|
|
"sent_to": ["10.0.0.1", "10.0.0.2"],
|
|
"success_count": 1,
|
|
"failure_count": 1,
|
|
"results": [
|
|
{"ip": "10.0.0.1", "ok": True, "kind": "ok", "result": {}},
|
|
{
|
|
"ip": "10.0.0.2",
|
|
"ok": False,
|
|
"kind": "timeout",
|
|
"result": None,
|
|
"error": "timeout",
|
|
},
|
|
],
|
|
}
|
|
resp = GroupControlResponse(**data)
|
|
self.assertEqual(resp.status, "partial")
|
|
self.assertEqual(resp.failure_count, 1)
|
|
self.assertFalse(resp.results[1].ok)
|
|
|
|
def test_rescan_response(self):
|
|
data = {
|
|
"status": "ok",
|
|
"found": 5,
|
|
"added": 2,
|
|
"updated": 3,
|
|
"removed_offline": 1,
|
|
"pending_removal": 0,
|
|
"online": 4,
|
|
}
|
|
resp = RescanResponse(**data)
|
|
self.assertEqual(resp.found, 5)
|
|
self.assertEqual(resp.online, 4)
|
|
|
|
def test_schedule_tasks_response(self):
|
|
data = {
|
|
"tasks": [
|
|
{
|
|
"id": "cron_abc123",
|
|
"target_id": "grp-1",
|
|
"is_group": True,
|
|
"state": True,
|
|
"action_params": {"state": True},
|
|
"trigger_type": "cron",
|
|
"next_run": "2026-06-01T08:00:00",
|
|
"hour": "8",
|
|
"minute": "0",
|
|
"day_of_week": "*",
|
|
"job_present": True,
|
|
}
|
|
]
|
|
}
|
|
resp = ScheduleTasksResponse(**data)
|
|
self.assertEqual(len(resp.tasks), 1)
|
|
self.assertEqual(resp.tasks[0].target_id, "grp-1")
|
|
|
|
def test_server_info_response_diagnostics_visible(self):
|
|
data = {
|
|
"app_name": "Ignis Core",
|
|
"instance_name": "Home",
|
|
"timezone": "Asia/Novosibirsk",
|
|
"uptime_seconds": 3600,
|
|
"diagnostics_visible": True,
|
|
"started_at": "2026-05-01T00:00:00",
|
|
"build": {"version": "1.0.0", "git_sha": None, "build_date": None},
|
|
"urls": {
|
|
"observed_base_url": None,
|
|
"configured_public_base_url": None,
|
|
"effective_public_base_url": None,
|
|
},
|
|
"configuration": {
|
|
"configured": True,
|
|
"master_key_configured": True,
|
|
"scan_network_configured": False,
|
|
"public_base_url_configured": False,
|
|
"build_metadata_complete": False,
|
|
},
|
|
"discovery": {
|
|
"online": 3,
|
|
"last_scan_at": "2026-05-01T00:00:00",
|
|
"last_scan_mode": "startup",
|
|
},
|
|
}
|
|
resp = ServerInfoResponse(**data)
|
|
self.assertEqual(resp.app_name, "Ignis Core")
|
|
self.assertTrue(resp.diagnostics_visible)
|
|
self.assertIsNotNone(resp.build)
|
|
self.assertEqual(resp.discovery.online, 3)
|
|
|
|
def test_server_info_response_diagnostics_hidden(self):
|
|
data = {
|
|
"app_name": "Ignis Core",
|
|
"instance_name": "Home",
|
|
"uptime_seconds": 3600,
|
|
"diagnostics_visible": False,
|
|
"discovery": {
|
|
"last_scan_at": "2026-05-01T00:00:00",
|
|
"last_scan_mode": "startup",
|
|
"online": 3,
|
|
},
|
|
}
|
|
resp = ServerInfoResponse(**data)
|
|
self.assertFalse(resp.diagnostics_visible)
|
|
self.assertIsNone(resp.timezone)
|
|
self.assertIsNone(resp.build)
|
|
self.assertIsNone(resp.configuration)
|
|
|
|
def test_key_action_request(self):
|
|
req = KeyActionRequest(key="some-token")
|
|
self.assertEqual(req.key, "some-token")
|
|
|
|
def test_group_command_result_error(self):
|
|
item = GroupCommandResult(
|
|
ip="10.0.0.1", ok=False, kind="timeout", error="timeout"
|
|
)
|
|
self.assertFalse(item.ok)
|
|
self.assertEqual(item.kind, "timeout")
|
|
self.assertIsNone(item.result)
|
|
|
|
|
|
class SyncClientTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.client = IgnisClient("http://127.0.0.1:8000", "test-key")
|
|
self.client._client = MagicMock()
|
|
self.client._client.request = MagicMock()
|
|
|
|
def _mock(self, method, path, status=200, body=None):
|
|
self.client._client.request.return_value = _make_response(
|
|
status, body if body is not None else {}
|
|
)
|
|
|
|
def test_auth_me(self):
|
|
self._mock(
|
|
"GET",
|
|
"/auth/me",
|
|
body={"is_admin": True, "is_master": True, "name": "master"},
|
|
)
|
|
result = self.client.auth_me()
|
|
self.assertEqual(result["name"], "master")
|
|
self.assertTrue(result["is_master"])
|
|
|
|
def test_list_devices_returns_dict(self):
|
|
self._mock(
|
|
"GET",
|
|
"/devices",
|
|
body={
|
|
"aa:bb:cc:dd": {
|
|
"id": "aa:bb:cc:dd",
|
|
"ip": "10.0.0.1",
|
|
"name": "WiZ dd",
|
|
"room": "Default",
|
|
}
|
|
},
|
|
)
|
|
result = self.client.list_devices()
|
|
self.assertIsInstance(result, dict)
|
|
self.assertIn("aa:bb:cc:dd", result)
|
|
|
|
def test_list_scenes_local(self):
|
|
scenes = self.client.list_scenes()
|
|
self.assertEqual(scenes["cozy"], 6)
|
|
self.assertEqual(len(scenes), 31)
|
|
|
|
def test_control_device(self):
|
|
body = {"device_id": "aa:bb:cc:dd", "applied": {"state": True}, "status": "ok"}
|
|
self._mock("POST", "/control/device/aa:bb:cc:dd", body=body)
|
|
result = self.client.control_device("aa:bb:cc:dd", CommandRequest(state=True))
|
|
self.assertEqual(result.status, "ok")
|
|
self.assertEqual(result.device_id, "aa:bb:cc:dd")
|
|
|
|
def test_control_group(self):
|
|
body = {
|
|
"status": "ok",
|
|
"applied": {"state": True},
|
|
"sent_to": ["10.0.0.1"],
|
|
"success_count": 1,
|
|
"failure_count": 0,
|
|
"results": [{"ip": "10.0.0.1", "ok": True, "kind": "ok", "result": {}}],
|
|
}
|
|
self._mock("POST", "/control/group/bedroom", body=body)
|
|
result = self.client.control_group("bedroom", CommandRequest(state=True))
|
|
self.assertEqual(result.status, "ok")
|
|
self.assertEqual(result.success_count, 1)
|
|
|
|
def test_blink_device(self):
|
|
self._mock(
|
|
"POST",
|
|
"/control/device/aa:bb:cc:dd/blink",
|
|
body={"status": "blink_done", "original": True},
|
|
)
|
|
result = self.client.blink_device("aa:bb:cc:dd")
|
|
self.assertEqual(result.status, "blink_done")
|
|
self.assertTrue(result.original)
|
|
|
|
def test_device_status(self):
|
|
body = {"device_id": "aa:bb:cc:dd", "status": {"state": True, "dimming": 80}}
|
|
self._mock("GET", "/control/device/aa:bb:cc:dd/status", body=body)
|
|
result = self.client.device_status("aa:bb:cc:dd")
|
|
self.assertTrue(result.status["state"])
|
|
|
|
def test_group_status(self):
|
|
body = {
|
|
"group_id": "bedroom",
|
|
"results": [{"ip": "10.0.0.1", "status": {"state": True}}],
|
|
}
|
|
self._mock("GET", "/control/group/bedroom/status", body=body)
|
|
result = self.client.group_status("bedroom")
|
|
self.assertEqual(len(result.results), 1)
|
|
|
|
def test_create_once_schedule(self):
|
|
body = {
|
|
"status": "scheduled",
|
|
"job_id": "once_abc123",
|
|
"run_at": "2026-06-01T12:00:00",
|
|
}
|
|
self._mock("POST", "/schedules/once", body=body)
|
|
ts = datetime(2026, 6, 1, 12, 0, 0, tzinfo=timezone.utc)
|
|
result = self.client.create_once_schedule(
|
|
ScheduleOnceRequest(target_id="grp-1", run_at=ts, state=True)
|
|
)
|
|
self.assertEqual(result.status, "scheduled")
|
|
self.assertEqual(result.job_id, "once_abc123")
|
|
|
|
def test_create_cron_schedule(self):
|
|
body = {"status": "cron_scheduled", "job_id": "cron_def456"}
|
|
self._mock("POST", "/schedules/cron", body=body)
|
|
result = self.client.create_cron_schedule(
|
|
ScheduleCronRequest(target_id="grp-1", hour="8", minute="0", state=True)
|
|
)
|
|
self.assertEqual(result.status, "cron_scheduled")
|
|
|
|
def test_list_schedules(self):
|
|
body = {"tasks": []}
|
|
self._mock("GET", "/schedules/tasks", body=body)
|
|
result = self.client.list_schedules()
|
|
self.assertEqual(len(result.tasks), 0)
|
|
|
|
def test_delete_schedule(self):
|
|
self._mock("DELETE", "/schedules/cron_abc", body={"status": "deleted"})
|
|
result = self.client.delete_schedule("cron_abc")
|
|
self.assertEqual(result.status, "deleted")
|
|
|
|
def test_stats_summary(self):
|
|
body = {"period_days": 7, "since": "2026-05-20T00:00:00", "groups": []}
|
|
self._mock("GET", "/stats/summary", body=body)
|
|
result = self.client.stats_summary(days=7)
|
|
self.assertEqual(result["period_days"], 7)
|
|
|
|
def test_stats_log(self):
|
|
body = [
|
|
{
|
|
"id": 1,
|
|
"timestamp": "...",
|
|
"key_name": "master",
|
|
"action": "toggle_on",
|
|
"target_type": "device",
|
|
"target_id": "aa:bb:cc:dd",
|
|
"params": '{"command":{"state":true}}',
|
|
}
|
|
]
|
|
self._mock("GET", "/stats/log", body=body)
|
|
result = self.client.stats_log(limit=10)
|
|
self.assertEqual(len(result), 1)
|
|
self.assertEqual(result[0]["action"], "toggle_on")
|
|
self.assertIsInstance(result[0]["params"], str)
|
|
|
|
def test_create_api_key_uses_query_params(self):
|
|
self._mock(
|
|
"POST",
|
|
"/api-keys",
|
|
body={"key": "secret-token", "key_id": "abc", "name": "bot"},
|
|
)
|
|
result = self.client.create_api_key("bot", is_admin=False)
|
|
self.assertEqual(result["name"], "bot")
|
|
self.assertIn("key", result)
|
|
|
|
def test_list_api_keys(self):
|
|
self._mock("GET", "/api-keys", body=[])
|
|
result = self.client.list_api_keys()
|
|
self.assertEqual(result, [])
|
|
|
|
def test_revoke_api_key(self):
|
|
self._mock(
|
|
"POST",
|
|
"/api-keys/revoke",
|
|
body={"status": "revoked", "name": "bot", "key_id": "abc"},
|
|
)
|
|
result = self.client.revoke_api_key("secret-or-public-id")
|
|
self.assertEqual(result["status"], "revoked")
|
|
|
|
def test_activate_api_key(self):
|
|
self._mock(
|
|
"POST",
|
|
"/api-keys/activate",
|
|
body={"status": "activated", "name": "bot", "key_id": "abc"},
|
|
)
|
|
result = self.client.activate_api_key("secret-or-public-id")
|
|
self.assertEqual(result["status"], "activated")
|
|
|
|
def test_create_group(self):
|
|
self._mock(
|
|
"POST", "/devices/groups", body={"status": "created", "group": "bedroom"}
|
|
)
|
|
result = self.client.create_group("bedroom", "Bedroom", ["aa:bb:cc:dd"])
|
|
self.assertEqual(result["status"], "created")
|
|
|
|
def test_delete_group(self):
|
|
self._mock(
|
|
"DELETE",
|
|
"/devices/groups/bedroom",
|
|
body={"status": "deleted", "id": "bedroom"},
|
|
)
|
|
result = self.client.delete_group("bedroom")
|
|
self.assertEqual(result["status"], "deleted")
|
|
|
|
def test_rescan(self):
|
|
body = {
|
|
"status": "ok",
|
|
"found": 3,
|
|
"added": 1,
|
|
"updated": 2,
|
|
"removed_offline": 0,
|
|
"pending_removal": 0,
|
|
"online": 3,
|
|
}
|
|
self._mock("POST", "/devices/rescan", body=body)
|
|
result = self.client.rescan()
|
|
self.assertEqual(result.found, 3)
|
|
|
|
def test_system_info(self):
|
|
body = {
|
|
"app_name": "Ignis Core",
|
|
"uptime_seconds": 3600,
|
|
"diagnostics_visible": False,
|
|
"discovery": {"online": 3},
|
|
}
|
|
self._mock("GET", "/system/info", body=body)
|
|
result = self.client.system_info()
|
|
self.assertFalse(result.diagnostics_visible)
|
|
|
|
# --- Error handling ---
|
|
|
|
def test_403_raises_ignis_error(self):
|
|
self._mock(
|
|
"GET", "/api-keys", status=403, body={"detail": "Требуется мастер-ключ"}
|
|
)
|
|
with self.assertRaises(IgnisError) as ctx:
|
|
self.client.list_api_keys()
|
|
self.assertEqual(ctx.exception.status_code, 403)
|
|
|
|
def test_404_raises_ignis_error(self):
|
|
self._mock(
|
|
"POST",
|
|
"/control/device/no-such",
|
|
status=404,
|
|
body={"detail": "Лампа не в сети"},
|
|
)
|
|
with self.assertRaises(IgnisError) as ctx:
|
|
self.client.control_device("no-such", CommandRequest(state=True))
|
|
self.assertEqual(ctx.exception.status_code, 404)
|
|
|
|
def test_503_fail_closed(self):
|
|
self._mock(
|
|
"GET",
|
|
"/auth/me",
|
|
status=503,
|
|
body={"detail": "Сервер не настроен: задайте IGNIS_API_KEY"},
|
|
)
|
|
with self.assertRaises(IgnisError) as ctx:
|
|
self.client.auth_me()
|
|
self.assertEqual(ctx.exception.status_code, 503)
|
|
|
|
def test_504_timeout(self):
|
|
body = {"detail": "Команда лампе не доставлена: таймаут ответа"}
|
|
self._mock("POST", "/control/device/aa:bb:cc:dd", status=504, body=body)
|
|
with self.assertRaises(IgnisError) as ctx:
|
|
self.client.control_device("aa:bb:cc:dd", CommandRequest(state=True))
|
|
self.assertEqual(ctx.exception.status_code, 504)
|
|
|
|
def test_context_manager(self):
|
|
with IgnisClient("http://127.0.0.1:8000", "key") as client:
|
|
client._client.request = MagicMock(
|
|
return_value=_make_response(
|
|
200, {"name": "master", "is_admin": True, "is_master": True}
|
|
)
|
|
)
|
|
result = client.auth_me()
|
|
self.assertEqual(result["name"], "master")
|
|
|
|
|
|
class AsyncClientTests(unittest.IsolatedAsyncioTestCase):
|
|
def setUp(self):
|
|
self._responses = {}
|
|
|
|
def _mock(self, method, path, status=200, body=None):
|
|
self._responses[(method, path)] = _make_response(
|
|
status, body if body is not None else {}
|
|
)
|
|
|
|
async def _fake_request(self, method, path, **kwargs):
|
|
resp = self._responses.get((method, path))
|
|
if resp is None:
|
|
raise RuntimeError(f"Unexpected request: {method} {path}")
|
|
return resp
|
|
|
|
async def asyncSetUp(self):
|
|
self.client = AsyncIgnisClient("http://127.0.0.1:8000", "test-key")
|
|
self.client._client = AsyncMock()
|
|
self.client._client.request = self._fake_request
|
|
|
|
async def asyncTearDown(self):
|
|
await self.client.close()
|
|
|
|
async def test_auth_me_async(self):
|
|
self._mock(
|
|
"GET",
|
|
"/auth/me",
|
|
body={"name": "master", "is_admin": True, "is_master": True},
|
|
)
|
|
result = await self.client.auth_me()
|
|
self.assertEqual(result["name"], "master")
|
|
|
|
async def test_control_device_async(self):
|
|
body = {"device_id": "aa:bb:cc:dd", "applied": {"state": True}, "status": "ok"}
|
|
self._mock("POST", "/control/device/aa:bb:cc:dd", body=body)
|
|
result = await self.client.control_device(
|
|
"aa:bb:cc:dd", CommandRequest(state=True)
|
|
)
|
|
self.assertEqual(result.status, "ok")
|
|
|
|
async def test_async_error_handling(self):
|
|
self._mock(
|
|
"POST",
|
|
"/control/group/no-such",
|
|
status=404,
|
|
body={"detail": "Группа не найдена или оффлайн"},
|
|
)
|
|
with self.assertRaises(IgnisError) as ctx:
|
|
await self.client.control_group("no-such", CommandRequest(state=True))
|
|
self.assertEqual(ctx.exception.status_code, 404)
|
|
|
|
async def test_async_context_manager(self):
|
|
self._mock(
|
|
"GET",
|
|
"/auth/me",
|
|
body={"name": "master", "is_admin": True, "is_master": True},
|
|
)
|
|
result = await self.client.auth_me()
|
|
self.assertEqual(result["name"], "master")
|
|
|
|
async def test_blink_device_async(self):
|
|
self._mock(
|
|
"POST",
|
|
"/control/device/aa:bb:cc:dd/blink",
|
|
body={"status": "blink_done", "original": False},
|
|
)
|
|
result = await self.client.blink_device("aa:bb:cc:dd")
|
|
self.assertEqual(result.status, "blink_done")
|
|
self.assertFalse(result.original)
|
|
|
|
async def test_create_api_key_async(self):
|
|
self._mock(
|
|
"POST", "/api-keys", body={"key": "tok", "key_id": "abc", "name": "bot"}
|
|
)
|
|
result = await self.client.create_api_key("bot", is_admin=True)
|
|
self.assertEqual(result["name"], "bot")
|
|
|
|
|
|
class EdgeCaseTests(unittest.TestCase):
|
|
def test_request_handles_non_json_error(self):
|
|
client = IgnisClient("http://127.0.0.1:8000", "key")
|
|
client._client = MagicMock()
|
|
resp = MagicMock()
|
|
resp.status_code = 502
|
|
resp.json.side_effect = ValueError("not json")
|
|
resp.text = "<html>502 Bad Gateway</html>"
|
|
client._client.request.return_value = resp
|
|
|
|
with self.assertRaises(IgnisError) as ctx:
|
|
client.auth_me()
|
|
self.assertEqual(ctx.exception.status_code, 502)
|
|
self.assertIn("502 Bad Gateway", ctx.exception.detail)
|
|
|
|
def test_request_handles_empty_non_json_error(self):
|
|
client = IgnisClient("http://127.0.0.1:8000", "key")
|
|
client._client = MagicMock()
|
|
resp = MagicMock()
|
|
resp.status_code = 500
|
|
resp.json.side_effect = ValueError("not json")
|
|
resp.text = ""
|
|
client._client.request.return_value = resp
|
|
|
|
with self.assertRaises(IgnisError) as ctx:
|
|
client.auth_me()
|
|
self.assertEqual(ctx.exception.status_code, 500)
|
|
self.assertEqual(ctx.exception.detail, "HTTP 500")
|
|
|
|
def test_list_scenes_returns_all_scenes(self):
|
|
client = IgnisClient("http://127.0.0.1:8000", "key")
|
|
client._client = MagicMock()
|
|
scenes = client.list_scenes()
|
|
self.assertIsInstance(scenes, dict)
|
|
self.assertEqual(len(scenes), 31)
|
|
self.assertEqual(scenes["ocean"], 1)
|
|
self.assertEqual(scenes["steampunk"], 35)
|
|
self.assertNotIn(4, scenes.values())
|
|
|
|
def test_command_request_temp_min_boundary(self):
|
|
cmd = CommandRequest(temp=2200)
|
|
self.assertEqual(cmd.to_wiz_params(), {"temp": 2200})
|
|
|
|
def test_command_request_temp_max_boundary(self):
|
|
cmd = CommandRequest(temp=6500)
|
|
self.assertEqual(cmd.to_wiz_params(), {"temp": 6500})
|
|
|
|
def test_command_request_state_false_with_brightness(self):
|
|
cmd = CommandRequest(state=False, brightness=10)
|
|
self.assertEqual(cmd.to_wiz_params(), {"state": False, "dimming": 10})
|
|
|
|
def test_group_command_result_without_error_field(self):
|
|
item = GroupCommandResult(
|
|
ip="10.0.0.1", ok=True, kind="ok", result={"mac": "aa:bb"}
|
|
)
|
|
self.assertTrue(item.ok)
|
|
self.assertIsNone(item.error)
|
|
|
|
def test_group_command_result_with_error(self):
|
|
item = GroupCommandResult(
|
|
ip="10.0.0.2", ok=False, kind="timeout", error="timed out"
|
|
)
|
|
self.assertFalse(item.ok)
|
|
self.assertEqual(item.error, "timed out")
|
|
|
|
def test_fake_response_text_property(self):
|
|
resp = _make_response(400, {"detail": "bad request"})
|
|
self.assertEqual(resp.status_code, 400)
|
|
self.assertEqual(resp.json(), {"detail": "bad request"})
|
|
self.assertIn("bad request", resp.text)
|
|
|
|
def test_fake_response_text_with_list_body(self):
|
|
resp = _make_response(200, [])
|
|
self.assertEqual(resp.text, "[]")
|
|
|
|
def test_schedule_cron_request_default_day_of_week(self):
|
|
req = ScheduleCronRequest(target_id="grp-1", hour="8", minute="0", state=True)
|
|
self.assertEqual(req.day_of_week, "*")
|
|
|
|
def test_schedule_once_request_is_group_default(self):
|
|
from datetime import datetime, timedelta, timezone
|
|
|
|
ts = datetime.now(timezone.utc) + timedelta(hours=1)
|
|
req = ScheduleOnceRequest(target_id="grp-1", run_at=ts, state=True)
|
|
self.assertTrue(req.is_group)
|
|
|
|
def test_scenes_contains_expected_entries(self):
|
|
self.assertIn("cozy", SCENES)
|
|
self.assertIn("fireplace", SCENES)
|
|
self.assertIn("wake_up", SCENES)
|
|
self.assertIn("bedtime", SCENES)
|
|
self.assertIn("night_light", SCENES)
|
|
self.assertIn("ocean", SCENES)
|
|
self.assertIn("club", SCENES)
|
|
self.assertIn("steampunk", SCENES)
|
|
|
|
def test_ignis_error_is_exception(self):
|
|
err = IgnisError(418, "I'm a teapot")
|
|
try:
|
|
raise err
|
|
except IgnisError as caught:
|
|
self.assertEqual(caught.status_code, 418)
|
|
self.assertEqual(caught.detail, "I'm a teapot")
|
|
|
|
def test_group_create_schema(self):
|
|
from ignis_client.models import GroupCreateSchema
|
|
|
|
g = GroupCreateSchema(id="test", name="Test Group", macs=["aa:bb", "cc:dd"])
|
|
self.assertEqual(g.id, "test")
|
|
self.assertEqual(len(g.macs), 2)
|
|
|
|
def test_device_schema(self):
|
|
from ignis_client.models import DeviceSchema
|
|
|
|
d = DeviceSchema(id="aa:bb:cc:dd", ip="10.0.0.1", name="WiZ dd", room="Kitchen")
|
|
self.assertEqual(d.room, "Kitchen")
|