From 5b7d9ddbde58fe256e9c3d7506d013853cf3d5ea Mon Sep 17 00:00:00 2001 From: arabcoders Date: Wed, 21 Jan 2026 18:29:16 +0300 Subject: [PATCH] refactor: migrate notifications to the new db model --- app/features/core/schemas.py | 1 + app/features/notifications/deps.py | 5 + app/features/notifications/migration.py | 168 ++++ app/features/notifications/models.py | 39 + app/features/notifications/repository.py | 162 +++ app/features/notifications/router.py | 218 +++++ app/features/notifications/schemas.py | 145 +++ app/features/notifications/service.py | 347 +++++++ .../tests/test_notifications_repository.py | 154 +++ app/library/Notifications.py | 559 ----------- app/library/router.py | 28 +- app/main.py | 4 +- .../20260121134838_add_notifications_table.py | 40 + app/routes/api/notifications.py | 109 +-- app/tests/test_notifications.py | 921 ------------------ ui/app/components/ConditionForm.vue | 4 +- ui/app/components/DLFieldForm.vue | 58 +- ui/app/components/NotificationForm.vue | 27 +- ui/app/composables/useNotifications.ts | 427 ++++++++ ui/app/pages/conditions.vue | 2 +- ui/app/pages/notifications.vue | 188 +--- ui/app/stores/NotificationStore.ts | 1 - ui/app/types/notification.d.ts | 19 +- 23 files changed, 1846 insertions(+), 1780 deletions(-) create mode 100644 app/features/notifications/deps.py create mode 100644 app/features/notifications/migration.py create mode 100644 app/features/notifications/models.py create mode 100644 app/features/notifications/repository.py create mode 100644 app/features/notifications/router.py create mode 100644 app/features/notifications/schemas.py create mode 100644 app/features/notifications/service.py create mode 100644 app/features/notifications/tests/test_notifications_repository.py delete mode 100644 app/library/Notifications.py create mode 100644 app/migrations/20260121134838_add_notifications_table.py delete mode 100644 app/tests/test_notifications.py create mode 100644 ui/app/composables/useNotifications.ts diff --git a/app/features/core/schemas.py b/app/features/core/schemas.py index 9fa1235d..ce2e8378 100644 --- a/app/features/core/schemas.py +++ b/app/features/core/schemas.py @@ -19,6 +19,7 @@ class CEFeature(str, Enum): PRESETS = "presets" DL_FIELDS = "dl_fields" CONDITIONS = "conditions" + NOTIFICATIONS = "notifications" def __str__(self) -> str: return self.value diff --git a/app/features/notifications/deps.py b/app/features/notifications/deps.py new file mode 100644 index 00000000..ae270442 --- /dev/null +++ b/app/features/notifications/deps.py @@ -0,0 +1,5 @@ +from app.features.notifications.repository import NotificationsRepository + + +def get_notifications_repo() -> NotificationsRepository: + return NotificationsRepository.get_instance() diff --git a/app/features/notifications/migration.py b/app/features/notifications/migration.py new file mode 100644 index 00000000..63e934cf --- /dev/null +++ b/app/features/notifications/migration.py @@ -0,0 +1,168 @@ +from __future__ import annotations + +import json +import logging +from pathlib import Path +from typing import TYPE_CHECKING, Any + +from app.features.core.migration import Migration as FeatureMigration +from app.features.notifications.schemas import NotificationEvents +from app.library.config import Config +from app.library.Presets import Presets + +if TYPE_CHECKING: + from app.features.notifications.repository import NotificationsRepository + +LOG: logging.Logger = logging.getLogger(__name__) + + +class Migration(FeatureMigration): + name: str = "notifications" + + def __init__(self, repo: NotificationsRepository, config: Config | None = None): + self._config: Config = config or Config.get_instance() + super().__init__(config=self._config) + self._repo: NotificationsRepository = repo + self._source_file: Path = Path(self._config.config_path) / "notifications.json" + + async def should_run(self) -> bool: + return self._source_file.exists() + + async def migrate(self) -> None: + if await self._repo.count() > 0: + LOG.warning("Notification targets already exist in the database; skipping migration.") + await self._move_file(self._source_file) + return + + try: + items: list[dict] | None = json.loads(self._source_file.read_text()) + except Exception as exc: + LOG.exception("Failed to read %s: %s. Ignoring", self._source_file, exc) + await self._move_file(self._source_file) + return + + if items is None: + LOG.warning("No notification targets found in %s; skipping migration.", self._source_file) + await self._move_file(self._source_file) + return + + inserted = 0 + for index, item in enumerate(items): + normalized = self._normalize(item, index) + if not normalized: + continue + try: + await self._repo.create(normalized) + inserted += 1 + except Exception as exc: + LOG.exception("Failed to insert notification '%s': %s", normalized.get("name"), exc) + + LOG.info("Migrated %s notification target(s) from %s.", inserted, self._source_file) + await self._move_file(self._source_file) + + def _normalize(self, item: Any, index: int) -> dict[str, Any] | None: + if not isinstance(item, dict): + LOG.warning("Skipping notification at index %s due to invalid type.", index) + return None + + name: str | None = item.get("name") + if not name or not isinstance(name, str): + LOG.warning("Skipping notification at index %s due to missing name.", index) + return None + + request = item.get("request") if isinstance(item.get("request"), dict) else {} + url: str | None = request.get("url") + if not url or not isinstance(url, str): + LOG.warning("Skipping notification '%s' due to missing request url.", name) + return None + + method = request.get("method") if isinstance(request.get("method"), str) else "POST" + method = method.upper() + if method not in {"POST", "PUT"}: + LOG.warning("Skipping notification '%s' due to invalid method '%s'.", name, method) + return None + + req_type = request.get("type") if isinstance(request.get("type"), str) else "json" + req_type = req_type.lower() + if req_type not in {"json", "form"}: + LOG.warning("Skipping notification '%s' due to invalid type '%s'.", name, req_type) + return None + + data_key = request.get("data_key") if isinstance(request.get("data_key"), str) else "data" + if not data_key: + data_key = "data" + + headers = self._normalize_headers(request.get("headers")) + + enabled: bool = item.get("enabled") if isinstance(item.get("enabled"), bool) else True + events = self._normalize_events(item.get("on")) + if events is None: + LOG.warning("Skipping notification '%s' due to invalid events.", name) + return None + + presets = self._normalize_presets(item.get("presets")) + if presets is None: + LOG.warning("Skipping notification '%s' due to invalid presets.", name) + return None + + return { + "name": name, + "on": events, + "presets": presets, + "enabled": enabled, + "request_url": url, + "request_method": method, + "request_type": req_type, + "request_data_key": data_key, + "request_headers": headers, + } + + def _normalize_events(self, events: Any) -> list[str] | None: + if events is None: + return [] + if not isinstance(events, list): + return [] + + allowed = set(NotificationEvents.get_events().values()) + valid = [event for event in events if event in allowed] + invalid = [event for event in events if event not in allowed] + + if len(invalid) > 0 and len(valid) < 1: + return None + + if len(invalid) > 0: + LOG.warning("Dropping invalid notification events: %s", ", ".join(invalid)) + + return valid + + def _normalize_presets(self, presets: Any) -> list[str] | None: + if presets is None: + return [] + if not isinstance(presets, list): + return [] + + allowed = {preset.name for preset in Presets.get_instance().get_all()} + valid = [preset for preset in presets if preset in allowed] + invalid = [preset for preset in presets if preset not in allowed] + + if len(invalid) > 0 and len(valid) < 1: + return None + + if len(invalid) > 0: + LOG.warning("Dropping invalid notification presets: %s", ", ".join(invalid)) + + return valid + + def _normalize_headers(self, headers: Any) -> list[dict[str, str]]: + if not isinstance(headers, list): + return [] + + normalized: list[dict[str, str]] = [] + for header in headers: + if not isinstance(header, dict): + continue + key = str(header.get("key", "")).strip() + value = str(header.get("value", "")).strip() + if key and value: + normalized.append({"key": key, "value": value}) + return normalized diff --git a/app/features/notifications/models.py b/app/features/notifications/models.py new file mode 100644 index 00000000..38c698a5 --- /dev/null +++ b/app/features/notifications/models.py @@ -0,0 +1,39 @@ +from __future__ import annotations + +from datetime import datetime # noqa: TC003 + +from sqlalchemy import JSON, Boolean, Index, Integer, String, Text +from sqlalchemy import Enum as SQLAEnum +from sqlalchemy.orm import Mapped, mapped_column + +from app.features.core.models import Base, UTCDateTime, utcnow +from app.features.notifications.schemas import NotificationRequestMethod, NotificationRequestType + + +class NotificationModel(Base): + __tablename__: str = "notifications" + __table_args__: tuple[Index, ...] = ( + Index("ix_notifications_name", "name"), + Index("ix_notifications_enabled", "enabled"), + ) + + id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True) + name: Mapped[str] = mapped_column(String(255), nullable=False, unique=True) + on: Mapped[list[str]] = mapped_column(JSON, nullable=False, default=list) + presets: Mapped[list[str]] = mapped_column(JSON, nullable=False, default=list) + enabled: Mapped[bool] = mapped_column(Boolean, nullable=False, default=True) + request_url: Mapped[str] = mapped_column(Text, nullable=False) + request_method: Mapped[NotificationRequestMethod] = mapped_column( + SQLAEnum(NotificationRequestMethod, name="notification_request_method", native_enum=False), + nullable=False, + default=NotificationRequestMethod.POST, + ) + request_type: Mapped[NotificationRequestType] = mapped_column( + SQLAEnum(NotificationRequestType, name="notification_request_type", native_enum=False), + nullable=False, + default=NotificationRequestType.JSON, + ) + request_data_key: Mapped[str] = mapped_column(String(255), nullable=False, default="data") + request_headers: Mapped[list[dict]] = mapped_column(JSON, nullable=False, default=list) + created_at: Mapped[datetime] = mapped_column(UTCDateTime, default=utcnow, nullable=False) + updated_at: Mapped[datetime] = mapped_column(UTCDateTime, default=utcnow, onupdate=utcnow, nullable=False) diff --git a/app/features/notifications/repository.py b/app/features/notifications/repository.py new file mode 100644 index 00000000..073fd355 --- /dev/null +++ b/app/features/notifications/repository.py @@ -0,0 +1,162 @@ +from __future__ import annotations + +import logging +from typing import TYPE_CHECKING, Any + +from sqlalchemy import func, or_, select + +from app.features.core.deps import get_session +from app.features.notifications.migration import Migration +from app.features.notifications.models import NotificationModel +from app.library.Singleton import Singleton + +if TYPE_CHECKING: + from collections.abc import AsyncGenerator + + from sqlalchemy.engine.result import Result + from sqlalchemy.ext.asyncio import AsyncSession + from sqlalchemy.sql.elements import ColumnElement + from sqlalchemy.sql.selectable import Select + +LOG: logging.Logger = logging.getLogger(__name__) + + +class NotificationsRepository(metaclass=Singleton): + def __init__(self, session: AsyncGenerator[AsyncSession] | None = None) -> None: + self._migrated = False + self.session: AsyncGenerator[AsyncSession] = session or get_session + + async def run_migrations(self) -> None: + if self._migrated: + return + + self._migrated = True + await Migration(repo=self, config=None).run() + + @staticmethod + def get_instance() -> NotificationsRepository: + return NotificationsRepository() + + async def list(self) -> list[NotificationModel]: + async with self.session() as session: + result: Result[tuple[NotificationModel]] = await session.execute( + select(NotificationModel).order_by(NotificationModel.name.asc()) + ) + return list(result.scalars().all()) + + async def list_paginated(self, page: int, per_page: int) -> tuple[list[NotificationModel], int, int, int]: + async with self.session() as session: + total: int = await self.count() + total_pages: int = (total + per_page - 1) // per_page if total > 0 else 1 + + if page > total_pages and total > 0: + page = total_pages + + query: Select[tuple[NotificationModel]] = ( + select(NotificationModel) + .order_by(NotificationModel.name.asc()) + .limit(per_page) + .offset((page - 1) * per_page) + ) + result: Result[tuple[NotificationModel]] = await session.execute(query) + return list(result.scalars().all()), total, page, total_pages + + async def count(self) -> int: + async with self.session() as session: + result: Result[tuple[int]] = await session.execute(select(func.count()).select_from(NotificationModel)) + return int(result.scalar_one()) + + async def get(self, identifier: int | str) -> NotificationModel | None: + async with self.session() as session: + if not identifier: + return None + + if isinstance(identifier, int): + clause: ColumnElement[bool] = NotificationModel.id == identifier + elif isinstance(identifier, str) and identifier.isdigit(): + clause = or_(NotificationModel.id == int(identifier), NotificationModel.name == identifier) + else: + clause = NotificationModel.name == identifier + + result: Result[tuple[NotificationModel]] = await session.execute( + select(NotificationModel).where(clause).limit(1) + ) + return result.scalar_one_or_none() + + async def get_by_name(self, name: str, exclude_id: int | None = None) -> NotificationModel | None: + async with self.session() as session: + query: Select[tuple[NotificationModel]] = select(NotificationModel).where(NotificationModel.name == name) + if exclude_id is not None: + query = query.where(NotificationModel.id != exclude_id) + + result: Result[tuple[NotificationModel]] = await session.execute(query.limit(1)) + return result.scalar_one_or_none() + + async def create(self, payload: NotificationModel | dict) -> NotificationModel: + async with self.session() as session: + model: NotificationModel = NotificationModel(**payload) if isinstance(payload, dict) else payload + if model.id is not None: + model.id = None + + if await self.get_by_name(name=model.name) is not None: + msg: str = f"Notification target with name '{model.name}' already exists." + raise ValueError(msg) + + session.add(model) + await session.commit() + await session.refresh(model) + return model + + async def update(self, identifier: int | str, payload: dict[str, Any]) -> NotificationModel: + async with self.session() as session: + if isinstance(identifier, int): + clause: ColumnElement[bool] = NotificationModel.id == identifier + elif isinstance(identifier, str) and identifier.isdigit(): + clause = or_(NotificationModel.id == int(identifier), NotificationModel.name == identifier) + else: + clause = NotificationModel.name == identifier + + result: Result[tuple[NotificationModel]] = await session.execute( + select(NotificationModel).where(clause).limit(1) + ) + model: NotificationModel | None = result.scalar_one_or_none() + + if not model: + msg: str = f"Notification target '{identifier}' not found." + raise KeyError(msg) + + payload.pop("id", None) + + if "name" in payload and await self.get_by_name(name=payload["name"], exclude_id=model.id) is not None: + msg = f"Notification target with name '{payload['name']}' already exists." + raise ValueError(msg) + + for key, value in payload.items(): + if hasattr(model, key): + setattr(model, key, value) + + await session.commit() + await session.refresh(model) + return model + + async def delete(self, identifier: int | str) -> NotificationModel: + async with self.session() as session: + if isinstance(identifier, int): + clause: ColumnElement[bool] = NotificationModel.id == identifier + elif isinstance(identifier, str) and identifier.isdigit(): + clause = or_(NotificationModel.id == int(identifier), NotificationModel.name == identifier) + else: + clause = NotificationModel.name == identifier + + result: Result[tuple[NotificationModel]] = await session.execute( + select(NotificationModel).where(clause).limit(1) + ) + model = result.scalar_one_or_none() + + if not model: + msg: str = f"Notification target '{identifier}' not found." + raise KeyError(msg) + + await session.delete(model) + await session.commit() + return model diff --git a/app/features/notifications/router.py b/app/features/notifications/router.py new file mode 100644 index 00000000..b357eb7e --- /dev/null +++ b/app/features/notifications/router.py @@ -0,0 +1,218 @@ +import logging +from typing import Any + +from aiohttp import web +from aiohttp.web import Request, Response +from pydantic import ValidationError + +from app.features.core.schemas import CEAction, CEFeature, ConfigEvent, Pagination +from app.features.core.utils import build_pagination, format_validation_errors, normalize_pagination +from app.features.notifications.schemas import Notification, NotificationEvents, NotificationList, NotificationPatch +from app.features.notifications.service import Notifications +from app.library.encoder import Encoder +from app.library.Events import EventBus, Events +from app.library.router import route + +LOG: logging.Logger = logging.getLogger(__name__) + + +def _model(model: Any) -> Notification: + return Notifications.get_instance().model_to_schema(model) + + +def _serialize(model: Any) -> dict: + return _model(model).model_dump() + + +@route("GET", "api/notifications/", "notifications_list") +async def notifications_list(request: Request, encoder: Encoder) -> Response: + page, per_page = normalize_pagination(request) + items, total, current_page, total_pages = await Notifications.get_instance().list_paginated(page, per_page) + return web.json_response( + data=NotificationList( + items=[_model(model) for model in items], + pagination=Pagination.model_validate(build_pagination(total, current_page, per_page, total_pages)), + ), + status=web.HTTPOk.status_code, + dumps=encoder.encode, + ) + + +@route("GET", "api/notifications/events/", "notifications_events") +async def notifications_events(encoder: Encoder) -> Response: + return web.json_response( + data={"events": list(NotificationEvents.get_events().values())}, + status=web.HTTPOk.status_code, + dumps=encoder.encode, + ) + + +@route("POST", "api/notifications/test/", "notification_test") +async def notification_test(encoder: Encoder, notify: EventBus) -> Response: + notify.emit(Events.TEST, title="Test Notification", message="This is a test notification.") + return web.json_response(data={}, status=web.HTTPOk.status_code, dumps=encoder.encode) + + +@route("POST", "api/notifications/", "notification_add") +async def notifications_add(request: Request, encoder: Encoder, notify: EventBus) -> Response: + data = await request.json() + + if not isinstance(data, dict): + return web.json_response( + {"error": "Invalid request body expecting dict."}, + status=web.HTTPBadRequest.status_code, + ) + + try: + item: Notification = Notification.model_validate(data) + except ValidationError as exc: + return web.json_response( + data={"error": "Failed to validate notification.", "detail": format_validation_errors(exc)}, + status=web.HTTPBadRequest.status_code, + ) + + try: + saved = _serialize(await Notifications.get_instance().create(item)) + except ValueError as exc: + return web.json_response({"error": str(exc)}, status=web.HTTPBadRequest.status_code) + + notify.emit( + Events.CONFIG_UPDATE, + data=ConfigEvent(feature=CEFeature.NOTIFICATIONS, action=CEAction.CREATE, data=saved), + ) + return web.json_response(data=saved, status=web.HTTPOk.status_code, dumps=encoder.encode) + + +@route("GET", "api/notifications/{id}", "notification_get") +async def notifications_get(request: Request, encoder: Encoder) -> Response: + if not (identifier := request.match_info.get("id")): + return web.json_response({"error": "ID required"}, status=web.HTTPBadRequest.status_code) + + if not (model := await Notifications.get_instance().get(identifier)): + return web.json_response({"error": "Notification not found"}, status=web.HTTPNotFound.status_code) + + return web.json_response(data=_serialize(model), status=web.HTTPOk.status_code, dumps=encoder.encode) + + +@route("DELETE", "api/notifications/{id}", "notification_delete") +async def notifications_delete(request: Request, encoder: Encoder, notify: EventBus) -> Response: + if not (identifier := request.match_info.get("id")): + return web.json_response({"error": "ID required"}, status=web.HTTPBadRequest.status_code) + + try: + deleted = _serialize(await Notifications.get_instance().delete(identifier)) + notify.emit( + Events.CONFIG_UPDATE, + data=ConfigEvent(feature=CEFeature.NOTIFICATIONS, action=CEAction.DELETE, data=deleted), + ) + return web.json_response( + data=deleted, + status=web.HTTPOk.status_code, + dumps=encoder.encode, + ) + except KeyError as exc: + return web.json_response({"error": str(exc)}, status=web.HTTPNotFound.status_code) + + +@route("PATCH", "api/notifications/{id}", "notification_patch") +async def notifications_patch(request: Request, encoder: Encoder, notify: EventBus) -> Response: + if not (identifier := request.match_info.get("id")): + return web.json_response({"error": "ID required"}, status=web.HTTPBadRequest.status_code) + + if not (model := await Notifications.get_instance().get(identifier)): + return web.json_response({"error": "Notification not found"}, status=web.HTTPNotFound.status_code) + + data = await request.json() + + if not isinstance(data, dict): + return web.json_response( + {"error": "Invalid request body expecting dict."}, + status=web.HTTPBadRequest.status_code, + ) + + try: + validated = NotificationPatch.model_validate(data) + except ValidationError as exc: + return web.json_response( + data={"error": "Failed to validate notification.", "detail": format_validation_errors(exc)}, + status=web.HTTPBadRequest.status_code, + ) + + current = _model(model).model_dump() + if validated.name is not None: + current["name"] = validated.name + if validated.on is not None: + current["on"] = validated.on + if validated.presets is not None: + current["presets"] = validated.presets + if validated.enabled is not None: + current["enabled"] = validated.enabled + + if validated.request is not None: + request_payload = current.get("request", {}) + if validated.request.url is not None: + request_payload["url"] = validated.request.url + if validated.request.method is not None: + request_payload["method"] = validated.request.method + if validated.request.type is not None: + request_payload["type"] = validated.request.type + if validated.request.headers is not None: + request_payload["headers"] = [header.model_dump() for header in validated.request.headers] + if validated.request.data_key is not None: + request_payload["data_key"] = validated.request.data_key + current["request"] = request_payload + + try: + updated = _serialize(await Notifications.get_instance().update(model.id, current)) + except ValueError as exc: + return web.json_response({"error": str(exc)}, status=web.HTTPBadRequest.status_code) + except KeyError as exc: + return web.json_response({"error": str(exc)}, status=web.HTTPNotFound.status_code) + + notify.emit( + Events.CONFIG_UPDATE, + data=ConfigEvent(feature=CEFeature.NOTIFICATIONS, action=CEAction.UPDATE, data=updated), + ) + return web.json_response( + data=updated, + status=web.HTTPOk.status_code, + dumps=encoder.encode, + ) + + +@route("PUT", "api/notifications/{id}", "notification_update") +async def notifications_update(request: Request, encoder: Encoder, notify: EventBus) -> Response: + if not (identifier := request.match_info.get("id")): + return web.json_response({"error": "ID required"}, status=web.HTTPBadRequest.status_code) + + if not (model := await Notifications.get_instance().get(identifier)): + return web.json_response({"error": "Notification not found"}, status=web.HTTPNotFound.status_code) + + data = await request.json() + + if not isinstance(data, dict): + return web.json_response( + {"error": "Invalid request body expecting dict."}, + status=web.HTTPBadRequest.status_code, + ) + + try: + validated = Notification.model_validate(data) + except ValidationError as exc: + return web.json_response( + data={"error": "Failed to validate notification.", "detail": format_validation_errors(exc)}, + status=web.HTTPBadRequest.status_code, + ) + + try: + updated = _serialize(await Notifications.get_instance().update(model.id, validated)) + except ValueError as exc: + return web.json_response({"error": str(exc)}, status=web.HTTPBadRequest.status_code) + except KeyError as exc: + return web.json_response({"error": str(exc)}, status=web.HTTPNotFound.status_code) + + notify.emit( + Events.CONFIG_UPDATE, + data=ConfigEvent(feature=CEFeature.NOTIFICATIONS, action=CEAction.UPDATE, data=updated), + ) + return web.json_response(data=updated, status=web.HTTPOk.status_code, dumps=encoder.encode) diff --git a/app/features/notifications/schemas.py b/app/features/notifications/schemas.py new file mode 100644 index 00000000..d2eb024f --- /dev/null +++ b/app/features/notifications/schemas.py @@ -0,0 +1,145 @@ +from __future__ import annotations + +from enum import Enum +from typing import Any + +from pydantic import BaseModel, ConfigDict, Field, field_validator + +from app.features.core.schemas import Pagination +from app.library.Events import Events + + +class NotificationEvents: + TEST: str = Events.TEST + + ITEM_ADDED: str = Events.ITEM_ADDED + ITEM_COMPLETED: str = Events.ITEM_COMPLETED + ITEM_CANCELLED: str = Events.ITEM_CANCELLED + ITEM_DELETED: str = Events.ITEM_DELETED + ITEM_PAUSED: str = Events.ITEM_PAUSED + ITEM_RESUMED: str = Events.ITEM_RESUMED + ITEM_MOVED: str = Events.ITEM_MOVED + + PAUSED: str = Events.PAUSED + RESUMED: str = Events.RESUMED + + LOG_INFO: str = Events.LOG_INFO + LOG_SUCCESS: str = Events.LOG_SUCCESS + LOG_WARNING: str = Events.LOG_WARNING + LOG_ERROR: str = Events.LOG_ERROR + + TASK_DISPATCHED: str = Events.TASK_DISPATCHED + + @staticmethod + def get_events() -> dict[str, str]: + return {k: v for k, v in vars(NotificationEvents).items() if not k.startswith("__") and not callable(v)} + + @staticmethod + def events() -> list[str]: + return [ + getattr(NotificationEvents, ev) + for ev in dir(NotificationEvents) + if not ev.startswith("_") and not callable(getattr(NotificationEvents, ev)) + ] + + @staticmethod + def is_valid(event: str) -> bool: + return event in NotificationEvents.get_events().values() + + +class NotificationRequestType(str, Enum): + JSON = "json" + FORM = "form" + + def __str__(self) -> str: + return self.value + + +class NotificationRequestMethod(str, Enum): + POST = "POST" + PUT = "PUT" + + def __str__(self) -> str: + return self.value + + +class NotificationRequestHeader(BaseModel): + model_config = ConfigDict(str_strip_whitespace=True, from_attributes=True) + + key: str = Field(min_length=1) + value: str = Field(min_length=1) + + +class NotificationRequest(BaseModel): + model_config = ConfigDict(str_strip_whitespace=True, from_attributes=True) + + type: NotificationRequestType = NotificationRequestType.JSON + method: NotificationRequestMethod = NotificationRequestMethod.POST + url: str = Field(min_length=1) + headers: list[NotificationRequestHeader] = Field(default_factory=list) + data_key: str = Field(default="data", min_length=1) + + @field_validator("method", mode="before") + @classmethod + def _normalize_method(cls, value: Any) -> Any: + if value is None: + return value + if isinstance(value, NotificationRequestMethod): + return value + return str(value).upper() + + @field_validator("type", mode="before") + @classmethod + def _normalize_type(cls, value: Any) -> Any: + if value is None: + return value + if isinstance(value, NotificationRequestType): + return value + return str(value).lower() + + +class NotificationRequestPatch(BaseModel): + model_config = ConfigDict(str_strip_whitespace=True, from_attributes=True) + + type: NotificationRequestType | None = None + method: NotificationRequestMethod | None = None + url: str | None = None + headers: list[NotificationRequestHeader] | None = None + data_key: str | None = None + + +class Notification(BaseModel): + model_config = ConfigDict(str_strip_whitespace=True, from_attributes=True) + + id: int | None = None + name: str = Field(min_length=1) + on: list[str] = Field(default_factory=list) + presets: list[str] = Field(default_factory=list) + enabled: bool = True + request: NotificationRequest + + @field_validator("on", "presets", mode="before") + @classmethod + def _normalize_list(cls, value: Any) -> list[str]: + if value is None: + return [] + if isinstance(value, list): + return [str(item).strip() for item in value if str(item).strip()] + return [str(value).strip()] + + +class NotificationPatch(BaseModel): + model_config = ConfigDict(str_strip_whitespace=True, from_attributes=True) + + name: str | None = None + on: list[str] | None = None + presets: list[str] | None = None + enabled: bool | None = None + request: NotificationRequestPatch | None = None + + +class NotificationList(BaseModel): + model_config = ConfigDict(from_attributes=True) + + items: list[Notification] = Field(default_factory=list) + pagination: Pagination diff --git a/app/features/notifications/service.py b/app/features/notifications/service.py new file mode 100644 index 00000000..f01948b3 --- /dev/null +++ b/app/features/notifications/service.py @@ -0,0 +1,347 @@ +from __future__ import annotations + +import asyncio +import logging +import traceback +from pathlib import Path +from typing import TYPE_CHECKING, Any + +from app.features.notifications.models import NotificationModel +from app.features.notifications.repository import NotificationsRepository +from app.features.notifications.schemas import ( + Notification, + NotificationEvents, + NotificationRequest, + NotificationRequestHeader, + NotificationRequestType, +) +from app.library.BackgroundWorker import BackgroundWorker +from app.library.config import Config +from app.library.encoder import Encoder +from app.library.Events import Event, EventBus, Events +from app.library.httpx_client import async_client +from app.library.ItemDTO import Item, ItemDTO +from app.library.Presets import Preset, Presets +from app.library.Singleton import Singleton + +if TYPE_CHECKING: + from collections.abc import Awaitable + + import httpx + from aiohttp import web + +LOG: logging.Logger = logging.getLogger("feature.notifications") + + +class Notifications(metaclass=Singleton): + def __init__( + self, + repo: NotificationsRepository | None = None, + client: httpx.AsyncClient | None = None, + encoder: Encoder | None = None, + config: Config | None = None, + background_worker: BackgroundWorker | None = None, + ) -> None: + self._repo: NotificationsRepository = repo or NotificationsRepository.get_instance() + + config = config or Config.get_instance() + + self._debug: bool = config.debug + self._version: str = config.app_version + self._client: httpx.AsyncClient = client or async_client() + self._encoder: Encoder = encoder or Encoder() + self._offload: BackgroundWorker = background_worker or BackgroundWorker.get_instance() + + @staticmethod + def get_instance() -> Notifications: + return Notifications() + + async def on_shutdown(self, _: web.Application) -> None: + pass + + def attach(self, _: web.Application) -> None: + async def handle_event(_, __): + await self._repo.run_migrations() + + EventBus.get_instance().subscribe(Events.STARTED, handle_event, "NotificationsRepository.run_migrations") + EventBus.get_instance().subscribe(NotificationEvents.events(), self.emit, f"{__class__.__name__}.emit") + + async def list(self) -> list[NotificationModel]: + return await self._repo.list() + + async def list_paginated(self, page: int, per_page: int) -> tuple[list[NotificationModel], int, int, int]: + return await self._repo.list_paginated(page, per_page) + + async def get(self, identifier: int | str) -> NotificationModel | None: + return await self._repo.get(identifier) + + async def create(self, item: Notification | dict) -> NotificationModel: + if not isinstance(item, Notification): + item = Notification.model_validate(item) + normalized = self._normalize(item) + payload = self._payload_from_schema(normalized) + return await self._repo.create(payload) + + async def update(self, identifier: int | str, payload: Notification | dict) -> NotificationModel: + if not isinstance(payload, Notification): + payload = Notification.model_validate(payload) + normalized = self._normalize(payload) + update_payload = self._payload_from_schema(normalized) + return await self._repo.update(identifier, update_payload) + + async def delete(self, identifier: int | str) -> NotificationModel: + return await self._repo.delete(identifier) + + async def send(self, ev: Event, wait: bool = True) -> list[dict] | list[Awaitable[dict]]: + targets = await self._repo.list() + if len(targets) < 1: + return [] + + if not isinstance(ev.data, (ItemDTO, Item, dict)): + LOG.debug("Received invalid item type '%s' with event '%s'.", type(ev.data), ev.event) + return [] + + tasks: list[Awaitable[dict]] = [] + apprise_targets: list[Notification] = [] + + for target_model in targets: + target = self.model_to_schema(target_model) + if not target.enabled: + continue + + if len(target.on) > 0 and ev.event not in target.on and NotificationEvents.TEST != ev.event: + continue + + if NotificationEvents.TEST != ev.event and not self._check_preset(target, ev): + continue + + if not target.request.url.startswith("http"): + apprise_targets.append(target) + else: + tasks.append(self._send(target, ev)) + + if len(apprise_targets) > 0: + tasks.append(self._apprise(apprise_targets, ev)) + + if wait: + return await asyncio.gather(*tasks) + + return tasks + + def emit(self, e: Event, _, **__) -> None: + if not NotificationEvents.is_valid(e.event): + return + + self._offload.submit(self.send, e) + + def _normalize(self, item: Notification) -> Notification: + if item.enabled is not None and not isinstance(item.enabled, bool): + msg: str = "Enabled must be a boolean." + raise ValueError(msg) + + item.on = self._filter_events(item.on) + item.presets = self._filter_presets(item.presets) + item.request.headers = [ + NotificationRequestHeader(key=header.key, value=header.value) + for header in item.request.headers + if header.key and header.value + ] + return item + + def _filter_events(self, events: list[str]) -> list[str]: + if not events: + return [] + + allowed = set(NotificationEvents.get_events().values()) + valid = [event for event in events if event in allowed] + invalid = [event for event in events if event not in allowed] + + if len(invalid) > 0 and len(valid) < 1: + msg: str = f"Invalid notification target. Invalid events '{', '.join(invalid)}' found." + raise ValueError(msg) + + if len(invalid) > 0: + LOG.warning("Dropping invalid notification events: %s", ", ".join(invalid)) + + return valid + + def _filter_presets(self, presets: list[str]) -> list[str]: + if not presets: + return [] + + all_presets: list[Preset] = Presets.get_instance().get_all() + allowed = {preset.name for preset in all_presets} + valid = [preset for preset in presets if preset in allowed] + invalid = [preset for preset in presets if preset not in allowed] + + if len(invalid) > 0 and len(valid) < 1: + msg: str = f"Invalid notification target. Invalid presets '{', '.join(invalid)}' found." + raise ValueError(msg) + + if len(invalid) > 0: + LOG.warning("Dropping invalid notification presets: %s", ", ".join(invalid)) + + return valid + + def model_to_schema(self, model: NotificationModel) -> Notification: + headers: list[NotificationRequestHeader] = [] + if isinstance(model.request_headers, list): + for header in model.request_headers: + if not isinstance(header, dict): + continue + key = str(header.get("key", "")).strip() + value = str(header.get("value", "")).strip() + if key and value: + headers.append(NotificationRequestHeader(key=key, value=value)) + + return Notification( + id=model.id, + name=model.name, + on=list(model.on or []), + presets=list(model.presets or []), + enabled=model.enabled, + request=NotificationRequest.model_validate( + { + "type": model.request_type, + "method": model.request_method, + "url": model.request_url, + "headers": headers, + "data_key": model.request_data_key, + } + ), + ) + + def _payload_from_schema(self, item: Notification) -> dict[str, Any]: + return { + "name": item.name, + "on": list(item.on), + "presets": list(item.presets), + "enabled": item.enabled, + "request_url": item.request.url, + "request_method": str(item.request.method), + "request_type": str(item.request.type), + "request_data_key": item.request.data_key, + "request_headers": [header.model_dump() for header in item.request.headers], + } + + def _check_preset(self, target: Notification, ev: Event) -> bool: + if len(target.presets) < 1: + return True + + if not isinstance(ev.data, (Item, ItemDTO, dict)): + return False + + preset_name: str | None = None + + if isinstance(ev.data, Item): + preset_name = ev.data.preset + + if isinstance(ev.data, ItemDTO): + preset_name = ev.data.preset + + if isinstance(ev.data, dict): + preset_name = ev.data.get("preset", None) + + if not preset_name: + return False + + return preset_name in target.presets + + async def _apprise(self, targets: list[Notification], ev: Event) -> dict: + if not targets: + return {} + + import apprise + + try: + notify = apprise.Apprise(debug=self._debug) + apr_config = Path(Config.get_instance().apprise_config) + if apr_config.exists(): + apprise_file = apprise.AppriseConfig() + apprise_file.add(str(apr_config)) + notify.add(apprise_file) + + for target in targets: + notify.add(target.request.url) + + status = await notify.async_notify( + body=ev.message or self._encoder.encode(ev.serialize()), + title=ev.title or f"YTPTube Event: {ev.event}", + ) + + if not status: + msg = "Apprise failed to send notification." + raise RuntimeError(msg) # noqa: TRY301 + except Exception as exc: + LOG.exception(exc) + LOG.error("Error sending Apprise notification: %s", exc) + return {"error": str(exc), "event": ev.event, "id": ev.id, "targets": [t.name for t in targets]} + + return {} + + async def _send(self, target: Notification, ev: Event) -> dict: + try: + LOG.info("Sending notification event '%s: %s' to '%s'.", ev.event, ev.id, target.name) + + headers: dict[str, str] = { + "User-Agent": f"YTPTube/{self._version}", + "X-Event-Id": ev.id, + "X-Event": ev.event, + "Content-Type": "application/json" + if NotificationRequestType.JSON == target.request.type + else "application/x-www-form-urlencoded", + } + + if len(target.request.headers) > 0: + headers.update({h.key: h.value for h in target.request.headers if h.key and h.value}) + + payload_data: dict[str, Any] = ev.serialize() + + if "data" != target.request.data_key: + payload_data[target.request.data_key] = payload_data["data"] + payload_data.pop("data", None) + + if NotificationRequestType.FORM == target.request.type: + payload_data[target.request.data_key] = self._encoder.encode(payload_data[target.request.data_key]) + data_payload: dict[str, Any] | None = payload_data + content_payload: str | None = None + else: + data_payload = None + content_payload = self._encoder.encode(payload_data) + + response = await self._client.request( + method=str(target.request.method).upper(), + url=target.request.url, + headers=headers, + data=data_payload, + content=content_payload, + ) + + resp_data: dict[str, Any] = { + "url": target.request.url, + "status": response.status_code, + "text": response.text, + } + + msg: str = ( + f"Notification target '{target.name}' Responded to event '{ev.event}: {ev.id}' " + f"with status '{response.status_code}'." + ) + if self._debug and resp_data.get("text"): + msg += f" body '{resp_data.get('text', '??')}'." + + LOG.info(msg) + + return resp_data + except Exception as exc: + err_msg = str(exc) or type(exc).__name__ + tb = "".join(traceback.format_exception(type(exc), exc, exc.__traceback__)) + LOG.error( + "Error sending notification event '%s: %s' to '%s'. '%s'. %s", + ev.event, + ev.id, + target.name, + err_msg, + tb, + ) + return {"url": target.request.url, "status": 500, "text": str(ev)} diff --git a/app/features/notifications/tests/test_notifications_repository.py b/app/features/notifications/tests/test_notifications_repository.py new file mode 100644 index 00000000..51ef1260 --- /dev/null +++ b/app/features/notifications/tests/test_notifications_repository.py @@ -0,0 +1,154 @@ +"""Tests for NotificationsRepository.""" + +from __future__ import annotations + +import pytest +import pytest_asyncio + +from app.features.notifications.repository import NotificationsRepository +from app.library.sqlite_store import SqliteStore + + +@pytest_asyncio.fixture +async def repo(tmp_path): + """Provide a fresh repository instance with initialized database for each test.""" + NotificationsRepository._reset_singleton() + SqliteStore._reset_singleton() + + store = SqliteStore(db_path=":memory:") + await store.get_connection() + + repository = NotificationsRepository.get_instance() + + yield repository + + if store._conn: + await store._conn.close() + if store._engine: + await store._engine.dispose() + + NotificationsRepository._reset_singleton() + SqliteStore._reset_singleton() + + +class TestNotificationsRepository: + """Test suite for NotificationsRepository database operations.""" + + @pytest.mark.asyncio + async def test_list_empty(self, repo): + """List returns empty when no notifications exist.""" + notifications = await repo.list() + assert notifications == [], "Should return empty list when no notifications exist" + + @pytest.mark.asyncio + async def test_create_notification(self, repo): + """Create notification with valid data.""" + payload = { + "name": "Webhook", + "on": ["item_completed"], + "presets": [], + "enabled": True, + "request_url": "https://example.com/webhook", + "request_method": "POST", + "request_type": "json", + "request_data_key": "data", + "request_headers": [{"key": "Authorization", "value": "Bearer token"}], + } + + model = await repo.create(payload) + + assert model.id is not None, "Should generate ID for new notification" + assert model.name == "Webhook", "Should store name correctly" + assert model.request_url == "https://example.com/webhook", "Should store request url" + assert model.request_method == "POST", "Should store request method" + + @pytest.mark.asyncio + async def test_get_by_id(self, repo): + """Get notification by integer ID.""" + created = await repo.create( + { + "name": "Get Test", + "on": [], + "presets": [], + "enabled": True, + "request_url": "https://example.com", + "request_method": "POST", + "request_type": "json", + "request_data_key": "data", + "request_headers": [], + } + ) + + retrieved = await repo.get(created.id) + + assert retrieved is not None, "Should retrieve created notification" + assert retrieved.id == created.id, "Should match created notification id" + + @pytest.mark.asyncio + async def test_update_notification(self, repo): + """Update existing notification.""" + created = await repo.create( + { + "name": "Update Test", + "on": [], + "presets": [], + "enabled": True, + "request_url": "https://example.com", + "request_method": "POST", + "request_type": "json", + "request_data_key": "data", + "request_headers": [], + } + ) + + updated = await repo.update(created.id, {"name": "Updated Name", "enabled": False}) + + assert updated.name == "Updated Name", "Should update name" + assert updated.enabled is False, "Should update enabled flag" + + @pytest.mark.asyncio + async def test_delete_notification(self, repo): + """Delete existing notification.""" + created = await repo.create( + { + "name": "Delete Test", + "on": [], + "presets": [], + "enabled": True, + "request_url": "https://example.com", + "request_method": "POST", + "request_type": "json", + "request_data_key": "data", + "request_headers": [], + } + ) + + deleted = await repo.delete(created.id) + + assert deleted.id == created.id, "Should return deleted notification" + assert await repo.get(created.id) is None, "Deleted notification should not be retrievable" + + @pytest.mark.asyncio + async def test_list_paginated(self, repo): + """List paginated returns correct subset.""" + for i in range(4): + await repo.create( + { + "name": f"Item {i}", + "on": [], + "presets": [], + "enabled": True, + "request_url": "https://example.com", + "request_method": "POST", + "request_type": "json", + "request_data_key": "data", + "request_headers": [], + } + ) + + items, total, page, total_pages = await repo.list_paginated(page=1, per_page=2) + + assert len(items) == 2, "Should return 2 items per page" + assert total == 4, "Should report total count of 4" + assert page == 1, "Should be on page 1" + assert total_pages == 2, "Should have 2 pages total" diff --git a/app/library/Notifications.py b/app/library/Notifications.py deleted file mode 100644 index e10e5145..00000000 --- a/app/library/Notifications.py +++ /dev/null @@ -1,559 +0,0 @@ -import asyncio -import json -import logging -import traceback -from collections.abc import Awaitable -from dataclasses import dataclass, field -from pathlib import Path -from typing import Any - -import httpx -from aiohttp import web - -from .ag_utils import ag -from .BackgroundWorker import BackgroundWorker -from .config import Config -from .encoder import Encoder -from .Events import Event, EventBus, Events -from .httpx_client import async_client -from .ItemDTO import Item, ItemDTO -from .Presets import Preset, Presets -from .Singleton import Singleton -from .Utils import validate_uuid - -LOG = logging.getLogger("notifications") - - -@dataclass(kw_only=True) -class TargetRequestHeader: - """Request header details for a notification target.""" - - key: str - value: str - - def serialize(self) -> dict: - return {"key": self.key, "value": self.value} - - def json(self) -> str: - return Encoder().encode(self.serialize()) - - def get(self, key: str, default: Any = None) -> Any: - return self.serialize().get(key, default) - - -@dataclass(kw_only=True) -class TargetRequest: - """Request details for a notification target.""" - - type: str - method: str - url: str - headers: list[TargetRequestHeader] = field(default_factory=list) - data_key: str = "data" - - def serialize(self) -> dict: - return { - "type": self.type, - "method": self.method, - "url": self.url, - "data_key": self.data_key, - "headers": [h.serialize() for h in self.headers], - } - - def json(self) -> str: - return Encoder().encode(self.serialize()) - - def get(self, key: str, default: Any = None) -> Any: - return ag(self.serialize(), key, default) - - -@dataclass(kw_only=True) -class Target: - """Notification target details.""" - - id: str - name: str - on: list[str] = field(default_factory=list) - presets: list[str] = field(default_factory=list) - request: TargetRequest - enabled: bool = True - - def serialize(self) -> dict: - return { - "id": self.id, - "name": self.name, - "on": self.on, - "presets": self.presets, - "request": self.request.serialize(), - "enabled": self.enabled, - } - - def json(self) -> str: - return Encoder().encode(self.serialize()) - - def get(self, key: str, default: Any = None) -> Any: - return self.serialize().get(key, default) - - -class NotificationEvents: - TEST: str = Events.TEST - - ITEM_ADDED: str = Events.ITEM_ADDED - ITEM_COMPLETED: str = Events.ITEM_COMPLETED - ITEM_CANCELLED: str = Events.ITEM_CANCELLED - ITEM_DELETED: str = Events.ITEM_DELETED - ITEM_PAUSED: str = Events.ITEM_PAUSED - ITEM_RESUMED: str = Events.ITEM_RESUMED - ITEM_MOVED: str = Events.ITEM_MOVED - - PAUSED: str = Events.PAUSED - RESUMED: str = Events.RESUMED - - LOG_INFO: str = Events.LOG_INFO - LOG_SUCCESS: str = Events.LOG_SUCCESS - LOG_WARNING: str = Events.LOG_WARNING - LOG_ERROR: str = Events.LOG_ERROR - - TASK_DISPATCHED: str = Events.TASK_DISPATCHED - - @staticmethod - def get_events() -> dict[str, str]: - return {k: v for k, v in vars(NotificationEvents).items() if not k.startswith("__") and not callable(v)} - - def events() -> list: - return [ - getattr(NotificationEvents, ev) - for ev in dir(NotificationEvents) - if not ev.startswith("_") and not callable(getattr(NotificationEvents, ev)) - ] - - @staticmethod - def is_valid(event: str) -> bool: - return event in NotificationEvents.get_events().values() - - -class Notification(metaclass=Singleton): - def __init__( - self, - file: str | None = None, - client: httpx.AsyncClient | None = None, - encoder: Encoder | None = None, - config: Config | None = None, - background_worker: BackgroundWorker | None = None, - ): - self._targets: list[Target] = [] - "Notification targets to send events to." - - config: Config = config or Config.get_instance() - - self._debug: bool = config.debug - "Debug mode." - self._file: Path = Path(file) if file else Path(config.config_path).joinpath("notifications.json") - "File to store notification targets." - self._client: httpx.AsyncClient = client or async_client() - "HTTP client to send requests." - self._encoder: Encoder = encoder or Encoder() - "Encoder to encode data." - self._version: str = config.app_version - "Application version." - self._offload: BackgroundWorker = background_worker or BackgroundWorker.get_instance() - "Background worker to offload tasks to." - - if self._file.exists() and "600" != self._file.stat().st_mode: - try: - self._file.chmod(0o600) - except Exception: - pass - - @staticmethod - def get_instance( - file: str | None = None, - client: httpx.AsyncClient | None = None, - encoder: Encoder | None = None, - config: Config | None = None, - background_worker: BackgroundWorker | None = None, - ) -> "Notification": - """ - Get the instance of the class. - """ - return Notification( - file=file, client=client, encoder=encoder, config=config, background_worker=background_worker - ) - - def attach(self, _: web.Application): - """ - Attach the class to the application. - - Args: - _ (web.Application): The aiohttp application. - - """ - self.load() - EventBus.get_instance().subscribe(NotificationEvents.events(), self.emit, f"{__class__.__name__}.emit") - - def get_targets(self) -> list[Target]: - """Get the list of notification targets.""" - return self._targets - - def clear(self) -> "Notification": - """Clear the list of notification targets.""" - self._targets.clear() - return self - - def save(self, targets: list[Target]) -> "Notification": - """ - Save notification targets to the file. - - Args: - targets (list[Target]|None): The list of targets to save. - - Returns: - Notification: The Notification instance. - - """ - try: - self._file.write_text(json.dumps([t.serialize() for t in targets], indent=4)) - LOG.info(f"Updated '{self._file}'.") - except Exception as e: - LOG.exception(e) - LOG.error(f"Error saving '{self._file}'. '{e!s}'") - - return self - - def load(self) -> "Notification": - """Load or reload notification targets from the file.""" - if len(self._targets) > 0: - self.clear() - - if not self._file.exists() or self._file.stat().st_size < 1: - return self - - targets: list = [] - - try: - LOG.info(f"Loading '{self._file}'.") - targets = json.loads(self._file.read_text()) - except Exception as e: - LOG.error(f"Error loading '{self._file}'. '{e!s}'") - - if not targets or len(targets) < 1: - LOG.debug(f"No targets were found in '{self._file}'.") - return self - - need_save = False - - for target in targets: - try: - if "enabled" not in target: - target["enabled"] = True - need_save = True - - try: - Notification.validate(target) - target: Target = self.make_target(target) - except ValueError as e: - name = target.get("name") or target.get("id") or target.get("request", {}).get("url") or "unknown" - LOG.error(f"Invalid notification target '{name}'. '{e!s}'") - continue - - self._targets.append(target) - - LOG.info( - f"Send '{target.request.type}' request on '{', '.join(target.on) if len(target.on) > 0 else 'all events'}' to '{target.name}'." - ) - except Exception as e: - LOG.error(f"Error loading notification target '{target}'. '{e!s}'") - - if need_save: - LOG.warning("Saving notifications due to schema changes.") - self.save(self._targets) - - return self - - def make_target(self, target: dict) -> Target: - """ - Make a notification target from a dictionary. - - Args: - target (dict): The target details. - - Returns: - Target: The notification target. - - """ - return Target( - id=target.get("id"), - name=target.get("name"), - on=target.get("on", []), - presets=target.get("presets", []), - enabled=target.get("enabled", True), - request=TargetRequest( - type=target.get("request", {}).get("type", "json"), - method=target.get("request", {}).get("method", "POST"), - url=target.get("request", {}).get("url"), - data_key=target.get("request", {}).get("data_key", "data"), - headers=[ - TargetRequestHeader( - key=str(h.get("key", "")).strip(), - value=str(h.get("value", "")).strip(), - ) - for h in target.get("request", {}).get("headers", []) - ], - ), - ) - - @staticmethod - def validate(target: Target | dict) -> bool: - """ - Validate a notification target. - - Args: - target (Target|dict): The target to validate. - - Returns: - bool: True if the target is valid, False otherwise. - - """ - if not isinstance(target, dict): - target = target.serialize() - - if "id" not in target or validate_uuid(target["id"], version=4) is False: - msg = "Invalid notification target. No ID found." - raise ValueError(msg) - - if "name" not in target: - msg = "Invalid notification target. No name found." - raise ValueError(msg) - - if "request" not in target: - msg = "Invalid notification target. No request details found." - raise ValueError(msg) - - if "url" not in target["request"]: - msg = "Invalid notification target. No URL found." - raise ValueError(msg) - - if "data_key" not in target["request"]: - target["request"]["data_key"] = "data" - - if "enabled" not in target: - target["enabled"] = True - - if target.get("enabled") is not None and not isinstance(target.get("enabled"), bool): - msg = "Enabled must be a boolean." - raise ValueError(msg) - - if "method" in target["request"] and target["request"]["method"].upper() not in ["POST", "PUT"]: - msg = "Invalid notification target. Invalid method found." - raise ValueError(msg) - - if "type" in target["request"] and target["request"]["type"].lower() not in ["json", "form"]: - msg = "Invalid notification target. Invalid type found." - raise ValueError(msg) - - if "on" in target: - if not isinstance(target["on"], list): - msg = "Invalid notification target. Invalid 'on' event list found." - raise ValueError(msg) - - removed_events: list = [] - all_events: dict[str, str] = NotificationEvents.get_events().values() - for e in target["on"]: - if e not in all_events: - removed_events.append(e) - target["on"].remove(e) - continue - - if len(removed_events) > 0 and len(target["on"]) < 1: - msg: str = f"Invalid notification target. Invalid events '{', '.join(removed_events)}' found." - raise ValueError(msg) - - if "presets" in target: - if not isinstance(target["presets"], list): - msg = "Invalid notification target. Invalid 'presets' list found." - raise ValueError(msg) - - removed_presets: list = [] - all_presets: list[Preset] = Presets.get_instance().get_all() - - for p in target["presets"]: - if p not in [ap.name for ap in all_presets]: - removed_presets.append(p) - target["presets"].remove(p) - continue - - if len(removed_presets) > 0 and len(target["presets"]) < 1: - msg: str = f"Invalid notification target. Invalid presets '{', '.join(removed_presets)}' found." - raise ValueError(msg) - - if "headers" in target["request"]: - if not isinstance(target["request"]["headers"], list): - msg = "Invalid notification target. Invalid headers list found." - raise ValueError(msg) - - for h in target["request"]["headers"]: - if "key" not in h: - msg = "Invalid notification target. No header key found." - raise ValueError(msg) - if "value" not in h: - msg = "Invalid notification target. No header value found." - raise ValueError(msg) - - return True - - async def send(self, ev: Event, wait: bool = True) -> list[dict] | Awaitable[list[dict]]: - if len(self._targets) < 1: - return [] - - if not isinstance(ev.data, ItemDTO) and not isinstance(ev.data, dict): - LOG.debug(f"Received invalid item type '{type(ev.data)}' with event '{ev.event}'.") - return [] - - tasks = [] - - apprise_targets: list[Target] = [] - - for target in self._targets: - if not target.enabled: - continue - - if len(target.on) > 0 and ev.event not in target.on and "test" != ev.event: - continue - - if "test" != ev.event and not self._check_preset(target, ev): - continue - - if not target.request.url.startswith("http"): - apprise_targets.append(target) - else: - tasks.append(self._send(target, ev)) - - if len(apprise_targets) > 0: - tasks.append(self._apprise(apprise_targets, ev)) - - if wait: - return await asyncio.gather(*tasks) - - return tasks - - def _check_preset(self, target: Target, ev: Event) -> bool: - if len(target.presets) < 1: - return True - - if not isinstance(ev.data, (Item, ItemDTO, dict)): - return False - - preset_name: str | None = None - - if isinstance(ev.data, Item): - preset_name = ev.data.preset - - if isinstance(ev.data, ItemDTO): - preset_name = ev.data.preset - - if isinstance(ev.data, dict): - preset_name = ev.data.get("preset", None) - - if not preset_name: - return False - - return preset_name in target.presets - - async def _apprise(self, target: list[Target], ev: Event) -> dict: - if not target or not isinstance(target, list): - return {} - - import apprise - - try: - notify = apprise.Apprise(debug=self._debug) - apr_config = Path(Config.get_instance().apprise_config) - if apr_config.exists(): - apprise_config = notify.AppriseConfig() - apprise_config.add(apr_config) - notify.add(apprise_config) - - for t in target: - notify.add(t.request.url) - - status = await notify.async_notify( - body=ev.message or self._encoder.encode(ev.serialize()), - title=ev.title or f"YTPTube Event: {ev.event}", - ) - - if not status: - msg = "Apprise failed to send notification." - raise RuntimeError(msg) # noqa: TRY301 - except Exception as e: - LOG.exception(e) - LOG.error(f"Error sending Apprise notification: {e!s}") - return {"error": str(e), "event": ev.event, "id": ev.id, "targets": [t.name for t in target]} - - return {} - - async def _send(self, target: Target, ev: Event) -> dict: - try: - LOG.info(f"Sending Notification event '{ev.event}: {ev.id}' to '{target.name}'.") - - headers: dict[str, str] = { - "User-Agent": f"YTPTube/{self._version}", - "X-Event-Id": ev.id, - "X-Event": ev.event, - "Content-Type": "application/json" - if "json" == target.request.type.lower() - else "application/x-www-form-urlencoded", - } - - if len(target.request.headers) > 0: - headers.update({h.key: h.value for h in target.request.headers if h.key and h.value}) - - payload: dict = ev.serialize() - - if "data" != target.request.data_key: - payload[target.request.data_key] = payload["data"] - payload.pop("data", None) - - if "form" == target.request.type.lower(): - payload[target.request.data_key] = self._encoder.encode(payload[target.request.data_key]) - else: - payload = self._encoder.encode(payload) - - response = await self._client.request( - method=target.request.method.upper(), - url=target.request.url, - headers=headers, - data=payload if "form" == target.request.type.lower() else None, - content=payload if "json" == target.request.type.lower() else None, - ) - - respData: dict[str, Any] = { - "url": target.request.url, - "status": response.status_code, - "text": response.text, - } - - msg: str = f"Notification target '{target.name}' Responded to event '{ev.event}: {ev.id}' with status '{response.status_code}'." - if self._debug and respData.get("text"): - msg += f" body '{respData.get('text', '??')}'." - - LOG.info(msg) - - return respData - except Exception as e: - err_msg = str(e) - if not err_msg: - err_msg: str = type(e).__name__ - - tb = "".join(traceback.format_exception(type(e), e, e.__traceback__)) - LOG.error(f"Error sending Notification event '{ev.event}: {ev.id}' to '{target.name}'. '{err_msg!s}'. {tb}") - return {"url": target.request.url, "status": 500, "text": str(ev)} - - def emit(self, e: Event, _, **__) -> None: - if len(self._targets) < 1 or not NotificationEvents.is_valid(e.event): - return - - self._offload.submit(self.send, e) - return - - async def noop(self) -> None: - return None diff --git a/app/library/router.py b/app/library/router.py index 2efc793e..22e356b9 100644 --- a/app/library/router.py +++ b/app/library/router.py @@ -1,8 +1,9 @@ import logging import re -from collections.abc import Awaitable +from collections.abc import Awaitable, Callable from enum import Enum from functools import wraps +from typing import Any LOG: logging.Logger = logging.getLogger(__name__) @@ -25,15 +26,15 @@ class Route: method (str): The HTTP method (GET, POST, etc.). path (str): The path for the route. name (str): The name of the route. - handler (Awaitable): The function that handles the route. + handler (Callable[..., Awaitable]): The function that handles the route. """ - def __init__(self, method: str, path: str, name: str, handler: Awaitable): + def __init__(self, method: str, path: str, name: str, handler: Callable[..., Awaitable[Any]]): self.method: str = method.upper() self.path: str = path self.name: str = name - self.handler: Awaitable = handler + self.handler: Callable[..., Awaitable[Any]] = handler ROUTES: dict[str, dict[str, Route]] = {} @@ -55,7 +56,12 @@ def make_route_name(method: str, path: str) -> str: return f"{method}:" + ".".join(segments or ["root"]) -def route(method: RouteType | str | list[str], path: str, name: str | None = None, **kwargs) -> Awaitable: +def route( + method: RouteType | str | list[str], + path: str, + name: str | None = None, + **kwargs, +) -> Callable[[Callable[..., Awaitable[Any]]], Callable[..., Awaitable[Any]]]: """ Decorator to mark a method as an HTTP route handler. @@ -71,7 +77,7 @@ def route(method: RouteType | str | list[str], path: str, name: str | None = Non """ methods = [method] if isinstance(method, (str, RouteType)) else method - def decorator(func): + def decorator(func: Callable[..., Awaitable[Any]]): @wraps(func) async def wrapper(*args, **kwargs): return await func(*args, **kwargs) @@ -97,7 +103,13 @@ def route(method: RouteType | str | list[str], path: str, name: str | None = Non return decorator -def add_route(method: RouteType | str | list[str], path: str, handler: Awaitable, name: str | None = None, **kwargs): +def add_route( + method: RouteType | str | list[str], + path: str, + handler: Callable[..., Awaitable[Any]], + name: str | None = None, + **kwargs, +): """ Decorator to mark a method as an HTTP route handler. @@ -129,7 +141,7 @@ def add_route(method: RouteType | str | list[str], path: str, handler: Awaitable ) -def get_route(route_type: RouteType, name: str) -> dict[str, Route] | None: +def get_route(route_type: RouteType, name: str) -> Route | None: """ Get the route information by name. diff --git a/app/main.py b/app/main.py index 0f5e5b06..bbab7563 100644 --- a/app/main.py +++ b/app/main.py @@ -16,6 +16,7 @@ from aiohttp import web from app.features.conditions.service import Conditions from app.features.dl_fields.service import DLFields +from app.features.notifications.service import Notifications from app.library.BackgroundWorker import BackgroundWorker from app.library.cache import Cache from app.library.config import Config @@ -23,7 +24,6 @@ from app.library.downloads import DownloadQueue from app.library.Events import EventBus, Events from app.library.HttpAPI import HttpAPI from app.library.HttpSocket import HttpSocket -from app.library.Notifications import Notification from app.library.Presets import Presets from app.library.Scheduler import Scheduler from app.library.Services import Services @@ -118,7 +118,7 @@ class Main: Presets.get_instance().attach(self._app) Tasks.get_instance().attach(self._app) - Notification.get_instance().attach(self._app) + Notifications.get_instance().attach(self._app) Conditions.get_instance().attach(self._app) DLFields.get_instance().attach(self._app) TaskDefinitions.get_instance().attach(self._app) diff --git a/app/migrations/20260121134838_add_notifications_table.py b/app/migrations/20260121134838_add_notifications_table.py new file mode 100644 index 00000000..0ef80bb5 --- /dev/null +++ b/app/migrations/20260121134838_add_notifications_table.py @@ -0,0 +1,40 @@ +""" +This module contains a db migration. + +Migration Name: add_notifications_table +Migration Version: 20260121134838 +""" + +from sqlalchemy import text + + +async def upgrade(c): + sql: list[str] = [ + """ + CREATE TABLE IF NOT EXISTS "notifications" ( + "id" INTEGER PRIMARY KEY AUTOINCREMENT, + "name" TEXT NOT NULL UNIQUE, + "on" JSON NOT NULL DEFAULT '[]', + "presets" JSON NOT NULL DEFAULT '[]', + "enabled" BOOLEAN NOT NULL DEFAULT 1, + "request_url" TEXT NOT NULL, + "request_method" TEXT NOT NULL DEFAULT 'POST', + "request_type" TEXT NOT NULL DEFAULT 'json', + "request_data_key" TEXT NOT NULL DEFAULT 'data', + "request_headers" JSON NOT NULL DEFAULT '[]', + "created_at" DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP, + "updated_at" DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP + ); + """, + 'CREATE INDEX IF NOT EXISTS "ix_notifications_name" ON "notifications" ("name");', + 'CREATE INDEX IF NOT EXISTS "ix_notifications_enabled" ON "notifications" ("enabled");', + ] + for sql_stmt in sql: + await c.execute(text(sql_stmt)) + + +async def downgrade(c): + sql = """ + DROP TABLE IF EXISTS "notifications"; + """ + await c.execute(text(sql)) diff --git a/app/routes/api/notifications.py b/app/routes/api/notifications.py index 6174b62b..efb10124 100644 --- a/app/routes/api/notifications.py +++ b/app/routes/api/notifications.py @@ -1,108 +1,3 @@ -import logging -import uuid +"""Migrated API routes for feature submodule.""" -from aiohttp import web -from aiohttp.web import Request, Response - -from app.library.encoder import Encoder -from app.library.Events import EventBus, Events -from app.library.Notifications import Notification, NotificationEvents -from app.library.router import route -from app.library.Utils import validate_uuid - -LOG: logging.Logger = logging.getLogger(__name__) - - -@route("GET", "api/notifications/", "notifications_list") -async def notifications_list(encoder: Encoder) -> Response: - """ - Get the notification targets. - - Args: - encoder (Encoder): The encoder instance. - - Returns: - Response: The response object. - - """ - return web.json_response( - data={ - "notifications": Notification.get_instance().get_targets(), - "allowedTypes": list(NotificationEvents.get_events().values()), - }, - status=web.HTTPOk.status_code, - dumps=encoder.encode, - ) - - -@route("PUT", "api/notifications/", "notification_add") -async def notification_add(request: Request, encoder: Encoder) -> Response: - """ - Add notification targets. - - Args: - request (Request): The request object. - encoder (Encoder): The encoder instance. - - Returns: - Response: The response object. - - """ - post = await request.json() - if not isinstance(post, list): - return web.json_response( - {"error": "Invalid request body expecting list with dicts."}, - status=web.HTTPBadRequest.status_code, - ) - - targets: list = [] - - ins: Notification = Notification.get_instance() - for item in post: - if not isinstance(item, dict): - return web.json_response( - {"error": "Invalid request body expecting list with dicts."}, - status=web.HTTPBadRequest.status_code, - ) - - if not item.get("id", None) or validate_uuid(item.get("id"), version=4): - item["id"] = str(uuid.uuid4()) - - try: - Notification.validate(item) - except ValueError as e: - return web.json_response( - {"error": f"Invalid notification target settings. {e!s}", "data": item}, - status=web.HTTPBadRequest.status_code, - ) - - targets.append(ins.make_target(item)) - - try: - ins.save(targets=targets) - ins.load() - except Exception as e: - LOG.exception(e) - return web.json_response({"error": "Failed to save tasks."}, status=web.HTTPInternalServerError.status_code) - - data = {"notifications": targets, "allowedTypes": list(NotificationEvents.get_events().values())} - - return web.json_response(data=data, status=web.HTTPOk.status_code, dumps=encoder.encode) - - -@route("POST", "api/notifications/test/", "notification_test") -async def notification_test(encoder: Encoder, notify: EventBus) -> Response: - """ - Test the notification. - - Args: - encoder (Encoder): The encoder instance. - notify (EventBus): The event bus instance. - - Returns: - Response: The response object. - - """ - notify.emit(Events.TEST, title="Test Notification", message="This is a test notification.") - - return web.json_response(data={}, status=web.HTTPOk.status_code, dumps=encoder.encode) +import app.features.notifications.router # noqa: F401 diff --git a/app/tests/test_notifications.py b/app/tests/test_notifications.py deleted file mode 100644 index 4fd912f9..00000000 --- a/app/tests/test_notifications.py +++ /dev/null @@ -1,921 +0,0 @@ -import json -import tempfile -import uuid -from pathlib import Path -from unittest.mock import AsyncMock, Mock, patch - -import httpx -import pytest -from aiohttp import web - -from app.library.BackgroundWorker import BackgroundWorker -from app.library.encoder import Encoder -from app.library.Events import Event -from app.library.ItemDTO import ItemDTO -from app.library.Notifications import ( - Notification, - NotificationEvents, - Target, - TargetRequest, - TargetRequestHeader, -) - - -class TestTargetRequestHeader: - """Test the TargetRequestHeader dataclass.""" - - def test_target_request_header_creation(self): - """Test creating a TargetRequestHeader.""" - header = TargetRequestHeader(key="Authorization", value="Bearer token123") - - assert header.key == "Authorization" - assert header.value == "Bearer token123" - - def test_target_request_header_serialize(self): - """Test serializing a TargetRequestHeader.""" - header = TargetRequestHeader(key="Content-Type", value="application/json") - serialized = header.serialize() - - expected = {"key": "Content-Type", "value": "application/json"} - assert serialized == expected - - def test_target_request_header_json(self): - """Test JSON serialization of TargetRequestHeader.""" - header = TargetRequestHeader(key="X-API-Key", value="secret123") - json_str = header.json() - - # Should be valid JSON - parsed = json.loads(json_str) - assert parsed["key"] == "X-API-Key" - assert parsed["value"] == "secret123" - - def test_target_request_header_get(self): - """Test get method of TargetRequestHeader.""" - header = TargetRequestHeader(key="Authorization", value="Bearer token") - - assert header.get("key") == "Authorization" - assert header.get("value") == "Bearer token" - assert header.get("nonexistent", "default") == "default" - - -class TestTargetRequest: - """Test the TargetRequest dataclass.""" - - def test_target_request_creation_defaults(self): - """Test creating a TargetRequest with defaults.""" - request = TargetRequest(type="json", method="POST", url="https://example.com/webhook") - - assert request.type == "json" - assert request.method == "POST" - assert request.url == "https://example.com/webhook" - assert request.headers == [] - assert request.data_key == "data" - - def test_target_request_creation_with_headers(self): - """Test creating a TargetRequest with headers.""" - headers = [ - TargetRequestHeader(key="Authorization", value="Bearer token"), - TargetRequestHeader(key="Content-Type", value="application/json"), - ] - request = TargetRequest( - type="json", method="POST", url="https://example.com/webhook", headers=headers, data_key="payload" - ) - - assert len(request.headers) == 2 - assert request.headers[0].key == "Authorization" - assert request.data_key == "payload" - - def test_target_request_serialize(self): - """Test serializing a TargetRequest.""" - headers = [TargetRequestHeader(key="X-Token", value="abc123")] - request = TargetRequest( - type="json", method="PUT", url="https://api.example.com/notify", headers=headers, data_key="content" - ) - - serialized = request.serialize() - expected = { - "type": "json", - "method": "PUT", - "url": "https://api.example.com/notify", - "data_key": "content", - "headers": [{"key": "X-Token", "value": "abc123"}], - } - assert serialized == expected - - def test_target_request_json(self): - """Test JSON serialization of TargetRequest.""" - request = TargetRequest(type="form", method="POST", url="https://webhook.site/test") - json_str = request.json() - - parsed = json.loads(json_str) - assert parsed["type"] == "form" - assert parsed["method"] == "POST" - assert parsed["url"] == "https://webhook.site/test" - - def test_target_request_get(self): - """Test get method of TargetRequest.""" - request = TargetRequest(type="json", method="POST", url="https://example.com/webhook") - - assert request.get("type") == "json" - assert request.get("method") == "POST" - assert request.get("nonexistent", "default") == "default" - - -class TestTarget: - """Test the Target dataclass.""" - - def test_target_creation_minimal(self): - """Test creating a Target with minimal required fields.""" - request = TargetRequest(type="json", method="POST", url="https://example.com/webhook") - target = Target(id=str(uuid.uuid4()), name="Test Webhook", request=request) - - assert target.name == "Test Webhook" - assert target.on == [] - assert target.presets == [] - assert target.enabled is True - assert target.request == request - - def test_target_creation_full(self): - """Test creating a Target with all fields.""" - target_id = str(uuid.uuid4()) - request = TargetRequest(type="json", method="POST", url="https://example.com/webhook") - target = Target( - id=target_id, - name="Full Test Webhook", - on=["item_completed", "item_failed"], - presets=["default", "audio_only"], - enabled=False, - request=request, - ) - - assert target.id == target_id - assert target.name == "Full Test Webhook" - assert target.on == ["item_completed", "item_failed"] - assert target.presets == ["default", "audio_only"] - assert target.enabled is False - - def test_target_serialize(self): - """Test serializing a Target.""" - target_id = str(uuid.uuid4()) - request = TargetRequest(type="json", method="POST", url="https://example.com/webhook") - target = Target(id=target_id, name="Test Target", on=["item_completed"], presets=["default"], request=request) - - serialized = target.serialize() - assert serialized["id"] == target_id - assert serialized["name"] == "Test Target" - assert serialized["on"] == ["item_completed"] - assert serialized["presets"] == ["default"] - assert isinstance(serialized["request"], dict) - - def test_target_json(self): - """Test JSON serialization of Target.""" - request = TargetRequest(type="json", method="POST", url="https://example.com/webhook") - target = Target(id=str(uuid.uuid4()), name="JSON Test", request=request) - - json_str = target.json() - parsed = json.loads(json_str) - assert parsed["name"] == "JSON Test" - - def test_target_get(self): - """Test get method of Target.""" - request = TargetRequest(type="json", method="POST", url="https://example.com/webhook") - target = Target(id=str(uuid.uuid4()), name="Get Test", request=request) - - assert target.get("name") == "Get Test" - assert target.get("nonexistent", "default") == "default" - - -class TestNotificationEvents: - """Test the NotificationEvents class.""" - - def test_notification_events_constants(self): - """Test that NotificationEvents has expected constants.""" - assert hasattr(NotificationEvents, "TEST") - assert hasattr(NotificationEvents, "ITEM_ADDED") - assert hasattr(NotificationEvents, "ITEM_COMPLETED") - assert hasattr(NotificationEvents, "ITEM_CANCELLED") - assert hasattr(NotificationEvents, "ITEM_DELETED") - assert hasattr(NotificationEvents, "LOG_INFO") - assert hasattr(NotificationEvents, "LOG_ERROR") - - def test_get_events(self): - """Test get_events static method.""" - events = NotificationEvents.get_events() - - assert isinstance(events, dict) - assert "TEST" in events - assert "ITEM_ADDED" in events - assert "ITEM_COMPLETED" in events - - def test_events_function(self): - """Test events function.""" - events_list = NotificationEvents.events() - - assert isinstance(events_list, list) - assert len(events_list) > 0 - - def test_is_valid(self): - """Test is_valid static method.""" - assert NotificationEvents.is_valid("test"), "Valid events" - assert NotificationEvents.is_valid("item_added") - assert NotificationEvents.is_valid("item_completed") - - assert not NotificationEvents.is_valid("invalid_event"), "Invalid events" - assert not NotificationEvents.is_valid("") - assert not NotificationEvents.is_valid(None) - - -class TestNotification: - """Test the Notification singleton class.""" - - def setup_method(self): - """Set up test environment before each test.""" - # Reset singleton instance to ensure clean state - Notification._reset_singleton() - - def teardown_method(self): - """Clean up after each test.""" - # Reset singleton instance to prevent test pollution - Notification._reset_singleton() - - @patch("app.library.Notifications.Config") - @patch("app.library.Notifications.BackgroundWorker") - def test_notification_singleton(self, mock_background_worker, mock_config): - """Test that Notification follows singleton pattern.""" - mock_config.get_instance.return_value = Mock(debug=False, config_path="/tmp", app_version="1.0.0") - mock_background_worker.get_instance.return_value = Mock() - - instance1 = Notification.get_instance() - instance2 = Notification.get_instance() - - assert instance1 is instance2 - assert isinstance(instance1, Notification) - - @patch("app.library.Notifications.Config") - @patch("app.library.Notifications.BackgroundWorker") - def test_notification_init_default_params(self, mock_background_worker, mock_config): - """Test Notification initialization with default parameters.""" - mock_config_instance = Mock(debug=False, config_path="/tmp/test", app_version="1.0.0") - mock_config.get_instance.return_value = mock_config_instance - mock_background_worker.get_instance.return_value = Mock() - - with patch("pathlib.Path.exists", return_value=False): - notification = Notification.get_instance() - - assert notification._debug is False - assert notification._version == "1.0.0" - assert isinstance(notification._targets, list) - assert len(notification._targets) == 0 - - @patch("app.library.Notifications.Config") - @patch("app.library.Notifications.BackgroundWorker") - def test_notification_init_custom_params(self, mock_background_worker, mock_config): - """Test Notification initialization with custom parameters.""" - _ = mock_background_worker, mock_config # Suppress unused variable warnings - mock_config_instance = Mock(debug=True, config_path="/custom/path", app_version="2.0.0") - mock_client = Mock(spec=httpx.AsyncClient) - mock_encoder = Mock(spec=Encoder) - mock_worker = Mock(spec=BackgroundWorker) - - with tempfile.NamedTemporaryFile(delete=False) as temp_file: - temp_path = temp_file.name - - notification = Notification.get_instance( - file=temp_path, - client=mock_client, - encoder=mock_encoder, - config=mock_config_instance, - background_worker=mock_worker, - ) - - assert notification._debug is True - assert notification._version == "2.0.0" - assert notification._client == mock_client - assert notification._encoder == mock_encoder - assert notification._offload == mock_worker - - def test_get_targets_empty(self): - """Test get_targets when no targets are loaded.""" - with ( - patch("app.library.Notifications.Config") as mock_config, - patch("app.library.Notifications.BackgroundWorker") as mock_worker, - ): - mock_config.get_instance.return_value = Mock(debug=False, config_path="/tmp", app_version="1.0.0") - mock_worker.get_instance.return_value = Mock() - - notification = Notification.get_instance() - targets = notification.get_targets() - - assert isinstance(targets, list) - assert len(targets) == 0 - - def test_clear_targets(self): - """Test clearing notification targets.""" - with ( - patch("app.library.Notifications.Config") as mock_config, - patch("app.library.Notifications.BackgroundWorker") as mock_worker, - ): - mock_config.get_instance.return_value = Mock(debug=False, config_path="/tmp", app_version="1.0.0") - mock_worker.get_instance.return_value = Mock() - - notification = Notification.get_instance() - - # Add a dummy target - notification._targets = ["dummy_target"] - assert len(notification._targets) == 1 - - # Clear targets - result = notification.clear() - - assert result == notification # Should return self - assert len(notification._targets) == 0 - - @patch("app.library.Notifications.Config") - @patch("app.library.Notifications.BackgroundWorker") - def test_save_targets(self, mock_worker, mock_config): - """Test saving targets to file.""" - mock_config.get_instance.return_value = Mock(debug=False, config_path="/tmp", app_version="1.0.0") - mock_worker.get_instance.return_value = Mock() - - with tempfile.NamedTemporaryFile(mode="w", delete=False) as temp_file: - temp_path = temp_file.name - - # Create a test target - request = TargetRequest(type="json", method="POST", url="https://example.com/webhook") - target = Target(id=str(uuid.uuid4()), name="Test Target", request=request) - - notification = Notification.get_instance(file=temp_path) - result = notification.save([target]) - - assert result == notification # Should return self - - # Verify file was written - with open(temp_path) as f: - saved_data = json.load(f) - - assert isinstance(saved_data, list) - assert len(saved_data) == 1 - assert saved_data[0]["name"] == "Test Target" - - # Clean up - Path(temp_path).unlink() - - @patch("app.library.Notifications.Config") - @patch("app.library.Notifications.BackgroundWorker") - def test_load_targets_empty_file(self, mock_worker, mock_config): - """Test loading targets from empty file.""" - mock_config.get_instance.return_value = Mock(debug=False, config_path="/tmp", app_version="1.0.0") - mock_worker.get_instance.return_value = Mock() - - with tempfile.NamedTemporaryFile(mode="w", delete=False) as temp_file: - temp_file.write("") # Empty file - temp_path = temp_file.name - - notification = Notification.get_instance(file=temp_path) - result = notification.load() - - assert result == notification # Should return self - assert len(notification._targets) == 0 - - # Clean up - Path(temp_path).unlink() - - @patch("app.library.Notifications.Config") - @patch("app.library.Notifications.BackgroundWorker") - @patch("app.library.Notifications.Presets") - def test_load_targets_schema_update(self, mock_presets, mock_worker, mock_config): - """Test that load automatically updates file schema when enabled field is missing.""" - mock_config.get_instance.return_value = Mock(debug=False, config_path="/tmp", app_version="1.0.0") - mock_worker.get_instance.return_value = Mock() - mock_preset = Mock() - mock_preset.name = "default" - mock_presets.get_instance.return_value.get_all.return_value = [mock_preset] - - # Create target data without enabled field (old schema) - target_data = [ - { - "id": str(uuid.uuid4()), - "name": "Old Schema Target", - "on": ["item_completed"], - "presets": ["default"], - "request": {"type": "json", "method": "POST", "url": "https://example.com/webhook"}, - } - ] - - with tempfile.NamedTemporaryFile(mode="w", delete=False) as temp_file: - json.dump(target_data, temp_file) - temp_path = temp_file.name - - notification = Notification.get_instance(file=temp_path) - - # Mock the save method to verify it's called - with patch.object(notification, "save") as mock_save: - notification.load() - - # Verify save was called due to schema update - mock_save.assert_called_once() - - assert len(notification._targets) == 1, "Verify the target has enabled=True by default" - assert notification._targets[0].enabled is True - - # Clean up - Path(temp_path).unlink() - - @patch("app.library.Notifications.Config") - @patch("app.library.Notifications.BackgroundWorker") - @patch("app.library.Notifications.Presets") - def test_load_targets_valid_file(self, mock_presets, mock_worker, mock_config): - """Test loading targets from valid file.""" - mock_config.get_instance.return_value = Mock(debug=False, config_path="/tmp", app_version="1.0.0") - mock_worker.get_instance.return_value = Mock() - - # Mock preset with name attribute - mock_preset = Mock() - mock_preset.name = "default" - mock_presets.get_instance.return_value.get_all.return_value = [mock_preset] - - target_data = [ - { - "id": str(uuid.uuid4()), - "name": "Test Webhook", - "on": ["item_completed"], - "presets": ["default"], - "request": { - "type": "json", - "method": "POST", - "url": "https://example.com/webhook", - "data_key": "data", - "headers": [], - }, - } - ] - - with tempfile.NamedTemporaryFile(mode="w", delete=False) as temp_file: - json.dump(target_data, temp_file) - temp_path = temp_file.name - - notification = Notification.get_instance(file=temp_path) - result = notification.load() - - assert result == notification # Should return self - assert len(notification._targets) == 1 - assert notification._targets[0].name == "Test Webhook" - - # Clean up - Path(temp_path).unlink() - - def test_make_target(self): - """Test make_target method.""" - with ( - patch("app.library.Notifications.Config") as mock_config, - patch("app.library.Notifications.BackgroundWorker") as mock_worker, - ): - mock_config.get_instance.return_value = Mock(debug=False, config_path="/tmp", app_version="1.0.0") - mock_worker.get_instance.return_value = Mock() - - notification = Notification.get_instance() - - target_dict = { - "id": str(uuid.uuid4()), - "name": "Test Target", - "on": ["item_completed"], - "presets": ["default"], - "request": { - "type": "json", - "method": "POST", - "url": "https://example.com/webhook", - "data_key": "payload", - "headers": [{"key": "Authorization", "value": "Bearer token"}], - }, - } - - target = notification.make_target(target_dict) - - assert isinstance(target, Target) - assert target.name == "Test Target" - assert target.on == ["item_completed"] - assert target.presets == ["default"] - assert target.request.url == "https://example.com/webhook" - assert target.request.data_key == "payload" - assert len(target.request.headers) == 1 - assert target.request.headers[0].key == "Authorization" - - def test_validate_target_valid(self): - """Test validate method with valid target.""" - target_dict = { - "id": str(uuid.uuid4()), - "name": "Valid Target", - "request": {"url": "https://example.com/webhook"}, - } - - with ( - patch("app.library.Notifications.NotificationEvents.get_events") as mock_events, - patch("app.library.Notifications.Presets") as mock_presets, - ): - mock_events.return_value.values.return_value = ["item_completed"] - mock_presets.get_instance.return_value.get_all.return_value = [] - - result = Notification.validate(target_dict) - assert result is True - - def test_validate_target_missing_id(self): - """Test validate method with missing ID.""" - target_dict = {"name": "Missing ID Target", "request": {"url": "https://example.com/webhook"}} - - with pytest.raises(ValueError, match=r"Invalid notification target\. No ID found\."): - Notification.validate(target_dict) - - def test_validate_target_invalid_id(self): - """Test validate method with invalid UUID.""" - target_dict = { - "id": "invalid-uuid", - "name": "Invalid ID Target", - "request": {"url": "https://example.com/webhook"}, - } - - with pytest.raises(ValueError, match=r"Invalid notification target\. No ID found\."): - Notification.validate(target_dict) - - def test_validate_target_missing_name(self): - """Test validate method with missing name.""" - target_dict = {"id": str(uuid.uuid4()), "request": {"url": "https://example.com/webhook"}} - - with pytest.raises(ValueError, match=r"Invalid notification target\. No name found\."): - Notification.validate(target_dict) - - def test_validate_target_missing_request(self): - """Test validate method with missing request.""" - target_dict = {"id": str(uuid.uuid4()), "name": "Missing Request Target"} - - with pytest.raises(ValueError, match=r"Invalid notification target\. No request details found\."): - Notification.validate(target_dict) - - def test_validate_target_missing_url(self): - """Test validate method with missing URL.""" - target_dict = {"id": str(uuid.uuid4()), "name": "Missing URL Target", "request": {}} - - with pytest.raises(ValueError, match=r"Invalid notification target\. No URL found\."): - Notification.validate(target_dict) - - def test_validate_target_invalid_method(self): - """Test validate method with invalid HTTP method.""" - target_dict = { - "id": str(uuid.uuid4()), - "name": "Invalid Method Target", - "request": { - "url": "https://example.com/webhook", - "method": "GET", # Only POST and PUT are allowed - }, - } - - with pytest.raises(ValueError, match=r"Invalid notification target\. Invalid method found\."): - Notification.validate(target_dict) - - def test_validate_target_invalid_type(self): - """Test validate method with invalid request type.""" - target_dict = { - "id": str(uuid.uuid4()), - "name": "Invalid Type Target", - "request": { - "url": "https://example.com/webhook", - "type": "xml", # Only json and form are allowed - }, - } - - with pytest.raises(ValueError, match=r"Invalid notification target\. Invalid type found\."): - Notification.validate(target_dict) - - def test_validate_target_invalid_enabled(self): - """Test validating a target with invalid enabled field.""" - target_dict = { - "id": str(uuid.uuid4()), - "name": "Test Target", - "enabled": "yes", # Should be boolean - "request": { - "url": "https://example.com/webhook", - "method": "POST", - "type": "json", - }, - } - - with pytest.raises(ValueError, match=r"Enabled must be a boolean\."): - Notification.validate(target_dict) - - def test_validate_target_enabled_defaults_to_true(self): - """Test that enabled defaults to True when not provided.""" - target_dict = { - "id": str(uuid.uuid4()), - "name": "Test Target", - "request": { - "url": "https://example.com/webhook", - "method": "POST", - "type": "json", - }, - } - - result = Notification.validate(target_dict) - assert result is True - assert target_dict.get("enabled") is True - - @patch("app.library.Notifications.Config") - @patch("app.library.Notifications.BackgroundWorker") - @patch("app.library.Notifications.EventBus") - def test_attach(self, mock_eventbus, mock_worker, mock_config): - """Test attach method.""" - mock_config.get_instance.return_value = Mock(debug=False, config_path="/tmp", app_version="1.0.0") - mock_worker.get_instance.return_value = Mock() - mock_eventbus_instance = Mock() - mock_eventbus.get_instance.return_value = mock_eventbus_instance - - notification = Notification.get_instance() - app = Mock(spec=web.Application) - - with patch.object(notification, "load") as mock_load: - notification.attach(app) - - mock_load.assert_called_once() - mock_eventbus_instance.subscribe.assert_called_once() - - @pytest.mark.asyncio - @patch("app.library.Notifications.Config") - @patch("app.library.Notifications.BackgroundWorker") - async def test_send_no_targets(self, mock_worker, mock_config): - """Test send method with no targets.""" - mock_config.get_instance.return_value = Mock(debug=False, config_path="/tmp", app_version="1.0.0") - mock_worker.get_instance.return_value = Mock() - - notification = Notification.get_instance() - event = Event(event="test", data={"test": "data"}) - - result = await notification.send(event) - assert result == [] - - @pytest.mark.asyncio - @patch("app.library.Notifications.Config") - @patch("app.library.Notifications.BackgroundWorker") - async def test_send_invalid_event_data(self, mock_worker, mock_config): - """Test send method with invalid event data.""" - mock_config.get_instance.return_value = Mock(debug=False, config_path="/tmp", app_version="1.0.0") - mock_worker.get_instance.return_value = Mock() - - notification = Notification.get_instance() - # Add a target - request = TargetRequest(type="json", method="POST", url="https://example.com/webhook") - target = Target(id=str(uuid.uuid4()), name="Test Target", request=request) - notification._targets = [target] - - event = Event( - event="test", - data="invalid_string_data", # Should be ItemDTO or dict - ) - - result = await notification.send(event) - assert result == [] - - @pytest.mark.asyncio - @patch("app.library.Notifications.Config") - @patch("app.library.Notifications.BackgroundWorker") - async def test_send_with_http_target(self, mock_worker, mock_config): - """Test send method with HTTP target.""" - mock_config.get_instance.return_value = Mock(debug=False, config_path="/tmp", app_version="1.0.0") - mock_worker.get_instance.return_value = Mock() - - # Mock HTTP client - mock_response = Mock() - mock_response.status_code = 200 - mock_response.text = "OK" - mock_client = AsyncMock() - mock_client.request = AsyncMock(return_value=mock_response) - - notification = Notification.get_instance(client=mock_client) - - # Add HTTP target - request = TargetRequest(type="json", method="POST", url="https://example.com/webhook") - target = Target(id=str(uuid.uuid4()), name="Test HTTP Target", request=request) - notification._targets = [target] - - # Create test event - item_dto = ItemDTO( - id="test_id", url="https://youtube.com/watch?v=test", title="Test Video", folder="/downloads" - ) - event = Event(event="item_completed", data=item_dto) - - result = await notification.send(event) - - assert len(result) == 1 - assert result[0]["status"] == 200 - assert result[0]["text"] == "OK" - mock_client.request.assert_called_once() - - @pytest.mark.asyncio - @patch("app.library.Notifications.Config") - @patch("app.library.Notifications.BackgroundWorker") - async def test_send_skips_disabled_targets(self, mock_worker, mock_config): - """Test that send method skips disabled targets.""" - mock_config.get_instance.return_value = Mock(debug=False, config_path="/tmp", app_version="1.0.0") - mock_worker.get_instance.return_value = Mock() - - # Mock HTTP client - mock_response = Mock() - mock_response.status_code = 200 - mock_response.text = "OK" - mock_client = AsyncMock() - mock_client.request = AsyncMock(return_value=mock_response) - - notification = Notification.get_instance(client=mock_client) - - # Add enabled and disabled targets - enabled_request = TargetRequest(type="json", method="POST", url="https://example.com/enabled") - enabled_target = Target(id=str(uuid.uuid4()), name="Enabled Target", enabled=True, request=enabled_request) - - disabled_request = TargetRequest(type="json", method="POST", url="https://example.com/disabled") - disabled_target = Target(id=str(uuid.uuid4()), name="Disabled Target", enabled=False, request=disabled_request) - - notification._targets = [enabled_target, disabled_target] - - # Create test event - item_dto = ItemDTO( - id="test_id", url="https://youtube.com/watch?v=test", title="Test Video", folder="/downloads" - ) - event = Event(event="item_completed", data=item_dto) - - result = await notification.send(event) - - assert len(result) == 1, "Only enabled target should be called" - assert result[0]["status"] == 200 - mock_client.request.assert_called_once() - - @pytest.mark.asyncio - @patch("app.library.Notifications.Config") - @patch("app.library.Notifications.BackgroundWorker") - async def test_send_with_apprise_target(self, mock_worker, mock_config): - """Test send method with Apprise target.""" - mock_config.get_instance.return_value = Mock( - debug=False, config_path="/tmp", app_version="1.0.0", apprise_config="/tmp/apprise.conf" - ) - mock_worker.get_instance.return_value = Mock() - - notification = Notification.get_instance() - - # Add Apprise target (non-HTTP URL) - request = TargetRequest(type="json", method="POST", url="discord://webhook_id/webhook_token") - target = Target(id=str(uuid.uuid4()), name="Test Discord Target", request=request) - notification._targets = [target] - - # Create test event - event = Event(event="item_completed", data={"test": "data"}) - - with patch("app.library.Notifications.Path") as mock_path, patch("builtins.__import__") as mock_import: - # Mock apprise config file doesn't exist - mock_path.return_value.exists.return_value = False - - # Mock apprise import - mock_apprise = Mock() - mock_notify = Mock() - # Mock async_notify as an AsyncMock that returns True - mock_notify.async_notify = AsyncMock(return_value=True) - mock_apprise.Apprise.return_value = mock_notify - mock_import.return_value = mock_apprise - - result = await notification.send(event) - - assert len(result) == 1, "Should return empty dict from _apprise method" - assert result[0] == {} - - def test_check_preset_no_presets(self): - """Test _check_preset method with target having no preset filters.""" - with ( - patch("app.library.Notifications.Config") as mock_config, - patch("app.library.Notifications.BackgroundWorker") as mock_worker, - ): - mock_config.get_instance.return_value = Mock(debug=False, config_path="/tmp", app_version="1.0.0") - mock_worker.get_instance.return_value = Mock() - - notification = Notification.get_instance() - - request = TargetRequest(type="json", method="POST", url="https://example.com/webhook") - target = Target( - id=str(uuid.uuid4()), - name="No Preset Filter", - presets=[], # No preset filter - request=request, - ) - - event = Event(event="item_completed", data={"preset": "default"}) - - result = notification._check_preset(target, event) - assert result is True # Should pass when no preset filter - - def test_check_preset_with_matching_preset(self): - """Test _check_preset method with matching preset.""" - with ( - patch("app.library.Notifications.Config") as mock_config, - patch("app.library.Notifications.BackgroundWorker") as mock_worker, - ): - mock_config.get_instance.return_value = Mock(debug=False, config_path="/tmp", app_version="1.0.0") - mock_worker.get_instance.return_value = Mock() - - notification = Notification.get_instance() - - request = TargetRequest(type="json", method="POST", url="https://example.com/webhook") - target = Target( - id=str(uuid.uuid4()), name="Preset Filter", presets=["default", "audio_only"], request=request - ) - - # Test with ItemDTO - item_dto = ItemDTO( - id="test_id", url="https://youtube.com/test", title="Test Video", folder="/downloads", preset="default" - ) - event = Event(event="item_completed", data=item_dto) - - result = notification._check_preset(target, event) - assert result is True - - def test_check_preset_with_non_matching_preset(self): - """Test _check_preset method with non-matching preset.""" - with ( - patch("app.library.Notifications.Config") as mock_config, - patch("app.library.Notifications.BackgroundWorker") as mock_worker, - ): - mock_config.get_instance.return_value = Mock(debug=False, config_path="/tmp", app_version="1.0.0") - mock_worker.get_instance.return_value = Mock() - - notification = Notification.get_instance() - - request = TargetRequest(type="json", method="POST", url="https://example.com/webhook") - target = Target(id=str(uuid.uuid4()), name="Preset Filter", presets=["audio_only"], request=request) - - event = Event( - event="item_completed", - data={"preset": "video_only"}, # Different preset - ) - - result = notification._check_preset(target, event) - assert result is False - - def test_emit_invalid_event(self): - """Test emit method with invalid event.""" - with ( - patch("app.library.Notifications.Config") as mock_config, - patch("app.library.Notifications.BackgroundWorker") as mock_worker, - ): - mock_config.get_instance.return_value = Mock(debug=False, config_path="/tmp", app_version="1.0.0") - mock_worker_instance = Mock() - mock_worker.get_instance.return_value = mock_worker_instance - - notification = Notification.get_instance() - - # Add a target - request = TargetRequest(type="json", method="POST", url="https://example.com/webhook") - target = Target(id=str(uuid.uuid4()), name="Test Target", request=request) - notification._targets = [target] - - # Create invalid event - event = Event(event="invalid_event", data={"test": "data"}) - - with patch("app.library.Notifications.NotificationEvents.is_valid", return_value=False): - result = notification.emit(event, None) - - assert result is None, "Should return None and not submit to background worker" - mock_worker_instance.submit.assert_not_called() - - def test_emit_valid_event(self): - """Test emit method with valid event.""" - with ( - patch("app.library.Notifications.Config") as mock_config, - patch("app.library.Notifications.BackgroundWorker") as mock_worker, - ): - mock_config.get_instance.return_value = Mock(debug=False, config_path="/tmp", app_version="1.0.0") - mock_worker_instance = Mock() - mock_worker.get_instance.return_value = mock_worker_instance - - notification = Notification.get_instance() - - # Add a target - request = TargetRequest(type="json", method="POST", url="https://example.com/webhook") - target = Target(id=str(uuid.uuid4()), name="Test Target", request=request) - notification._targets = [target] - - # Create valid event - event = Event(event="item_completed", data={"test": "data"}) - - with patch("app.library.Notifications.NotificationEvents.is_valid", return_value=True): - result = notification.emit(event, None) - - assert result is None, "Should return None but submit to background worker" - mock_worker_instance.submit.assert_called_once_with(notification.send, event) - - @pytest.mark.asyncio - async def test_noop(self): - """Test noop method.""" - with ( - patch("app.library.Notifications.Config") as mock_config, - patch("app.library.Notifications.BackgroundWorker") as mock_worker, - ): - mock_config.get_instance.return_value = Mock(debug=False, config_path="/tmp", app_version="1.0.0") - mock_worker.get_instance.return_value = Mock() - - notification = Notification.get_instance() - result = await notification.noop() - - assert result is None diff --git a/ui/app/components/ConditionForm.vue b/ui/app/components/ConditionForm.vue index d05d2c0e..aafda99c 100644 --- a/ui/app/components/ConditionForm.vue +++ b/ui/app/components/ConditionForm.vue @@ -2,7 +2,7 @@
-
+
@@ -21,7 +21,7 @@
-
+
diff --git a/ui/app/components/DLFieldForm.vue b/ui/app/components/DLFieldForm.vue index 7e652a1f..b85397e0 100644 --- a/ui/app/components/DLFieldForm.vue +++ b/ui/app/components/DLFieldForm.vue @@ -2,7 +2,7 @@
-
+
@@ -21,8 +21,8 @@
-
-
+
+
+ +
+
+ + + + The name of the field, it will be shown in the UI. + +
+
+
+
+ + + + A short description of the field, it will be shown in the UI. + +
+
+
-
-
- - - - The name of the field, it will be shown in the UI. - -
-
+
-
-
- - - - A short description of the field, it will be shown in the UI. - -
-
+
diff --git a/ui/app/components/NotificationForm.vue b/ui/app/components/NotificationForm.vue index 475d7c9b..aff43c96 100644 --- a/ui/app/components/NotificationForm.vue +++ b/ui/app/components/NotificationForm.vue @@ -92,7 +92,7 @@
-
+
-
+
-
+
-
+
-
+
+
+ +
+
-
@@ -171,12 +176,6 @@
-
- -
@@ -196,11 +195,6 @@

-
-
- 
{{ JSON.stringify(cleanObject(item, ['raw']), null, 2) }}
-
-