"""Unit tests for the Planka webhook parser. Pure-function tests against ``parse_webhook`` using realistic Planka webhook payload shapes. The parser is forgiving about missing ``included`` data (older Planka builds), so we mix payloads with and without it to catch regressions in the fallback paths. """ from __future__ import annotations from notify_bridge_core.models.events import EventType from notify_bridge_core.providers.base import ServiceProviderType from notify_bridge_core.providers.planka.event_parser import parse_webhook _BASE_URL = "https://planka.example.com" def _user() -> dict: return {"id": "u1", "username": "alexei", "name": "Alexei"} def test_card_created() -> None: payload = { "user": _user(), "item": { "id": "c1", "name": "Implement metrics", "description": "Wire prometheus client.", "boardId": "b1", "listId": "l1", "position": 1, }, "included": { "board": {"id": "b1", "name": "Roadmap"}, "lists": [{"id": "l1", "name": "Todo"}], }, } evt = parse_webhook("cardCreate", payload, provider_name="planka", base_url=_BASE_URL) assert evt is not None assert evt.event_type is EventType.CARD_CREATED assert evt.provider_type is ServiceProviderType.PLANKA assert evt.collection_id == "b1" assert evt.collection_name == "Roadmap" assert evt.extra["card_name"] == "Implement metrics" assert evt.extra["card_url"] == f"{_BASE_URL}/cards/c1" assert evt.extra["list_name"] == "Todo" assert evt.extra["sender"] == "alexei" def test_card_moved_when_list_changes() -> None: """beforeUpdate.listId != item.listId is the signal Planka uses for a card move; the parser must promote the generic cardUpdate event into CARD_MOVED so trackers can subscribe to moves specifically.""" payload = { "user": _user(), "beforeUpdate": {"listId": "l1"}, "item": { "id": "c1", "name": "Implement metrics", "description": "", "boardId": "b1", "listId": "l2", }, "included": { "board": {"id": "b1", "name": "Roadmap"}, "lists": [ {"id": "l1", "name": "Todo"}, {"id": "l2", "name": "In progress"}, ], }, } evt = parse_webhook("cardUpdate", payload, provider_name="planka", base_url=_BASE_URL) assert evt is not None assert evt.event_type is EventType.CARD_MOVED assert evt.extra["old_list_id"] == "l1" assert evt.extra["new_list_id"] == "l2" assert evt.extra["old_list_name"] == "Todo" assert evt.extra["new_list_name"] == "In progress" def test_card_update_without_list_change_is_card_updated() -> None: payload = { "user": _user(), "beforeUpdate": {"name": "Old name"}, "item": { "id": "c1", "name": "New name", "description": "", "boardId": "b1", "listId": "l1", }, } evt = parse_webhook("cardUpdate", payload, provider_name="planka", base_url=_BASE_URL) assert evt is not None assert evt.event_type is EventType.CARD_UPDATED def test_comment_created() -> None: payload = { "user": _user(), "item": { "id": "cm1", "text": "LGTM, ship it.", "cardId": "c1", "userId": "u1", }, "included": { "card": {"id": "c1", "name": "Implement metrics", "boardId": "b1"}, "board": {"id": "b1", "name": "Roadmap"}, }, } evt = parse_webhook( "commentCreate", payload, provider_name="planka", base_url=_BASE_URL, ) assert evt is not None assert evt.event_type is EventType.CARD_COMMENTED assert evt.collection_id == "b1" assert evt.extra["comment_text"] == "LGTM, ship it." assert evt.extra["card_id"] == "c1" assert evt.extra["card_url"] == f"{_BASE_URL}/cards/c1" def test_task_completion_emits_only_on_transition() -> None: """Task updates should only produce TASK_COMPLETED when the task flips from incomplete to complete — toggling the description or other fields on a task that was already complete must NOT spam notifications.""" completing = { "user": _user(), "beforeUpdate": {"isCompleted": False}, "item": {"id": "t1", "name": "Step 1", "isCompleted": True, "cardId": "c1"}, "included": { "card": {"id": "c1", "name": "Implement metrics", "boardId": "b1"}, "board": {"id": "b1", "name": "Roadmap"}, }, } evt = parse_webhook("taskUpdate", completing, provider_name="planka", base_url=_BASE_URL) assert evt is not None assert evt.event_type is EventType.TASK_COMPLETED # Editing a task that was already completed -> no event. re_edit = { "user": _user(), "beforeUpdate": {"isCompleted": True}, "item": {"id": "t1", "name": "Step 1 v2", "isCompleted": True, "cardId": "c1"}, } assert parse_webhook("taskUpdate", re_edit, provider_name="planka", base_url=_BASE_URL) is None def test_unknown_event_returns_none() -> None: assert parse_webhook("nonexistent", {"item": {}}, provider_name="planka", base_url="") is None