|
|
@@ -0,0 +1,896 @@
|
|
|
+"""Lifecycle protocol tests with real persistence and a fake Notify gateway."""
|
|
|
+
|
|
|
+import json
|
|
|
+from datetime import datetime, timedelta
|
|
|
+from types import SimpleNamespace
|
|
|
+from unittest.mock import AsyncMock
|
|
|
+
|
|
|
+import pytest
|
|
|
+from sqlalchemy import delete, event, select, text
|
|
|
+from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
|
|
|
+
|
|
|
+import backend.app.models # noqa: F401
|
|
|
+from backend.app.core.database import Base
|
|
|
+from backend.app.models.notification import NotificationProvider
|
|
|
+from backend.app.models.notification_live_activity import NotificationLiveActivity
|
|
|
+from backend.app.models.printer import Printer
|
|
|
+from backend.app.services import notify_live_activities as module
|
|
|
+from backend.app.services.notify_client import NotifyError
|
|
|
+from backend.app.services.notify_live_activities import NotifyLiveActivityService, PrintSnapshot
|
|
|
+
|
|
|
+
|
|
|
+@pytest.fixture
|
|
|
+async def setup(tmp_path, monkeypatch):
|
|
|
+ engine = create_async_engine(f"sqlite+aiosqlite:///{tmp_path / 'activities.db'}")
|
|
|
+
|
|
|
+ @event.listens_for(engine.sync_engine, "connect")
|
|
|
+ def enable_foreign_keys(connection, _):
|
|
|
+ cursor = connection.cursor()
|
|
|
+ cursor.execute("PRAGMA foreign_keys=ON")
|
|
|
+ cursor.close()
|
|
|
+
|
|
|
+ async with engine.begin() as conn:
|
|
|
+ await conn.run_sync(Base.metadata.create_all)
|
|
|
+ factory = async_sessionmaker(engine, expire_on_commit=False)
|
|
|
+ async with factory() as db:
|
|
|
+ db.add(Printer(id=1, name="P1", serial_number="serial1", ip_address="127.0.0.1", access_code="12345678"))
|
|
|
+ db.add(
|
|
|
+ NotificationProvider(
|
|
|
+ id=1,
|
|
|
+ name="Phone",
|
|
|
+ provider_type="notify",
|
|
|
+ enabled=True,
|
|
|
+ config=json.dumps({"device_id": "ABC12345", "token": "secret", "live_activities": True}),
|
|
|
+ on_print_start=False,
|
|
|
+ on_print_progress=False,
|
|
|
+ on_print_complete=False,
|
|
|
+ daily_digest_enabled=True,
|
|
|
+ )
|
|
|
+ )
|
|
|
+ await db.commit()
|
|
|
+ api = SimpleNamespace(
|
|
|
+ start_activity=AsyncMock(return_value={"activityId": "LA123456"}),
|
|
|
+ get_activity=AsyncMock(return_value={"state": "active"}),
|
|
|
+ update_activity=AsyncMock(return_value={"success": True}),
|
|
|
+ end_activity=AsyncMock(return_value={"success": True}),
|
|
|
+ )
|
|
|
+ clock = [datetime(2026, 1, 1, 12)]
|
|
|
+ monkeypatch.setattr(module, "_now", lambda: clock[0])
|
|
|
+ service = NotifyLiveActivityService(factory, api)
|
|
|
+ monkeypatch.setattr(service, "_quiet", lambda provider: False)
|
|
|
+ yield service, api, factory, clock
|
|
|
+ await engine.dispose()
|
|
|
+
|
|
|
+
|
|
|
+def state(**values):
|
|
|
+ return (
|
|
|
+ SimpleNamespace(
|
|
|
+ connected=True,
|
|
|
+ state="RUNNING",
|
|
|
+ subtask_id="job1",
|
|
|
+ subtask_name="Test print",
|
|
|
+ current_print="test.3mf",
|
|
|
+ progress=42,
|
|
|
+ remaining_time=15,
|
|
|
+ layer_num=42,
|
|
|
+ total_layers=100,
|
|
|
+ raw_data={},
|
|
|
+ hms_errors=[],
|
|
|
+ temperatures={"nozzle": 210, "bed": 60},
|
|
|
+ stg_cur=-1,
|
|
|
+ **values,
|
|
|
+ )
|
|
|
+ if not values
|
|
|
+ else SimpleNamespace(
|
|
|
+ **{
|
|
|
+ **vars(state()),
|
|
|
+ **values,
|
|
|
+ }
|
|
|
+ )
|
|
|
+ )
|
|
|
+
|
|
|
+
|
|
|
+async def rows(factory):
|
|
|
+ async with factory() as db:
|
|
|
+ return (await db.scalars(select(NotificationLiveActivity))).all()
|
|
|
+
|
|
|
+
|
|
|
+async def start(setup):
|
|
|
+ service, api, factory, clock = setup
|
|
|
+ service.observe(1, state())
|
|
|
+ await service.tick()
|
|
|
+ return service, api, factory, clock
|
|
|
+
|
|
|
+
|
|
|
+@pytest.mark.asyncio
|
|
|
+async def test_start_independent_of_push_digest_and_group_thread(setup):
|
|
|
+ service, api, factory, clock = setup
|
|
|
+ async with factory() as db:
|
|
|
+ provider = await db.get(NotificationProvider, 1)
|
|
|
+ config = json.loads(provider.config)
|
|
|
+ config["group_type"] = "Workshop"
|
|
|
+ provider.config = json.dumps(config)
|
|
|
+ await db.commit()
|
|
|
+ await start(setup)
|
|
|
+ assert api.start_activity.await_count == 1
|
|
|
+ assert (await rows(factory))[0].activity_id == "LA123456"
|
|
|
+ await service.tick()
|
|
|
+ assert api.start_activity.await_count == 1
|
|
|
+
|
|
|
+
|
|
|
+@pytest.mark.asyncio
|
|
|
+async def test_startup_drops_rows_of_provider_deleted_before_cleanup_drained(setup):
|
|
|
+ service, api, factory, clock = await start(setup)
|
|
|
+ assert len(await rows(factory)) == 1
|
|
|
+ # Production SQLite does not enforce the cascade.
|
|
|
+ async with factory() as db:
|
|
|
+ await db.execute(text("PRAGMA foreign_keys=OFF"))
|
|
|
+ await db.execute(delete(NotificationProvider).where(NotificationProvider.id == 1))
|
|
|
+ await db.commit()
|
|
|
+ await db.execute(text("PRAGMA foreign_keys=ON"))
|
|
|
+ restarted = NotifyLiveActivityService(factory, api)
|
|
|
+ await restarted._purge_orphans()
|
|
|
+ assert await rows(factory) == []
|
|
|
+
|
|
|
+
|
|
|
+@pytest.mark.asyncio
|
|
|
+async def test_restart_reuses_exact_id_without_new_start(setup, monkeypatch):
|
|
|
+ service, api, factory, clock = await start(setup)
|
|
|
+ clock[0] += timedelta(minutes=2)
|
|
|
+ restarted = NotifyLiveActivityService(factory, api)
|
|
|
+ monkeypatch.setattr(restarted, "_quiet", lambda provider: False)
|
|
|
+ restarted.observe(1, state(progress=46))
|
|
|
+ await restarted.tick()
|
|
|
+ api.get_activity.assert_awaited_once_with("LA123456", "secret")
|
|
|
+ assert api.start_activity.await_count == 1
|
|
|
+ assert api.update_activity.call_args.args[0] == "LA123456"
|
|
|
+
|
|
|
+
|
|
|
+@pytest.mark.asyncio
|
|
|
+@pytest.mark.parametrize("phase,connected,expected", [("PAUSE", True, "Paused"), ("RUNNING", False, "Printer offline")])
|
|
|
+async def test_pause_disconnect_clear_countdown_then_resume(setup, phase, connected, expected):
|
|
|
+ service, api, factory, clock = await start(setup)
|
|
|
+ service.observe(1, state(state=phase, connected=connected))
|
|
|
+ await service.tick()
|
|
|
+ content = api.update_activity.call_args.args[2]
|
|
|
+ assert content["endsIn"] is None
|
|
|
+ assert content["status"] == expected
|
|
|
+ service.observe(1, state())
|
|
|
+ await service.tick()
|
|
|
+ assert api.update_activity.call_args.args[2]["endsIn"] == 900
|
|
|
+ assert api.start_activity.await_count == 1
|
|
|
+
|
|
|
+
|
|
|
+@pytest.mark.asyncio
|
|
|
+async def test_frozen_eta_does_not_drift(setup):
|
|
|
+ service, api, factory, clock = await start(setup)
|
|
|
+ clock[0] += timedelta(minutes=2)
|
|
|
+ await service.tick()
|
|
|
+ assert api.update_activity.call_args.args[2]["endsIn"] == 780
|
|
|
+ service.observe(1, state(remaining_time=10))
|
|
|
+ clock[0] += timedelta(minutes=1)
|
|
|
+ await service.tick()
|
|
|
+ assert api.update_activity.call_args.args[2]["endsIn"] == 600
|
|
|
+
|
|
|
+
|
|
|
+@pytest.mark.asyncio
|
|
|
+async def test_quiet_hours_block_starts_but_not_updates_or_ends(setup, monkeypatch):
|
|
|
+ service, api, factory, clock = setup
|
|
|
+ monkeypatch.setattr(service, "_quiet", lambda provider: True)
|
|
|
+ service.observe(1, state())
|
|
|
+ await service.tick()
|
|
|
+ api.start_activity.assert_not_awaited()
|
|
|
+ monkeypatch.setattr(service, "_quiet", lambda provider: False)
|
|
|
+ await service.tick()
|
|
|
+ monkeypatch.setattr(service, "_quiet", lambda provider: True)
|
|
|
+ clock[0] += timedelta(minutes=2)
|
|
|
+ await service.tick()
|
|
|
+ api.update_activity.assert_awaited_once()
|
|
|
+ service.print_finished(1, {"subtask_id": "job1", "status": "completed"})
|
|
|
+ await service.tick()
|
|
|
+ assert api.end_activity.call_args.args[2]["progress"] == 100
|
|
|
+
|
|
|
+
|
|
|
+@pytest.mark.asyncio
|
|
|
+@pytest.mark.parametrize("reason", ["dismissed", "dead-token", "never-started", "script"])
|
|
|
+async def test_terminal_suppression_survives_restart(setup, monkeypatch, reason):
|
|
|
+ service, api, factory, clock = await start(setup)
|
|
|
+ api.get_activity.return_value = {"state": "dismissed" if reason == "dismissed" else "ended", "endReason": reason}
|
|
|
+ clock[0] += timedelta(minutes=2)
|
|
|
+ await service.tick()
|
|
|
+ assert (await rows(factory))[0].state == "suppressed"
|
|
|
+ restarted = NotifyLiveActivityService(factory, api)
|
|
|
+ monkeypatch.setattr(restarted, "_quiet", lambda provider: False)
|
|
|
+ restarted.observe(1, state())
|
|
|
+ await restarted.tick()
|
|
|
+ assert api.start_activity.await_count == 1
|
|
|
+ restarted.observe(1, state(subtask_id="job2"))
|
|
|
+ await restarted.tick()
|
|
|
+ assert api.start_activity.await_count == 2
|
|
|
+
|
|
|
+
|
|
|
+@pytest.mark.asyncio
|
|
|
+@pytest.mark.parametrize("reason", ["expired", "overdue", "abandoned"])
|
|
|
+async def test_expiry_rolls_over_only_for_live_print(setup, reason):
|
|
|
+ service, api, factory, clock = await start(setup)
|
|
|
+ api.get_activity.return_value = {"state": "ended", "endReason": reason}
|
|
|
+ clock[0] += timedelta(hours=8)
|
|
|
+ api.start_activity.return_value = {"activityId": "LA654321"}
|
|
|
+ await service.tick()
|
|
|
+ assert api.start_activity.await_count == 2
|
|
|
+ assert (await rows(factory))[0].activity_id == "LA654321"
|
|
|
+
|
|
|
+
|
|
|
+@pytest.mark.asyncio
|
|
|
+async def test_eight_hour_ceiling_polls_before_rollover(setup):
|
|
|
+ service, api, factory, clock = await start(setup)
|
|
|
+ clock[0] += timedelta(hours=8)
|
|
|
+ await service.tick()
|
|
|
+ api.get_activity.assert_awaited_once()
|
|
|
+ api.end_activity.assert_awaited_once_with("LA123456", "secret")
|
|
|
+ assert api.start_activity.await_count == 2
|
|
|
+
|
|
|
+
|
|
|
+@pytest.mark.asyncio
|
|
|
+async def test_ambiguous_start_with_id_is_reconciled_without_duplication(setup):
|
|
|
+ service, api, factory, clock = setup
|
|
|
+ api.start_activity.side_effect = NotifyError(
|
|
|
+ "unknown", status_code=502, payload={"activityId": "LA123456", "deliveryState": "unknown"}
|
|
|
+ )
|
|
|
+ await start(setup)
|
|
|
+ assert (await rows(factory))[0].activity_id == "LA123456"
|
|
|
+ clock[0] += timedelta(minutes=2)
|
|
|
+ await service.tick()
|
|
|
+ api.get_activity.assert_awaited_once_with("LA123456", "secret")
|
|
|
+ assert api.start_activity.await_count == 1
|
|
|
+
|
|
|
+
|
|
|
+@pytest.mark.asyncio
|
|
|
+async def test_ambiguous_start_without_id_never_repeats(setup):
|
|
|
+ service, api, factory, clock = setup
|
|
|
+ api.start_activity.side_effect = NotifyError("timeout", delivery_state="unknown")
|
|
|
+ await start(setup)
|
|
|
+ clock[0] += timedelta(hours=2)
|
|
|
+ await service.tick()
|
|
|
+ assert api.start_activity.await_count == 1
|
|
|
+ assert (await rows(factory))[0].state == "suppressed"
|
|
|
+
|
|
|
+
|
|
|
+@pytest.mark.asyncio
|
|
|
+@pytest.mark.parametrize(
|
|
|
+ "error",
|
|
|
+ [
|
|
|
+ NotifyError("full", status_code=400, payload={"message": "Maximum 5 activities"}),
|
|
|
+ NotifyError("throttled", status_code=429, retry_after_seconds=1800),
|
|
|
+ NotifyError("not delivered", status_code=502, retry_after_seconds=1800, delivery_state="not-delivered"),
|
|
|
+ NotifyError("unavailable", status_code=503),
|
|
|
+ ],
|
|
|
+)
|
|
|
+async def test_retryable_starts_honor_backoff(setup, error):
|
|
|
+ service, api, factory, clock = setup
|
|
|
+ api.start_activity.side_effect = error
|
|
|
+ await start(setup)
|
|
|
+ await service.tick()
|
|
|
+ assert api.start_activity.await_count == 1
|
|
|
+ clock[0] += timedelta(minutes=31)
|
|
|
+ api.start_activity.side_effect = None
|
|
|
+ await service.tick()
|
|
|
+ assert api.start_activity.await_count == 2
|
|
|
+
|
|
|
+
|
|
|
+@pytest.mark.asyncio
|
|
|
+async def test_end_intent_survives_failure_restart_and_next_job(setup, monkeypatch):
|
|
|
+ service, api, factory, clock = await start(setup)
|
|
|
+ api.end_activity.side_effect = NotifyError("timeout", delivery_state="unknown")
|
|
|
+ service.print_finished(1, {"subtask_id": "job1", "status": "completed"})
|
|
|
+ await service.tick()
|
|
|
+ assert (await rows(factory))[0].state == "ending"
|
|
|
+ await service.tick()
|
|
|
+ assert api.end_activity.await_count == 1
|
|
|
+ restarted = NotifyLiveActivityService(factory, api)
|
|
|
+ monkeypatch.setattr(restarted, "_quiet", lambda provider: False)
|
|
|
+ restarted.observe(1, state(subtask_id="job2"))
|
|
|
+ clock[0] += timedelta(minutes=2)
|
|
|
+ api.end_activity.side_effect = None
|
|
|
+ await restarted.tick()
|
|
|
+ assert api.end_activity.call_args.args[2]["status"] == "Complete"
|
|
|
+ assert api.start_activity.await_count == 2
|
|
|
+ assert (await rows(factory))[0].state == "ended"
|
|
|
+
|
|
|
+
|
|
|
+@pytest.mark.asyncio
|
|
|
+async def test_late_complete_does_not_end_new_job(setup):
|
|
|
+ service, api, factory, clock = await start(setup)
|
|
|
+ service.observe(1, state(subtask_id="job2"))
|
|
|
+ await service.tick()
|
|
|
+ api.end_activity.reset_mock()
|
|
|
+ service.print_finished(1, {"subtask_id": "job1", "status": "completed"})
|
|
|
+ await service.tick()
|
|
|
+ api.end_activity.assert_not_awaited()
|
|
|
+
|
|
|
+
|
|
|
+@pytest.mark.asyncio
|
|
|
+async def test_fallback_survives_delta_and_late_job_id(setup):
|
|
|
+ service, api, factory, clock = setup
|
|
|
+ service.print_started(1, state(subtask_id=None, raw_data={"gcode_start_time": 12345}))
|
|
|
+ await service.tick()
|
|
|
+ clock[0] += timedelta(minutes=2)
|
|
|
+ service.observe(1, state(subtask_id=None, raw_data={}))
|
|
|
+ await service.tick()
|
|
|
+ service.observe(1, state(subtask_id="finally-known"))
|
|
|
+ clock[0] += timedelta(minutes=2)
|
|
|
+ await service.tick()
|
|
|
+ assert api.start_activity.await_count == 1
|
|
|
+ assert api.end_activity.await_count == 0
|
|
|
+
|
|
|
+
|
|
|
+@pytest.mark.asyncio
|
|
|
+async def test_scope_and_disabled_provider_end_tiles(setup):
|
|
|
+ service, api, factory, clock = await start(setup)
|
|
|
+ async with factory() as db:
|
|
|
+ provider = await db.get(NotificationProvider, 1)
|
|
|
+ provider.enabled = False
|
|
|
+ await db.commit()
|
|
|
+ await service.tick()
|
|
|
+ assert api.end_activity.await_count == 1
|
|
|
+ assert (await rows(factory))[0].state == "ended"
|
|
|
+
|
|
|
+
|
|
|
+@pytest.mark.asyncio
|
|
|
+async def test_delete_printer_retains_id_until_remote_end(setup):
|
|
|
+ service, api, factory, clock = await start(setup)
|
|
|
+ async with factory() as db:
|
|
|
+ await db.delete(await db.get(Printer, 1))
|
|
|
+ await db.commit()
|
|
|
+ await service.tick()
|
|
|
+ assert api.end_activity.await_count == 1
|
|
|
+ assert (await rows(factory))[0].state == "ended"
|
|
|
+
|
|
|
+
|
|
|
+@pytest.mark.asyncio
|
|
|
+async def test_cleanup_uses_old_token_and_removes_ownership(setup):
|
|
|
+ service, api, factory, clock = await start(setup)
|
|
|
+ await service.cleanup_provider(1, {"device_id": "ABC12345", "token": "secret"})
|
|
|
+ api.end_activity.assert_awaited_once_with("LA123456", "secret")
|
|
|
+ assert await rows(factory) == []
|
|
|
+
|
|
|
+
|
|
|
+@pytest.mark.asyncio
|
|
|
+async def test_http_never_holds_database_session_and_concurrent_ticks_deduplicate(setup):
|
|
|
+ import asyncio
|
|
|
+
|
|
|
+ service, api, factory, clock = setup
|
|
|
+ entered = 0
|
|
|
+
|
|
|
+ class TrackedSession:
|
|
|
+ async def __aenter__(self):
|
|
|
+ nonlocal entered
|
|
|
+ entered += 1
|
|
|
+ self.db = factory()
|
|
|
+ return await self.db.__aenter__()
|
|
|
+
|
|
|
+ async def __aexit__(self, *args):
|
|
|
+ nonlocal entered
|
|
|
+ await self.db.__aexit__(*args)
|
|
|
+ entered -= 1
|
|
|
+
|
|
|
+ service._session = TrackedSession
|
|
|
+
|
|
|
+ async def send(*args):
|
|
|
+ assert entered == 0
|
|
|
+ await asyncio.sleep(0)
|
|
|
+ return {"activityId": "LA123456"}
|
|
|
+
|
|
|
+ api.start_activity.side_effect = send
|
|
|
+ service.observe(1, state())
|
|
|
+ await asyncio.gather(service.tick(), service.tick(), service.tick())
|
|
|
+ assert api.start_activity.await_count == 1
|
|
|
+
|
|
|
+
|
|
|
+def test_advanced_content_and_utf8_budget():
|
|
|
+ snapshot = PrintSnapshot.from_state(state(subtask_name="🔒" * 255, remaining_time=1500))
|
|
|
+ config = {
|
|
|
+ "live_activity_style": "segments",
|
|
|
+ "live_activity_metrics": ["progress", "eta", "layers", "nozzle", "bed", "chamber"],
|
|
|
+ "live_activity_button_url": "https://bambuddy.example/" + "a" * 480,
|
|
|
+ "live_activity_privacy": True,
|
|
|
+ }
|
|
|
+ content = snapshot.content("🖨" * 100, config)
|
|
|
+ assert "🔒" not in content["body"]
|
|
|
+ assert content["steps"] == 10 and content["step"] == 4
|
|
|
+ assert content["endsIn"] is None and content["trailing"] == "25h 0m"
|
|
|
+ assert len(content["metrics"]) == 5
|
|
|
+ assert len(json.dumps(content, ensure_ascii=False).encode()) < 2048
|
|
|
+ assert content["button"] is None or content["button"]["open"] is True
|
|
|
+ assert snapshot.content("P1", {"live_activity_style": "none"})["progress"] is None
|
|
|
+
|
|
|
+
|
|
|
+@pytest.mark.asyncio
|
|
|
+async def test_fallback_dismissal_follows_late_id_across_restart(setup, monkeypatch):
|
|
|
+ service, api, factory, clock = setup
|
|
|
+ service.print_started(1, state(subtask_id=None))
|
|
|
+ await service.tick()
|
|
|
+ clock[0] += timedelta(minutes=2)
|
|
|
+ api.get_activity.return_value = {"state": "dismissed", "endReason": "dismissed"}
|
|
|
+ await service.tick()
|
|
|
+ service.observe(1, state(subtask_id="late-id"))
|
|
|
+ await service.tick()
|
|
|
+ assert (await rows(factory))[0].job_key == "job:late-id"
|
|
|
+ restarted = NotifyLiveActivityService(factory, api)
|
|
|
+ monkeypatch.setattr(restarted, "_quiet", lambda provider: False)
|
|
|
+ restarted.observe(1, state(subtask_id="late-id"))
|
|
|
+ await restarted.tick()
|
|
|
+ assert api.start_activity.await_count == 1
|
|
|
+
|
|
|
+
|
|
|
+@pytest.mark.asyncio
|
|
|
+async def test_end_retry_without_any_printer_snapshot(setup):
|
|
|
+ service, api, factory, clock = await start(setup)
|
|
|
+ api.end_activity.side_effect = NotifyError("unavailable", status_code=503)
|
|
|
+ service.print_finished(1, {"subtask_id": "job1", "status": "failed"})
|
|
|
+ await service.tick()
|
|
|
+ restarted = NotifyLiveActivityService(factory, api)
|
|
|
+ clock[0] += timedelta(minutes=2)
|
|
|
+ api.end_activity.side_effect = None
|
|
|
+ await restarted.tick()
|
|
|
+ assert api.end_activity.call_args.args[2]["status"] == "Failed"
|
|
|
+ assert (await rows(factory))[0].state == "ended"
|
|
|
+
|
|
|
+
|
|
|
+@pytest.mark.asyncio
|
|
|
+async def test_printer_scope_new_provider_separate_ownership(setup):
|
|
|
+ service, api, factory, clock = setup
|
|
|
+ async with factory() as db:
|
|
|
+ db.add(Printer(id=2, name="P2", serial_number="serial2", ip_address="127.0.0.2", access_code="12345678"))
|
|
|
+ db.add(
|
|
|
+ NotificationProvider(
|
|
|
+ id=2,
|
|
|
+ name="Other Phone",
|
|
|
+ provider_type="notify",
|
|
|
+ printer_id=2,
|
|
|
+ config=json.dumps({"device_id": "ABC12346", "token": "other", "live_activities": True}),
|
|
|
+ )
|
|
|
+ )
|
|
|
+ await db.commit()
|
|
|
+ await start(setup)
|
|
|
+ assert api.start_activity.await_count == 1
|
|
|
+ async with factory() as db:
|
|
|
+ provider = await db.get(NotificationProvider, 2)
|
|
|
+ provider.printer_id = 1
|
|
|
+ await db.commit()
|
|
|
+ await service.tick()
|
|
|
+ assert api.start_activity.await_count == 2
|
|
|
+ assert {r.provider_id for r in await rows(factory)} == {1, 2}
|
|
|
+
|
|
|
+
|
|
|
+@pytest.mark.asyncio
|
|
|
+async def test_unknown_start_record_after_process_crash_is_never_restarted(setup):
|
|
|
+ service, api, factory, clock = await start(setup)
|
|
|
+ async with factory() as db:
|
|
|
+ row = await db.get(NotificationLiveActivity, 1)
|
|
|
+ row.activity_id = None
|
|
|
+ row.state = "uncertain"
|
|
|
+ await db.commit()
|
|
|
+ clock[0] += timedelta(minutes=2)
|
|
|
+ await service.tick()
|
|
|
+ assert api.start_activity.await_count == 1
|
|
|
+ assert (await rows(factory))[0].state == "suppressed"
|
|
|
+
|
|
|
+
|
|
|
+@pytest.mark.asyncio
|
|
|
+async def test_410_is_polled_for_reason_before_any_rollover(setup):
|
|
|
+ service, api, factory, clock = await start(setup)
|
|
|
+ api.update_activity.side_effect = NotifyError("gone", status_code=410)
|
|
|
+ clock[0] += timedelta(minutes=2)
|
|
|
+ await service.tick()
|
|
|
+ api.get_activity.return_value = {"state": "dismissed", "endReason": "dismissed"}
|
|
|
+ clock[0] += timedelta(minutes=2)
|
|
|
+ await service.tick()
|
|
|
+ assert api.start_activity.await_count == 1
|
|
|
+ assert (await rows(factory))[0].state == "suppressed"
|
|
|
+
|
|
|
+
|
|
|
+def fault(description="Nozzle temperature malfunction", severity=2, full_code="05004003", actions=None):
|
|
|
+ return SimpleNamespace(description=description, severity=severity, full_code=full_code, actions=actions)
|
|
|
+
|
|
|
+
|
|
|
+@pytest.mark.asyncio
|
|
|
+async def test_running_printer_fault_replaces_counter_and_restores_after_clear(setup):
|
|
|
+ service, api, factory, clock = await start(setup)
|
|
|
+ async with factory() as db:
|
|
|
+ provider = await db.get(NotificationProvider, 1)
|
|
|
+ config = json.loads(provider.config)
|
|
|
+ config.update(live_activity_metrics=["progress", "eta"], live_activity_stage=True)
|
|
|
+ provider.config = json.dumps(config)
|
|
|
+ await db.commit()
|
|
|
+ service.observe(1, state(hms_errors=[fault()]))
|
|
|
+ await service.tick()
|
|
|
+ content = api.update_activity.call_args.args[2]
|
|
|
+ assert content["status"] == "Printer error"
|
|
|
+ assert content["body"] == "Nozzle temperature malfunction"
|
|
|
+ assert content["endsIn"] is None
|
|
|
+ assert content["metrics"] is None
|
|
|
+ service.observe(1, state(hms_errors=[]))
|
|
|
+ await service.tick()
|
|
|
+ content = api.update_activity.call_args.args[2]
|
|
|
+ assert content["status"] == "Printing"
|
|
|
+ assert content["endsIn"] == 900
|
|
|
+ assert len(content["metrics"]) == 2
|
|
|
+ assert api.start_activity.await_count == 1
|
|
|
+
|
|
|
+
|
|
|
+def test_runout_and_actionable_level_three_prompt():
|
|
|
+ snapshot = PrintSnapshot.from_state(state(hms_errors=[fault("Filament has run out", 3)]))
|
|
|
+ content = snapshot.content("P1", {"live_activity_privacy": True})
|
|
|
+ assert content["status"] == "Filament runout"
|
|
|
+ assert content["endsIn"] is None
|
|
|
+ assert "Test print" not in content["body"]
|
|
|
+ snapshot = PrintSnapshot.from_state(
|
|
|
+ state(hms_errors=[fault("Heatbreak fan problem", 3, "0000000012345678", ["retry"])])
|
|
|
+ )
|
|
|
+ assert snapshot.content("P1")["status"] == "Printer error"
|
|
|
+
|
|
|
+
|
|
|
+@pytest.mark.parametrize(
|
|
|
+ "error",
|
|
|
+ [
|
|
|
+ fault("The cover is open", 3, "0000000012345678"),
|
|
|
+ fault("Invalid level", 0),
|
|
|
+ fault("", 2),
|
|
|
+ ],
|
|
|
+)
|
|
|
+def test_advisory_does_not_interrupt_live_countdown(error):
|
|
|
+ content = PrintSnapshot.from_state(state(hms_errors=[error])).content("P1")
|
|
|
+ assert content["status"] == "Printing"
|
|
|
+ assert content["endsIn"] == 900
|
|
|
+
|
|
|
+
|
|
|
+@pytest.mark.asyncio
|
|
|
+@pytest.mark.parametrize("device_id", ["GRP12345", "WB123456", "MC123456"])
|
|
|
+async def test_non_ios_devices_never_start_activity(setup, device_id):
|
|
|
+ service, api, factory, clock = setup
|
|
|
+ async with factory() as db:
|
|
|
+ provider = await db.get(NotificationProvider, 1)
|
|
|
+ config = json.loads(provider.config)
|
|
|
+ config["device_id"] = device_id
|
|
|
+ provider.config = json.dumps(config)
|
|
|
+ await db.commit()
|
|
|
+ service.observe(1, state())
|
|
|
+ await service.tick()
|
|
|
+ api.start_activity.assert_not_awaited()
|
|
|
+
|
|
|
+
|
|
|
+@pytest.mark.asyncio
|
|
|
+async def test_new_start_resets_previous_progress_and_eta_until_fresh_telemetry(setup):
|
|
|
+ service, api, factory, clock = setup
|
|
|
+ service.print_started(1, state(progress=99, remaining_time=1))
|
|
|
+ await service.tick()
|
|
|
+ assert api.start_activity.call_args.args[2]["progress"] == 0
|
|
|
+ assert api.start_activity.call_args.args[2]["endsIn"] is None
|
|
|
+ service.observe(1, state(progress=99, remaining_time=1))
|
|
|
+ clock[0] += timedelta(seconds=61)
|
|
|
+ await service.tick()
|
|
|
+ assert api.update_activity.call_args.args[2]["progress"] == 0
|
|
|
+ assert api.update_activity.call_args.args[2]["endsIn"] is None
|
|
|
+ service.observe(1, state(progress=2, remaining_time=50))
|
|
|
+ clock[0] += timedelta(seconds=61)
|
|
|
+ await service.tick()
|
|
|
+ assert api.update_activity.call_args.args[2]["progress"] == 2
|
|
|
+ assert api.update_activity.call_args.args[2]["endsIn"] == 3000
|
|
|
+
|
|
|
+
|
|
|
+@pytest.mark.asyncio
|
|
|
+async def test_old_tile_must_end_before_starting_next_job(setup):
|
|
|
+ service, api, factory, clock = await start(setup)
|
|
|
+ api.end_activity.side_effect = NotifyError("unavailable", status_code=503)
|
|
|
+ service.observe(1, state(subtask_id="job2"))
|
|
|
+ await service.tick()
|
|
|
+ assert api.start_activity.await_count == 1
|
|
|
+ assert (await rows(factory))[0].state == "ending"
|
|
|
+ api.end_activity.side_effect = None
|
|
|
+ clock[0] += timedelta(minutes=2)
|
|
|
+ await service.tick()
|
|
|
+ assert api.start_activity.await_count == 2
|
|
|
+
|
|
|
+
|
|
|
+@pytest.mark.asyncio
|
|
|
+async def test_complete_matches_learned_job_id_for_fallback_row(setup):
|
|
|
+ service, api, factory, clock = setup
|
|
|
+ service.print_started(1, state(subtask_id=None))
|
|
|
+ await service.tick()
|
|
|
+ service.observe(1, state(subtask_id="learned"))
|
|
|
+ await service.tick()
|
|
|
+ service.print_finished(1, {"raw_data": {"subtask_id": "learned"}, "status": "completed"})
|
|
|
+ await service.tick()
|
|
|
+ assert api.end_activity.call_args.args[2]["status"] == "Complete"
|
|
|
+ assert (await rows(factory))[0].state == "ended"
|
|
|
+
|
|
|
+
|
|
|
+def test_preparation_does_not_display_previous_print_progress():
|
|
|
+ snapshot = PrintSnapshot.from_state(state(layer_num=0, total_layers=100, progress=85))
|
|
|
+ content = snapshot.content("P1", {"live_activity_metrics": ["progress"]})
|
|
|
+ assert content["status"] == "Preparing"
|
|
|
+ assert content["progress"] == 0
|
|
|
+ assert content["metrics"][0]["value"] == "0%"
|
|
|
+
|
|
|
+
|
|
|
+@pytest.mark.asyncio
|
|
|
+async def test_reenable_resumes_current_print_after_successful_cleanup(setup):
|
|
|
+ service, api, factory, clock = await start(setup)
|
|
|
+ async with factory() as db:
|
|
|
+ provider = await db.get(NotificationProvider, 1)
|
|
|
+ provider.enabled = False
|
|
|
+ await db.commit()
|
|
|
+ await service.tick()
|
|
|
+ assert (await rows(factory))[0].end_reason == "disabled"
|
|
|
+ async with factory() as db:
|
|
|
+ provider = await db.get(NotificationProvider, 1)
|
|
|
+ provider.enabled = True
|
|
|
+ await db.commit()
|
|
|
+ await service.tick()
|
|
|
+ assert api.start_activity.await_count == 1 # Wait for fresh telemetry after dormant time.
|
|
|
+ service.observe(1, state())
|
|
|
+ await service.tick()
|
|
|
+ assert api.start_activity.await_count == 2
|
|
|
+
|
|
|
+
|
|
|
+@pytest.mark.asyncio
|
|
|
+async def test_reenable_waits_for_failed_cleanup_before_restarting(setup):
|
|
|
+ service, api, factory, clock = await start(setup)
|
|
|
+ async with factory() as db:
|
|
|
+ provider = await db.get(NotificationProvider, 1)
|
|
|
+ provider.enabled = False
|
|
|
+ await db.commit()
|
|
|
+ api.end_activity.side_effect = NotifyError("unavailable", status_code=503)
|
|
|
+ await service.tick()
|
|
|
+ async with factory() as db:
|
|
|
+ provider = await db.get(NotificationProvider, 1)
|
|
|
+ provider.enabled = True
|
|
|
+ await db.commit()
|
|
|
+ await service.tick()
|
|
|
+ assert api.start_activity.await_count == 1
|
|
|
+ api.end_activity.side_effect = None
|
|
|
+ clock[0] += timedelta(minutes=2)
|
|
|
+ await service.tick()
|
|
|
+ service.observe(1, state())
|
|
|
+ await service.tick()
|
|
|
+ assert api.start_activity.await_count == 2
|
|
|
+
|
|
|
+
|
|
|
+@pytest.mark.parametrize("config", [{}, {"live_activity_metrics": []}])
|
|
|
+def test_default_tile_uses_native_timer_without_metric_chips(config):
|
|
|
+ content = PrintSnapshot.from_state(state()).content("P1", config)
|
|
|
+ assert content["endsIn"] == 900
|
|
|
+ assert content["metrics"] is None
|
|
|
+ assert content["trailing"] is None
|
|
|
+
|
|
|
+
|
|
|
+@pytest.mark.asyncio
|
|
|
+async def test_disabling_optional_metrics_clears_chips_and_preserves_timer(setup):
|
|
|
+ service, api, factory, clock = setup
|
|
|
+ async with factory() as db:
|
|
|
+ provider = await db.get(NotificationProvider, 1)
|
|
|
+ config = json.loads(provider.config)
|
|
|
+ config["live_activity_metrics"] = ["progress", "eta"]
|
|
|
+ provider.config = json.dumps(config)
|
|
|
+ await db.commit()
|
|
|
+ await start(setup)
|
|
|
+ content = api.start_activity.call_args.args[2]
|
|
|
+ assert content["endsIn"] == 900
|
|
|
+ assert content["metrics"][0] == {"label": "Progress", "value": "42%", "unit": ""}
|
|
|
+ async with factory() as db:
|
|
|
+ provider = await db.get(NotificationProvider, 1)
|
|
|
+ config["live_activity_metrics"] = []
|
|
|
+ provider.config = json.dumps(config)
|
|
|
+ await db.commit()
|
|
|
+ clock[0] += timedelta(minutes=1)
|
|
|
+ await service.tick()
|
|
|
+ content = api.update_activity.call_args.args[2]
|
|
|
+ assert content["metrics"] is None
|
|
|
+ assert content["endsIn"] == 840
|
|
|
+ assert api.start_activity.await_count == 1
|
|
|
+
|
|
|
+
|
|
|
+@pytest.mark.asyncio
|
|
|
+async def test_dormant_worker_has_no_periodic_sql_or_mqtt_wakeups(setup):
|
|
|
+ import asyncio
|
|
|
+
|
|
|
+ from sqlalchemy import event
|
|
|
+
|
|
|
+ service, api, factory, clock = setup
|
|
|
+ async with factory() as db:
|
|
|
+ provider = await db.get(NotificationProvider, 1)
|
|
|
+ provider.enabled = False
|
|
|
+ await db.commit()
|
|
|
+ queries = []
|
|
|
+ engine = factory.kw["bind"]
|
|
|
+
|
|
|
+ def record(connection, cursor, statement, parameters, context, many):
|
|
|
+ queries.append(statement)
|
|
|
+
|
|
|
+ event.listen(engine.sync_engine, "before_cursor_execute", record)
|
|
|
+ service.worker_interval = 0.01
|
|
|
+ service.start()
|
|
|
+ try:
|
|
|
+ for _ in range(100):
|
|
|
+ if not service._initial_discovery:
|
|
|
+ break
|
|
|
+ await asyncio.sleep(0.001)
|
|
|
+ queries.clear()
|
|
|
+ service.observe(1, state())
|
|
|
+ service.print_started(1, state())
|
|
|
+ service.print_finished(1, {"subtask_id": "job1", "status": "completed"})
|
|
|
+ await asyncio.sleep(0.05)
|
|
|
+ assert queries == []
|
|
|
+ assert service._snapshots == {}
|
|
|
+ api.start_activity.assert_not_awaited()
|
|
|
+ finally:
|
|
|
+ await service.close()
|
|
|
+ event.remove(engine.sync_engine, "before_cursor_execute", record)
|
|
|
+
|
|
|
+
|
|
|
+@pytest.mark.asyncio
|
|
|
+async def test_startup_waits_for_real_status_before_touching_existing_activity(setup):
|
|
|
+ service, api, factory, clock = await start(setup)
|
|
|
+ clock[0] += timedelta(minutes=2)
|
|
|
+ restarted = NotifyLiveActivityService(factory, api)
|
|
|
+ await restarted.tick()
|
|
|
+ api.get_activity.assert_not_awaited()
|
|
|
+ restarted.observe(1, state(state="unknown", progress=0))
|
|
|
+ await restarted.tick()
|
|
|
+ api.get_activity.assert_not_awaited()
|
|
|
+ restarted.observe(1, state())
|
|
|
+ await restarted.tick()
|
|
|
+ api.get_activity.assert_awaited_once()
|
|
|
+
|
|
|
+
|
|
|
+@pytest.mark.asyncio
|
|
|
+@pytest.mark.parametrize("connected", [None, False, True])
|
|
|
+async def test_startup_grace_expires_without_real_status_and_reconnect_resumes(setup, connected):
|
|
|
+ service, api, factory, clock = await start(setup)
|
|
|
+ clock[0] += timedelta(minutes=2)
|
|
|
+ restarted = NotifyLiveActivityService(factory, api)
|
|
|
+ # Importing the service before worker startup must not consume the grace.
|
|
|
+ clock[0] += timedelta(minutes=5)
|
|
|
+
|
|
|
+ # Neither complete silence nor broker-only status may extend the grace.
|
|
|
+ for seconds in (0, 90, 29):
|
|
|
+ clock[0] += timedelta(seconds=seconds)
|
|
|
+ if connected is not None:
|
|
|
+ restarted.observe(1, state(state="unknown", connected=connected, progress=0))
|
|
|
+ await restarted.tick()
|
|
|
+ api.get_activity.assert_not_awaited()
|
|
|
+ api.update_activity.assert_not_awaited()
|
|
|
+
|
|
|
+ clock[0] += timedelta(seconds=1)
|
|
|
+ await restarted.tick()
|
|
|
+ api.get_activity.assert_awaited_once_with("LA123456", "secret")
|
|
|
+ content = api.update_activity.call_args.args[2]
|
|
|
+ assert content["status"] == "Printer offline"
|
|
|
+ assert content["endsIn"] is None
|
|
|
+ assert content["progress"] == 42
|
|
|
+ saved = (await rows(factory))[0]
|
|
|
+ assert saved.eta_seconds is None
|
|
|
+ assert saved.eta_deadline is None
|
|
|
+ assert saved.activity_id == "LA123456"
|
|
|
+ api.end_activity.assert_not_awaited()
|
|
|
+ assert api.start_activity.await_count == 1
|
|
|
+
|
|
|
+ # A real status resumes the same tile immediately, even during throttling.
|
|
|
+ restarted.observe(1, state(progress=46, remaining_time=10))
|
|
|
+ await restarted.tick()
|
|
|
+ assert api.update_activity.await_count == 2
|
|
|
+ assert api.update_activity.call_args.args[0] == "LA123456"
|
|
|
+ content = api.update_activity.call_args.args[2]
|
|
|
+ assert content["status"] == "Printing"
|
|
|
+ assert content["endsIn"] == 600
|
|
|
+ assert content["progress"] == 46
|
|
|
+ assert api.start_activity.await_count == 1
|
|
|
+
|
|
|
+
|
|
|
+@pytest.mark.asyncio
|
|
|
+@pytest.mark.parametrize("reason", ["expired", "dismissed"])
|
|
|
+async def test_startup_grace_never_restarts_an_offline_printers_activity(setup, reason):
|
|
|
+ service, api, factory, clock = await start(setup)
|
|
|
+ restarted = NotifyLiveActivityService(factory, api)
|
|
|
+ await restarted.tick()
|
|
|
+ api.get_activity.return_value = {"state": "ended", "endReason": reason}
|
|
|
+ clock[0] += timedelta(minutes=2)
|
|
|
+ await restarted.tick()
|
|
|
+ api.get_activity.assert_awaited_once_with("LA123456", "secret")
|
|
|
+ api.update_activity.assert_not_awaited()
|
|
|
+ api.end_activity.assert_not_awaited()
|
|
|
+ assert api.start_activity.await_count == 1
|
|
|
+ assert (await rows(factory))[0].state == ("pending" if reason == "expired" else "suppressed")
|
|
|
+
|
|
|
+
|
|
|
+@pytest.mark.asyncio
|
|
|
+async def test_startup_offline_clears_saved_eta_labels(setup):
|
|
|
+ service, api, factory, clock = setup
|
|
|
+ async with factory() as db:
|
|
|
+ provider = await db.get(NotificationProvider, 1)
|
|
|
+ config = json.loads(provider.config)
|
|
|
+ config["live_activity_metrics"] = ["progress", "eta", "nozzle"]
|
|
|
+ provider.config = json.dumps(config)
|
|
|
+ await db.commit()
|
|
|
+ service.observe(1, state(remaining_time=25 * 60))
|
|
|
+ await service.tick()
|
|
|
+ original = api.start_activity.call_args.args[2]
|
|
|
+ assert original["trailing"] == "25h 0m"
|
|
|
+ assert any(metric["label"] == "Remaining" for metric in original["metrics"])
|
|
|
+
|
|
|
+ restarted = NotifyLiveActivityService(factory, api)
|
|
|
+ await restarted.tick()
|
|
|
+ clock[0] += timedelta(minutes=2)
|
|
|
+ await restarted.tick()
|
|
|
+ content = api.update_activity.call_args.args[2]
|
|
|
+ assert content["status"] == "Printer offline"
|
|
|
+ assert content["endsIn"] is None
|
|
|
+ assert content["trailing"] is None
|
|
|
+ assert content["metrics"] is None
|
|
|
+ assert content["progress"] == 42
|
|
|
+
|
|
|
+
|
|
|
+@pytest.mark.asyncio
|
|
|
+async def test_scheduled_cleanup_does_not_wait_for_create_and_keeps_late_handle(setup):
|
|
|
+ import asyncio
|
|
|
+
|
|
|
+ service, api, factory, clock = setup
|
|
|
+ entered, release = asyncio.Event(), asyncio.Event()
|
|
|
+
|
|
|
+ async def create(*args):
|
|
|
+ entered.set()
|
|
|
+ await release.wait()
|
|
|
+ return {"activityId": "LA123456"}
|
|
|
+
|
|
|
+ api.start_activity.side_effect = create
|
|
|
+ service.observe(1, state())
|
|
|
+ tick = asyncio.create_task(service.tick())
|
|
|
+ await entered.wait()
|
|
|
+ await asyncio.wait_for(service.schedule_cleanup(1, {"device_id": "ABC12345", "token": "secret"}), 0.2)
|
|
|
+ async with factory() as db:
|
|
|
+ await db.delete(await db.get(NotificationProvider, 1))
|
|
|
+ await db.commit()
|
|
|
+ release.set()
|
|
|
+ await tick
|
|
|
+ try:
|
|
|
+ for _ in range(100):
|
|
|
+ if api.end_activity.await_count:
|
|
|
+ break
|
|
|
+ await asyncio.sleep(0.005)
|
|
|
+ api.end_activity.assert_awaited_once_with("LA123456", "secret")
|
|
|
+ assert await rows(factory) == []
|
|
|
+ finally:
|
|
|
+ await service.close()
|
|
|
+
|
|
|
+
|
|
|
+@pytest.mark.asyncio
|
|
|
+async def test_boot_cleans_disabled_activity_without_printer_queries_or_pruning(setup):
|
|
|
+ import asyncio
|
|
|
+
|
|
|
+ from sqlalchemy import event
|
|
|
+
|
|
|
+ service, api, factory, clock = await start(setup)
|
|
|
+ async with factory() as db:
|
|
|
+ (await db.get(NotificationProvider, 1)).enabled = False
|
|
|
+ await db.commit()
|
|
|
+ statements = []
|
|
|
+ engine = factory.kw["bind"]
|
|
|
+
|
|
|
+ def record(connection, cursor, statement, parameters, context, many):
|
|
|
+ statements.append(statement.lower())
|
|
|
+
|
|
|
+ event.listen(engine.sync_engine, "before_cursor_execute", record)
|
|
|
+ restarted = NotifyLiveActivityService(factory, api)
|
|
|
+ restarted.start()
|
|
|
+ try:
|
|
|
+ for _ in range(100):
|
|
|
+ if (await rows(factory))[0].state == "ended":
|
|
|
+ break
|
|
|
+ await asyncio.sleep(0.005)
|
|
|
+ assert (await rows(factory))[0].state == "ended"
|
|
|
+ assert not any("from printers" in statement for statement in statements)
|
|
|
+ assert not any(statement.startswith("delete from notification_live_activities") for statement in statements)
|
|
|
+ finally:
|
|
|
+ await restarted.close()
|
|
|
+ event.remove(engine.sync_engine, "before_cursor_execute", record)
|