feat(http-endpoints): introduce HTTP endpoint output target stack
New output kind that POSTs the current strip frame to a user-configured HTTP endpoint, alongside WLED / MQTT / Hue. Stack mirrors the existing output-target shape end-to-end: storage model + store, FastAPI router + Pydantic schemas, JS feature module + modal template, router wiring in api/__init__.py and the modal include in index.html. Tests cover both the routes and the store.
This commit is contained in:
@@ -0,0 +1,168 @@
|
||||
"""Tests for HTTPEndpointStore — CRUD + auth_token encryption + header policy."""
|
||||
|
||||
import pytest
|
||||
|
||||
from ledgrab.storage.database import Database
|
||||
from ledgrab.storage.http_endpoint_store import HTTPEndpointStore
|
||||
from ledgrab.utils import secret_box
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def http_store(tmp_path):
|
||||
db = Database(tmp_path / "test.db")
|
||||
store = HTTPEndpointStore(db)
|
||||
yield store
|
||||
db.close()
|
||||
|
||||
|
||||
class TestCRUD:
|
||||
def test_create_basic(self, http_store):
|
||||
e = http_store.create_endpoint(
|
||||
name="Plex",
|
||||
url="https://plex.example.com/status/sessions",
|
||||
)
|
||||
assert e.id.startswith("htep_")
|
||||
assert e.url == "https://plex.example.com/status/sessions"
|
||||
assert e.method == "GET"
|
||||
assert e.timeout_s == 10.0
|
||||
assert e.headers == {}
|
||||
assert e.auth_token == ""
|
||||
|
||||
def test_create_with_auth_and_headers(self, http_store):
|
||||
e = http_store.create_endpoint(
|
||||
name="Plex",
|
||||
url="https://plex.example.com/status/sessions",
|
||||
auth_token="my-secret-token",
|
||||
headers={"Accept": "application/json"},
|
||||
timeout_s=5.0,
|
||||
)
|
||||
assert e.auth_token == "my-secret-token"
|
||||
assert e.headers == {"Accept": "application/json"}
|
||||
assert e.timeout_s == 5.0
|
||||
|
||||
def test_create_requires_url(self, http_store):
|
||||
with pytest.raises(ValueError, match="url is required"):
|
||||
http_store.create_endpoint(name="x", url="")
|
||||
|
||||
def test_create_rejects_unsupported_method(self, http_store):
|
||||
with pytest.raises(ValueError, match="Unsupported method"):
|
||||
http_store.create_endpoint(name="x", url="https://example.com", method="POST")
|
||||
|
||||
def test_create_rejects_zero_timeout(self, http_store):
|
||||
with pytest.raises(ValueError, match="timeout_s"):
|
||||
http_store.create_endpoint(name="x", url="https://example.com", timeout_s=0)
|
||||
|
||||
def test_unique_name(self, http_store):
|
||||
http_store.create_endpoint(name="dup", url="https://a")
|
||||
with pytest.raises(ValueError, match="already exists"):
|
||||
http_store.create_endpoint(name="dup", url="https://b")
|
||||
|
||||
def test_update(self, http_store):
|
||||
e = http_store.create_endpoint(name="Plex", url="https://a")
|
||||
updated = http_store.update_endpoint(
|
||||
e.id,
|
||||
url="https://b",
|
||||
auth_token="new-token",
|
||||
)
|
||||
assert updated.url == "https://b"
|
||||
assert updated.auth_token == "new-token"
|
||||
assert updated.created_at == e.created_at
|
||||
assert updated.updated_at > e.updated_at
|
||||
|
||||
def test_update_with_empty_token_clears(self, tmp_path):
|
||||
"""Sending ``auth_token=""`` MUST clear the stored token (the
|
||||
endpoint route distinguishes None=keep from ""=clear; the store
|
||||
layer is the source of truth for the on-disk state)."""
|
||||
db = Database(tmp_path / "test.db")
|
||||
store = HTTPEndpointStore(db)
|
||||
e = store.create_endpoint(name="x", url="https://a", auth_token="initial-secret")
|
||||
assert e.auth_token == "initial-secret"
|
||||
|
||||
cleared = store.update_endpoint(e.id, auth_token="")
|
||||
assert cleared.auth_token == ""
|
||||
|
||||
# Re-open the DB to confirm the change is persisted, not just in
|
||||
# the in-memory cache.
|
||||
db.close()
|
||||
db2 = Database(tmp_path / "test.db")
|
||||
store2 = HTTPEndpointStore(db2)
|
||||
reloaded = store2.get_endpoint(e.id)
|
||||
assert reloaded.auth_token == ""
|
||||
db2.close()
|
||||
|
||||
def test_delete(self, http_store):
|
||||
e = http_store.create_endpoint(name="x", url="https://a")
|
||||
http_store.delete_endpoint(e.id)
|
||||
from ledgrab.storage.base_store import EntityNotFoundError
|
||||
|
||||
with pytest.raises(EntityNotFoundError):
|
||||
http_store.get_endpoint(e.id)
|
||||
|
||||
|
||||
class TestTokenEncryption:
|
||||
"""auth_token must be encrypted at rest, decrypted on load."""
|
||||
|
||||
def test_token_encrypted_at_rest(self, tmp_path):
|
||||
db = Database(tmp_path / "test.db")
|
||||
store = HTTPEndpointStore(db)
|
||||
store.create_endpoint(name="x", url="https://a", auth_token="plaintext-secret")
|
||||
|
||||
rows = db.load_all("http_endpoints")
|
||||
assert len(rows) == 1
|
||||
raw = rows[0]["auth_token"]
|
||||
assert raw != "plaintext-secret"
|
||||
assert secret_box.is_encrypted(raw)
|
||||
db.close()
|
||||
|
||||
def test_token_decrypted_on_load(self, tmp_path):
|
||||
db_path = tmp_path / "test.db"
|
||||
db = Database(db_path)
|
||||
store = HTTPEndpointStore(db)
|
||||
eid = store.create_endpoint(name="x", url="https://a", auth_token="round-trip").id
|
||||
db.close()
|
||||
|
||||
db2 = Database(db_path)
|
||||
store2 = HTTPEndpointStore(db2)
|
||||
loaded = store2.get_endpoint(eid)
|
||||
assert loaded.auth_token == "round-trip"
|
||||
db2.close()
|
||||
|
||||
|
||||
class TestHeadersAndAuth:
|
||||
def test_build_request_headers_with_token(self, http_store):
|
||||
e = http_store.create_endpoint(
|
||||
name="x",
|
||||
url="https://a",
|
||||
auth_token="abc",
|
||||
headers={"Accept": "application/json"},
|
||||
)
|
||||
h = e.build_request_headers()
|
||||
assert h["Authorization"] == "Bearer abc"
|
||||
assert h["Accept"] == "application/json"
|
||||
|
||||
def test_custom_authorization_header_wins(self, http_store):
|
||||
e = http_store.create_endpoint(
|
||||
name="x",
|
||||
url="https://a",
|
||||
auth_token="bearer-token",
|
||||
headers={"Authorization": "Basic dXNlcjpwYXNz"},
|
||||
)
|
||||
h = e.build_request_headers()
|
||||
assert h["Authorization"] == "Basic dXNlcjpwYXNz"
|
||||
|
||||
def test_case_insensitive_authorization_check(self, http_store):
|
||||
"""Lowercase 'authorization' supplied by the user must win too."""
|
||||
e = http_store.create_endpoint(
|
||||
name="x",
|
||||
url="https://a",
|
||||
auth_token="should-not-be-sent",
|
||||
headers={"authorization": "Basic xyz"},
|
||||
)
|
||||
h = e.build_request_headers()
|
||||
# No second Authorization key — user's lowercase variant is preserved as-is
|
||||
assert h.get("authorization") == "Basic xyz"
|
||||
assert "Authorization" not in h
|
||||
|
||||
def test_no_auth_no_header(self, http_store):
|
||||
e = http_store.create_endpoint(name="x", url="https://a")
|
||||
assert "Authorization" not in e.build_request_headers()
|
||||
Reference in New Issue
Block a user