mirror of
https://github.com/maziggy/bambuddy.git
synced 2026-09-30 19:21:33 +02:00
maziggy
d9bc7ae47a
Adding on_stock_reorder_alert and on_stock_break_alert to the provider schema made them required on the way out as well as in: the response model inherits the write model. Every on_* column on notification_providers is nullable with no server default, and where the table was created from Base.metadata before run_migrations, the ALTER ... DEFAULT false that introduced those columns was swallowed as a duplicate and never backfilled existing rows. Those NULLs were harmless until the flags were read, at which point the row failed validation -- and a list is validated as a whole, so one row took every provider with it. The route returned 500 and the UI rendered an empty list, so configured providers looked deleted. Backfill them to off, which is what the sender already assumed: it selects providers with IS TRUE, so a NULL flag never sent anything. A NULL flag now also reads as off rather than failing the response, across all of them, so the next flag added to this schema cannot repeat it. Writes are unchanged.
698 lines
28 KiB
Python
698 lines
28 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
|
|
|
|
|
|
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"]
|
|
await db_session.execute(
|
|
text(f"UPDATE notification_providers SET {', '.join(f'{f} = NULL' for f in flags)} WHERE id = :id"),
|
|
{"id": provider.id},
|
|
)
|
|
await db_session.commit()
|
|
|
|
stored = await db_session.execute(
|
|
text(f"SELECT {', '.join(flags)} FROM notification_providers WHERE id = :id"), {"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
|
|
|
|
# Per-event toggles that live only in these 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",
|
|
["on_ha_sensor_alert", "on_location_ha_sensor_alert", "on_stock_reorder_alert", "on_stock_break_alert"],
|
|
)
|
|
async def test_create_persists_and_returns_the_toggle(self, async_client: AsyncClient, field: str):
|
|
response = await async_client.post(
|
|
"/api/v1/notifications/",
|
|
json={
|
|
"name": "Sensor Alert Test",
|
|
"provider_type": "ntfy",
|
|
"config": {"server": "https://ntfy.sh", "topic": "test"},
|
|
field: True,
|
|
},
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
assert response.json()[field] is True
|
|
|
|
# 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 True
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
@pytest.mark.parametrize(
|
|
"field",
|
|
["on_ha_sensor_alert", "on_location_ha_sensor_alert", "on_stock_reorder_alert", "on_stock_break_alert"],
|
|
)
|
|
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."""
|
|
provider = await notification_provider_factory(**{field: False})
|
|
|
|
response = await async_client.patch(f"/api/v1/notifications/{provider.id}", json={field: True})
|
|
assert response.status_code == 200
|
|
assert response.json()[field] is True
|
|
|
|
response = await async_client.get(f"/api/v1/notifications/{provider.id}")
|
|
assert response.json()[field] is True
|
|
|
|
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 True
|
|
|
|
|
|
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"]
|