mirror of
https://github.com/maziggy/bambuddy.git
synced 2026-09-30 19:21:33 +02:00
716 lines
29 KiB
Python
716 lines
29 KiB
Python
"""Integration tests for Notifications API endpoints.
|
|
|
|
Tests the full request/response cycle for /api/v1/notifications/ endpoints.
|
|
"""
|
|
|
|
import pytest
|
|
from httpx import AsyncClient
|
|
from sqlalchemy import text
|
|
|
|
from backend.tests._fixtures.notification_toggles import EVENT_TOGGLE_COLUMNS, TOGGLE_TARGET
|
|
|
|
|
|
class TestNotificationsAPI:
|
|
"""Integration tests for /api/v1/notifications/ endpoints."""
|
|
|
|
# ========================================================================
|
|
# List endpoints
|
|
# ========================================================================
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_list_notification_providers_empty(self, async_client: AsyncClient):
|
|
"""Verify empty list is returned when no providers exist."""
|
|
response = await async_client.get("/api/v1/notifications/")
|
|
|
|
assert response.status_code == 200
|
|
assert response.json() == []
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_list_notification_providers_with_data(
|
|
self, async_client: AsyncClient, notification_provider_factory, db_session
|
|
):
|
|
"""Verify list returns existing providers."""
|
|
_provider = await notification_provider_factory(name="Test Provider")
|
|
|
|
response = await async_client.get("/api/v1/notifications/")
|
|
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert len(data) >= 1
|
|
assert any(p["name"] == "Test Provider" for p in data)
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_a_row_with_null_event_flags_is_still_listable(
|
|
self, async_client: AsyncClient, notification_provider_factory, db_session
|
|
):
|
|
"""A legacy row whose flag columns were never backfilled must not 500 the list.
|
|
|
|
Every on_* column is nullable with no server default, so a row created
|
|
before a flag existed keeps NULL there until a migration backfills it --
|
|
and #1184's ALTER ... DEFAULT false silently did not, on any install
|
|
where create_all() had already added the column. Declaring those flags
|
|
on the response schema in #2827 turned those NULLs into a hard failure:
|
|
pydantic rejects None for a bool, so every provider row failed at once
|
|
and the list came back empty to the UI.
|
|
|
|
Written against the two flags that actually broke, but the whole set is
|
|
checked -- the next flag added to the schema has the same exposure.
|
|
"""
|
|
provider = await notification_provider_factory(name="Legacy Provider")
|
|
|
|
flags = ["on_stock_reorder_alert", "on_stock_break_alert"]
|
|
# nosec B608 - the only interpolated fragments are built from `flags`,
|
|
# the literal list directly above. A column name cannot be a bind
|
|
# parameter, which is why it is written into the string at all; the id,
|
|
# which is the one caller-supplied value here, is bound.
|
|
null_assignments = ", ".join(f"{f} = NULL" for f in flags)
|
|
await db_session.execute(
|
|
text(f"UPDATE notification_providers SET {null_assignments} WHERE id = :id"), # nosec B608
|
|
{"id": provider.id},
|
|
)
|
|
await db_session.commit()
|
|
|
|
columns = ", ".join(flags)
|
|
stored = await db_session.execute(
|
|
text(f"SELECT {columns} FROM notification_providers WHERE id = :id"), # nosec B608
|
|
{"id": provider.id},
|
|
)
|
|
assert all(value is None for value in stored.one()), "row under test must actually hold NULLs"
|
|
|
|
response = await async_client.get("/api/v1/notifications/")
|
|
|
|
assert response.status_code == 200
|
|
listed = next(p for p in response.json() if p["name"] == "Legacy Provider")
|
|
# Off, not the field default: the sender selects on `.is_(True)`, so a
|
|
# NULL flag never sent anything, and repairing the read must not switch
|
|
# a notification on.
|
|
assert all(listed[flag] is False for flag in flags)
|
|
|
|
# The single-provider route reads through the same schema.
|
|
single = await async_client.get(f"/api/v1/notifications/{provider.id}")
|
|
assert single.status_code == 200
|
|
assert all(single.json()[flag] is False for flag in flags)
|
|
|
|
# ========================================================================
|
|
# Create endpoints
|
|
# ========================================================================
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_create_callmebot_provider(self, async_client: AsyncClient):
|
|
"""Verify callmebot notification provider can be created."""
|
|
data = {
|
|
"name": "Test CallMeBot",
|
|
"provider_type": "callmebot",
|
|
"enabled": True,
|
|
"config": {"phone_number": "+1234567890", "api_key": "test-api-key"},
|
|
"on_print_start": True,
|
|
"on_print_complete": True,
|
|
"on_print_failed": True,
|
|
"on_print_stopped": False,
|
|
}
|
|
|
|
response = await async_client.post("/api/v1/notifications/", json=data)
|
|
|
|
assert response.status_code == 200
|
|
result = response.json()
|
|
assert result["name"] == "Test CallMeBot"
|
|
assert result["provider_type"] == "callmebot"
|
|
assert result["on_print_start"] is True
|
|
assert result["on_print_stopped"] is False
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_create_ntfy_provider(self, async_client: AsyncClient):
|
|
"""Verify ntfy notification provider can be created."""
|
|
data = {
|
|
"name": "Test Ntfy",
|
|
"provider_type": "ntfy",
|
|
"enabled": True,
|
|
"config": {
|
|
"server": "https://ntfy.sh",
|
|
"topic": "test-topic",
|
|
},
|
|
"on_print_complete": True,
|
|
}
|
|
|
|
response = await async_client.post("/api/v1/notifications/", json=data)
|
|
|
|
assert response.status_code == 200
|
|
result = response.json()
|
|
assert result["provider_type"] == "ntfy"
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_create_provider_with_printer(self, async_client: AsyncClient, printer_factory, db_session):
|
|
"""Verify provider can be linked to specific printer."""
|
|
printer = await printer_factory(name="Test Printer")
|
|
|
|
data = {
|
|
"name": "Printer Ntfy",
|
|
"provider_type": "ntfy",
|
|
"config": {"server": "https://ntfy.sh", "topic": "test-topic"},
|
|
"printer_id": printer.id,
|
|
}
|
|
|
|
response = await async_client.post("/api/v1/notifications/", json=data)
|
|
|
|
assert response.status_code == 200
|
|
result = response.json()
|
|
assert result["printer_id"] == printer.id
|
|
|
|
# ========================================================================
|
|
# Get single endpoint
|
|
# ========================================================================
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_get_notification_provider(
|
|
self, async_client: AsyncClient, notification_provider_factory, db_session
|
|
):
|
|
"""Verify single provider can be retrieved."""
|
|
provider = await notification_provider_factory(name="Get Test Provider")
|
|
|
|
response = await async_client.get(f"/api/v1/notifications/{provider.id}")
|
|
|
|
assert response.status_code == 200
|
|
result = response.json()
|
|
assert result["id"] == provider.id
|
|
assert result["name"] == "Get Test Provider"
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_get_provider_not_found(self, async_client: AsyncClient):
|
|
"""Verify 404 for non-existent provider."""
|
|
response = await async_client.get("/api/v1/notifications/9999")
|
|
|
|
assert response.status_code == 404
|
|
|
|
# ========================================================================
|
|
# Update endpoints (CRITICAL - toggle persistence)
|
|
# ========================================================================
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_update_event_toggles(self, async_client: AsyncClient, notification_provider_factory, db_session):
|
|
"""CRITICAL: Verify notification event toggles persist correctly."""
|
|
provider = await notification_provider_factory(
|
|
on_print_start=True,
|
|
on_print_complete=True,
|
|
on_print_stopped=False,
|
|
)
|
|
|
|
# Toggle on_print_stopped to True
|
|
response = await async_client.patch(f"/api/v1/notifications/{provider.id}", json={"on_print_stopped": True})
|
|
|
|
assert response.status_code == 200
|
|
assert response.json()["on_print_stopped"] is True
|
|
|
|
# Verify change persisted
|
|
response = await async_client.get(f"/api/v1/notifications/{provider.id}")
|
|
assert response.json()["on_print_stopped"] is True
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_update_ams_alarm_toggles(self, async_client: AsyncClient, notification_provider_factory, db_session):
|
|
"""CRITICAL: Verify AMS alarm toggles persist correctly."""
|
|
provider = await notification_provider_factory(
|
|
on_ams_humidity_high=False,
|
|
on_ams_temperature_high=False,
|
|
)
|
|
|
|
# Enable AMS alarms
|
|
response = await async_client.patch(
|
|
f"/api/v1/notifications/{provider.id}",
|
|
json={
|
|
"on_ams_humidity_high": True,
|
|
"on_ams_temperature_high": True,
|
|
},
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
result = response.json()
|
|
assert result["on_ams_humidity_high"] is True
|
|
assert result["on_ams_temperature_high"] is True
|
|
|
|
# Verify persistence
|
|
response = await async_client.get(f"/api/v1/notifications/{provider.id}")
|
|
result = response.json()
|
|
assert result["on_ams_humidity_high"] is True
|
|
assert result["on_ams_temperature_high"] is True
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_enable_disable_provider(self, async_client: AsyncClient, notification_provider_factory, db_session):
|
|
"""Verify provider can be enabled/disabled."""
|
|
provider = await notification_provider_factory(enabled=True)
|
|
|
|
# Disable
|
|
response = await async_client.patch(f"/api/v1/notifications/{provider.id}", json={"enabled": False})
|
|
|
|
assert response.status_code == 200
|
|
assert response.json()["enabled"] is False
|
|
|
|
# Enable
|
|
response = await async_client.patch(f"/api/v1/notifications/{provider.id}", json={"enabled": True})
|
|
|
|
assert response.status_code == 200
|
|
assert response.json()["enabled"] is True
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_update_quiet_hours(self, async_client: AsyncClient, notification_provider_factory, db_session):
|
|
"""Verify quiet hours can be configured."""
|
|
provider = await notification_provider_factory(quiet_hours_enabled=False)
|
|
|
|
response = await async_client.patch(
|
|
f"/api/v1/notifications/{provider.id}",
|
|
json={
|
|
"quiet_hours_enabled": True,
|
|
"quiet_hours_start": "22:00",
|
|
"quiet_hours_end": "07:00",
|
|
},
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
result = response.json()
|
|
assert result["quiet_hours_enabled"] is True
|
|
assert result["quiet_hours_start"] == "22:00"
|
|
assert result["quiet_hours_end"] == "07:00"
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_update_daily_digest(self, async_client: AsyncClient, notification_provider_factory, db_session):
|
|
"""Verify daily digest can be configured."""
|
|
provider = await notification_provider_factory(daily_digest_enabled=False)
|
|
|
|
response = await async_client.patch(
|
|
f"/api/v1/notifications/{provider.id}",
|
|
json={
|
|
"daily_digest_enabled": True,
|
|
"daily_digest_time": "09:00",
|
|
},
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
result = response.json()
|
|
assert result["daily_digest_enabled"] is True
|
|
assert result["daily_digest_time"] == "09:00"
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_update_multiple_event_toggles(
|
|
self, async_client: AsyncClient, notification_provider_factory, db_session
|
|
):
|
|
"""Verify multiple event toggles can be updated at once."""
|
|
provider = await notification_provider_factory(
|
|
on_print_start=True,
|
|
on_print_complete=True,
|
|
on_print_failed=True,
|
|
on_print_stopped=False,
|
|
on_printer_offline=False,
|
|
)
|
|
|
|
response = await async_client.patch(
|
|
f"/api/v1/notifications/{provider.id}",
|
|
json={
|
|
"on_print_start": False,
|
|
"on_print_stopped": True,
|
|
"on_printer_offline": True,
|
|
},
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
result = response.json()
|
|
assert result["on_print_start"] is False
|
|
assert result["on_print_stopped"] is True
|
|
assert result["on_printer_offline"] is True
|
|
# Unchanged fields should remain
|
|
assert result["on_print_complete"] is True
|
|
assert result["on_print_failed"] is True
|
|
|
|
# ========================================================================
|
|
# Test notification endpoint
|
|
# ========================================================================
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_test_notification(
|
|
self, async_client: AsyncClient, notification_provider_factory, mock_httpx_client, db_session
|
|
):
|
|
"""Verify test notification can be sent."""
|
|
provider = await notification_provider_factory()
|
|
|
|
response = await async_client.post(f"/api/v1/notifications/{provider.id}/test")
|
|
|
|
assert response.status_code == 200
|
|
result = response.json()
|
|
assert result["success"] is True
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_test_notification_disabled_provider(
|
|
self, async_client: AsyncClient, notification_provider_factory, db_session
|
|
):
|
|
"""Verify test notification works even for disabled provider."""
|
|
provider = await notification_provider_factory(enabled=False)
|
|
|
|
response = await async_client.post(f"/api/v1/notifications/{provider.id}/test")
|
|
|
|
# Test should still work for disabled providers
|
|
assert response.status_code == 200
|
|
|
|
# ========================================================================
|
|
# Delete endpoint
|
|
# ========================================================================
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_delete_notification_provider(
|
|
self, async_client: AsyncClient, notification_provider_factory, db_session
|
|
):
|
|
"""Verify notification provider can be deleted."""
|
|
provider = await notification_provider_factory()
|
|
provider_id = provider.id
|
|
|
|
response = await async_client.delete(f"/api/v1/notifications/{provider_id}")
|
|
|
|
assert response.status_code == 200
|
|
|
|
# Verify deleted
|
|
response = await async_client.get(f"/api/v1/notifications/{provider_id}")
|
|
assert response.status_code == 404
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_delete_nonexistent_provider(self, async_client: AsyncClient):
|
|
"""Verify deleting non-existent provider returns 404."""
|
|
response = await async_client.delete("/api/v1/notifications/9999")
|
|
|
|
assert response.status_code == 404
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_create_provider_with_first_layer_complete(self, async_client: AsyncClient):
|
|
"""Verify first layer complete toggle persists on create."""
|
|
data = {
|
|
"name": "First Layer Test",
|
|
"provider_type": "ntfy",
|
|
"config": {"server": "https://ntfy.sh", "topic": "test"},
|
|
"on_first_layer_complete": True,
|
|
}
|
|
|
|
response = await async_client.post("/api/v1/notifications/", json=data)
|
|
|
|
assert response.status_code == 200
|
|
result = response.json()
|
|
assert result["on_first_layer_complete"] is True
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_update_first_layer_complete_toggle(
|
|
self, async_client: AsyncClient, notification_provider_factory, db_session
|
|
):
|
|
"""CRITICAL: Verify first layer complete toggle persists correctly."""
|
|
provider = await notification_provider_factory(on_first_layer_complete=False)
|
|
|
|
response = await async_client.patch(
|
|
f"/api/v1/notifications/{provider.id}",
|
|
json={"on_first_layer_complete": True},
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
assert response.json()["on_first_layer_complete"] is True
|
|
|
|
# Verify persistence
|
|
response = await async_client.get(f"/api/v1/notifications/{provider.id}")
|
|
assert response.json()["on_first_layer_complete"] is True
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_first_layer_complete_independent_from_other_toggles(
|
|
self, async_client: AsyncClient, notification_provider_factory, db_session
|
|
):
|
|
"""Verify first layer complete is independent from bed cooled and print complete."""
|
|
provider = await notification_provider_factory(
|
|
on_print_complete=True,
|
|
on_bed_cooled=False,
|
|
on_first_layer_complete=True,
|
|
)
|
|
|
|
response = await async_client.get(f"/api/v1/notifications/{provider.id}")
|
|
result = response.json()
|
|
assert result["on_print_complete"] is True
|
|
assert result["on_bed_cooled"] is False
|
|
assert result["on_first_layer_complete"] is True
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_create_provider_with_missing_spool_assignment_toggle(self, async_client: AsyncClient):
|
|
"""Verify missing spool assignment toggle persists on create."""
|
|
data = {
|
|
"name": "Missing Spool Assignment Test",
|
|
"provider_type": "ntfy",
|
|
"config": {"server": "https://ntfy.sh", "topic": "test"},
|
|
"on_print_missing_spool_assignment": True,
|
|
}
|
|
|
|
response = await async_client.post("/api/v1/notifications/", json=data)
|
|
|
|
assert response.status_code == 200
|
|
result = response.json()
|
|
assert result["on_print_missing_spool_assignment"] is True
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_update_missing_spool_assignment_toggle(
|
|
self, async_client: AsyncClient, notification_provider_factory, db_session
|
|
):
|
|
"""CRITICAL: Verify missing spool assignment toggle persists correctly."""
|
|
provider = await notification_provider_factory(on_print_missing_spool_assignment=False)
|
|
|
|
response = await async_client.patch(
|
|
f"/api/v1/notifications/{provider.id}",
|
|
json={"on_print_missing_spool_assignment": True},
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
assert response.json()["on_print_missing_spool_assignment"] is True
|
|
|
|
response = await async_client.get(f"/api/v1/notifications/{provider.id}")
|
|
assert response.json()["on_print_missing_spool_assignment"] is True
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_update_billing_charge_failed_toggle(
|
|
self, async_client: AsyncClient, notification_provider_factory, db_session
|
|
):
|
|
"""Billing alerts can be enabled independently for each provider."""
|
|
provider = await notification_provider_factory(on_billing_charge_failed=True)
|
|
|
|
response = await async_client.patch(
|
|
f"/api/v1/notifications/{provider.id}",
|
|
json={"on_billing_charge_failed": False},
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
assert response.json()["on_billing_charge_failed"] is False
|
|
|
|
response = await async_client.get(f"/api/v1/notifications/{provider.id}")
|
|
assert response.json()["on_billing_charge_failed"] is False
|
|
|
|
# Every per-event toggle, across the hand-maintained field maps.
|
|
#
|
|
# These have to be exercised through the route, not the ORM: both
|
|
# directions of notifications.py are hand-maintained field-by-field maps,
|
|
# and a column missing from either one is invisible to any test that
|
|
# builds NotificationProvider objects directly. The failure mode is
|
|
# silent — NotificationProviderResponse inherits the field from
|
|
# NotificationProviderBase, so FastAPI serialises the schema default
|
|
# (False) instead of raising on the missing key, and the UI reads a
|
|
# toggle that is on in the database as off.
|
|
#
|
|
# The Home Assistant pair (#1148, #2824) was the first to be caught this
|
|
# way. The stock pair was caught by the same reasoning: its columns, its
|
|
# templates, its sending code and its whole UI shipped, but the schema
|
|
# never carried the fields, so Pydantic dropped them from every payload and
|
|
# the toggles could not be turned on at all.
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
@pytest.mark.parametrize("field", EVENT_TOGGLE_COLUMNS)
|
|
async def test_create_persists_and_returns_the_toggle(self, async_client: AsyncClient, field: str):
|
|
# Driven to whatever the column does not default to. Nine of these
|
|
# default to True on the model and on the response schema, so sending
|
|
# True and asserting True is answered by the default alone -- a field
|
|
# dropped by the create constructor still reads back True, and the
|
|
# mutation that would prove the constructor load-bearing survives.
|
|
target = TOGGLE_TARGET[field]
|
|
|
|
response = await async_client.post(
|
|
"/api/v1/notifications/",
|
|
json={
|
|
"name": "Sensor Alert Test",
|
|
"provider_type": "ntfy",
|
|
"config": {"server": "https://ntfy.sh", "topic": "test"},
|
|
field: target,
|
|
},
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
assert response.json()[field] is target
|
|
|
|
# Re-read it: a value dropped by the create constructor but echoed
|
|
# from the request body would still pass the assertion above.
|
|
provider_id = response.json()["id"]
|
|
response = await async_client.get(f"/api/v1/notifications/{provider_id}")
|
|
assert response.json()[field] is target
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
@pytest.mark.parametrize("field", EVENT_TOGGLE_COLUMNS)
|
|
async def test_patch_is_reflected_by_every_read_route(
|
|
self, async_client: AsyncClient, notification_provider_factory, field: str
|
|
):
|
|
"""PATCH already persisted (generic setattr loop) — the reads were the broken half.
|
|
|
|
Seeded at the target's opposite and driven to the target, so the value
|
|
asserted is never the one the column would have supplied on its own.
|
|
With ``True`` on both sides the nine True-default columns could not see
|
|
the half this test exists for: deleting a column from the read map left
|
|
them passing.
|
|
"""
|
|
target = TOGGLE_TARGET[field]
|
|
provider = await notification_provider_factory(**{field: not target})
|
|
|
|
response = await async_client.patch(f"/api/v1/notifications/{provider.id}", json={field: target})
|
|
assert response.status_code == 200
|
|
assert response.json()[field] is target
|
|
|
|
response = await async_client.get(f"/api/v1/notifications/{provider.id}")
|
|
assert response.json()[field] is target
|
|
|
|
response = await async_client.get("/api/v1/notifications/")
|
|
listed = next(p for p in response.json() if p["id"] == provider.id)
|
|
assert listed[field] is target
|
|
|
|
|
|
class TestNotificationTemplatesAPI:
|
|
"""Integration tests for /api/v1/notification-templates/ endpoints."""
|
|
|
|
@pytest.fixture
|
|
async def seeded_templates(self, db_session):
|
|
"""Seed notification templates for tests."""
|
|
from backend.app.models.notification_template import DEFAULT_TEMPLATES, NotificationTemplate
|
|
|
|
templates = []
|
|
for template_data in DEFAULT_TEMPLATES:
|
|
template = NotificationTemplate(**template_data)
|
|
db_session.add(template)
|
|
templates.append(template)
|
|
await db_session.commit()
|
|
for template in templates:
|
|
await db_session.refresh(template)
|
|
return templates
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_list_templates(self, async_client: AsyncClient, seeded_templates):
|
|
"""Verify default templates are seeded and can be listed."""
|
|
response = await async_client.get("/api/v1/notification-templates/")
|
|
|
|
assert response.status_code == 200
|
|
templates = response.json()
|
|
# Should have default templates seeded
|
|
assert len(templates) >= 1
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_get_template_by_id(self, async_client: AsyncClient, seeded_templates):
|
|
"""Verify template can be retrieved by ID."""
|
|
# Get first template ID from seeded data
|
|
template_id = seeded_templates[0].id
|
|
|
|
response = await async_client.get(f"/api/v1/notification-templates/{template_id}")
|
|
|
|
assert response.status_code == 200
|
|
template = response.json()
|
|
assert template["id"] == template_id
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_update_template(self, async_client: AsyncClient, seeded_templates):
|
|
"""Verify template can be updated."""
|
|
# Get first template
|
|
template_id = seeded_templates[0].id
|
|
|
|
# Update it (route uses PUT, not PATCH)
|
|
response = await async_client.put(
|
|
f"/api/v1/notification-templates/{template_id}",
|
|
json={
|
|
"title_template": "Custom Title: {printer}",
|
|
"body_template": "Custom body for {filename}",
|
|
},
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
result = response.json()
|
|
assert result["title_template"] == "Custom Title: {printer}"
|
|
assert result["body_template"] == "Custom body for {filename}"
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_reset_template_to_default(self, async_client: AsyncClient, seeded_templates):
|
|
"""Verify template can be reset to default."""
|
|
template_id = seeded_templates[0].id
|
|
|
|
response = await async_client.post(f"/api/v1/notification-templates/{template_id}/reset")
|
|
|
|
assert response.status_code == 200
|
|
result = response.json()
|
|
assert result["is_default"] is True
|
|
|
|
|
|
class TestHomeAssistantNotificationProvider:
|
|
"""Integration tests for Home Assistant notification provider."""
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_create_homeassistant_provider(self, async_client: AsyncClient):
|
|
"""Verify homeassistant notification provider can be created with empty config."""
|
|
data = {
|
|
"name": "HA Notifications",
|
|
"provider_type": "homeassistant",
|
|
"enabled": True,
|
|
"config": {},
|
|
"on_print_complete": True,
|
|
"on_print_failed": True,
|
|
}
|
|
|
|
response = await async_client.post("/api/v1/notifications/", json=data)
|
|
|
|
assert response.status_code == 200
|
|
result = response.json()
|
|
assert result["name"] == "HA Notifications"
|
|
assert result["provider_type"] == "homeassistant"
|
|
assert result["on_print_complete"] is True
|
|
assert result["on_print_failed"] is True
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_update_homeassistant_provider(
|
|
self, async_client: AsyncClient, notification_provider_factory, db_session
|
|
):
|
|
"""Verify homeassistant provider can be updated."""
|
|
provider = await notification_provider_factory(
|
|
name="HA Test",
|
|
provider_type="homeassistant",
|
|
config="{}",
|
|
)
|
|
|
|
response = await async_client.patch(
|
|
f"/api/v1/notifications/{provider.id}",
|
|
json={"on_print_start": True, "on_printer_offline": True},
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
result = response.json()
|
|
assert result["on_print_start"] is True
|
|
assert result["on_printer_offline"] is True
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_test_homeassistant_config_without_ha_settings(self, async_client: AsyncClient):
|
|
"""Verify test-config returns error when HA is not configured."""
|
|
response = await async_client.post(
|
|
"/api/v1/notifications/test-config",
|
|
json={"provider_type": "homeassistant", "config": {}},
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
result = response.json()
|
|
assert result["success"] is False
|
|
assert "not configured" in result["message"].lower() or "Home Assistant" in result["message"]
|