alexei.dolgolyov
39b0554444
A) Per-pixel smart-lights - LIFX multizone (SetExtendedColorZones msg 510, <=82 zones) + Tile (SetTileState64 715), auto-detected on connect with single-colour fallback; lifx_per_zone threaded like nanoleaf_per_panel - Hue gradient-lightstrip mapping: Entertainment v2 frame now keyed by channel id (was 1 light=1 LED), channels discovered on connect; hue_gradient_mode toggle (default on) B) Integrations bundle - Outbound webhook automation action (Discord/IFTTT/Zapier/Node-RED), SSRF-gated via validate_polling_url at both save and fire time; fires on activate/deactivate, best-effort, audited - Home Assistant MQTT auto-discovery: read-only binary_sensors per automation + connectivity, availability via birth/will, cleanup on disable/delete, live state from the engine Shared: pixel_reduce.resample_to_n nearest-neighbour helper. 57 new tests (lifx_multizone, hue_segment, webhook_action, ha_discovery). Gate: ruff + tsc + build clean, pytest 2719 passed / 2 skipped.
137 lines
4.8 KiB
Python
137 lines
4.8 KiB
Python
"""Tests for the outbound webhook automation action.
|
|
|
|
Covers the pure logic (template rendering, fire_on filtering, model
|
|
round-trip) and the fire path with a mocked transport (success, non-2xx,
|
|
SSRF-blocked). Live delivery to a real Discord/Zapier endpoint is out of
|
|
scope for CI and is verified manually.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from datetime import datetime, timezone
|
|
|
|
import httpx
|
|
import pytest
|
|
import respx
|
|
|
|
from ledgrab.core.automations.webhook_action import (
|
|
fire_webhook_action,
|
|
render_template,
|
|
should_fire,
|
|
)
|
|
from ledgrab.storage.automation import Action, Automation, WebhookAction
|
|
|
|
|
|
def _automation(actions=None) -> Automation:
|
|
now = datetime.now(timezone.utc)
|
|
return Automation(
|
|
id="auto_1",
|
|
name="Movie Night",
|
|
enabled=True,
|
|
rule_logic="or",
|
|
rules=[],
|
|
scene_preset_id=None,
|
|
deactivation_mode="none",
|
|
deactivation_scene_preset_id=None,
|
|
created_at=now,
|
|
updated_at=now,
|
|
actions=actions or [],
|
|
)
|
|
|
|
|
|
class TestRenderTemplate:
|
|
def test_substitutes_all_tokens(self):
|
|
out = render_template(
|
|
'{"name":"{{automation_name}}","id":"{{automation_id}}","ev":"{{event}}"}',
|
|
_automation(),
|
|
"activate",
|
|
)
|
|
assert '"name":"Movie Night"' in out
|
|
assert '"id":"auto_1"' in out
|
|
assert '"ev":"activate"' in out
|
|
|
|
def test_leaves_unknown_tokens(self):
|
|
assert render_template("{{unknown}}", _automation(), "activate") == "{{unknown}}"
|
|
|
|
|
|
class TestShouldFire:
|
|
def test_matches_event_or_both(self):
|
|
assert should_fire(WebhookAction(fire_on="activate"), "activate")
|
|
assert not should_fire(WebhookAction(fire_on="activate"), "deactivate")
|
|
assert should_fire(WebhookAction(fire_on="both"), "activate")
|
|
assert should_fire(WebhookAction(fire_on="both"), "deactivate")
|
|
|
|
|
|
class TestModelRoundTrip:
|
|
def test_webhook_action_round_trips(self):
|
|
a = WebhookAction(
|
|
webhook_url="https://example.com/hook",
|
|
method="PUT",
|
|
body_template="hi {{event}}",
|
|
content_type="text/plain",
|
|
fire_on="both",
|
|
)
|
|
back = Action.from_dict(a.to_dict())
|
|
assert isinstance(back, WebhookAction)
|
|
assert back.webhook_url == "https://example.com/hook"
|
|
assert back.method == "PUT"
|
|
assert back.fire_on == "both"
|
|
|
|
def test_unknown_action_type_raises(self):
|
|
with pytest.raises(ValueError):
|
|
Action.from_dict({"action_type": "nope"})
|
|
|
|
def test_from_dict_normalises_bad_method_and_fire_on(self):
|
|
a = WebhookAction.from_dict({"method": "delete", "fire_on": "whenever"})
|
|
assert a.method == "POST"
|
|
assert a.fire_on == "activate"
|
|
|
|
def test_automation_actions_survive_serialization(self):
|
|
auto = _automation([WebhookAction(webhook_url="https://x.test/h", fire_on="both")])
|
|
back = Automation.from_dict(auto.to_dict())
|
|
assert len(back.actions) == 1
|
|
assert isinstance(back.actions[0], WebhookAction)
|
|
assert back.actions[0].webhook_url == "https://x.test/h"
|
|
|
|
def test_no_actions_omitted_from_dict(self):
|
|
assert "actions" not in _automation().to_dict()
|
|
|
|
|
|
class TestFire:
|
|
@respx.mock
|
|
@pytest.mark.asyncio
|
|
async def test_success_returns_true(self):
|
|
route = respx.post("http://93.184.216.34/hook").mock(return_value=httpx.Response(204))
|
|
action = WebhookAction(webhook_url="http://93.184.216.34/hook", body_template="{{event}}")
|
|
ok, err = await fire_webhook_action(action, _automation(), "activate")
|
|
assert ok is True and err is None
|
|
assert route.called
|
|
# Body template was rendered and sent.
|
|
assert route.calls.last.request.content == b"activate"
|
|
|
|
@respx.mock
|
|
@pytest.mark.asyncio
|
|
async def test_non_2xx_returns_error(self):
|
|
respx.post("http://93.184.216.34/hook").mock(return_value=httpx.Response(500))
|
|
action = WebhookAction(webhook_url="http://93.184.216.34/hook")
|
|
ok, err = await fire_webhook_action(action, _automation(), "activate")
|
|
assert ok is False and err == "HTTP 500"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_ssrf_blocked_loopback(self):
|
|
# validate_polling_url must reject loopback — no HTTP call is made.
|
|
action = WebhookAction(webhook_url="http://127.0.0.1:8080/admin")
|
|
ok, err = await fire_webhook_action(action, _automation(), "activate")
|
|
assert ok is False and "SSRF" in (err or "")
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_empty_url_returns_error(self):
|
|
ok, err = await fire_webhook_action(
|
|
WebhookAction(webhook_url=""), _automation(), "activate"
|
|
)
|
|
assert ok is False and err == "no URL configured"
|
|
|
|
|
|
if __name__ == "__main__":
|
|
pytest.main([__file__, "-v"])
|