diff --git a/pydatalab/src/pydatalab/config.py b/pydatalab/src/pydatalab/config.py index 8b72dac4d..b62d8d935 100644 --- a/pydatalab/src/pydatalab/config.py +++ b/pydatalab/src/pydatalab/config.py @@ -237,6 +237,11 @@ class ServerConfig(BaseSettings): description="Whether to disable magic-link email authentication while retaining SMTP-backed notification emails.", ) + ENABLE_NOTIFICATIONS: bool = Field( + False, + description="Whether to enable in-app notifications and their API endpoints.", + ) + MAX_CONTENT_LENGTH: int = Field( 10 * 1000**3, description=r"""Direct mapping to the equivalent Flask setting. In practice, limits the file size that can be uploaded. diff --git a/pydatalab/src/pydatalab/feature_flags.py b/pydatalab/src/pydatalab/feature_flags.py index 1f56be2e3..3194ffd7e 100644 --- a/pydatalab/src/pydatalab/feature_flags.py +++ b/pydatalab/src/pydatalab/feature_flags.py @@ -7,7 +7,7 @@ from pydatalab.config import CONFIG from pydatalab.logger import LOGGER -__all__ = ("FEATURE_FLAGS", "check_feature_flags", "FeatureFlags") +__all__ = ("FEATURE_FLAGS", "check_feature_flags", "FeatureFlags", "NotificationFeatures") class AuthMechanisms(BaseModel): @@ -23,10 +23,15 @@ class AIIntegrations(BaseModel): anthropic: bool = False +class NotificationFeatures(BaseModel): + enabled: bool = False + + class FeatureFlags(BaseModel): auth_mechanisms: AuthMechanisms = AuthMechanisms() ai_integrations: AIIntegrations = AIIntegrations() email_notifications: bool = False + notifications: NotificationFeatures = NotificationFeatures() FEATURE_FLAGS: FeatureFlags = FeatureFlags() @@ -59,6 +64,8 @@ def check_feature_flags(app): """ + FEATURE_FLAGS.notifications = NotificationFeatures(enabled=CONFIG.ENABLE_NOTIFICATIONS) + if CONFIG.EMAIL_AUTH_SMTP_SETTINGS is None: LOGGER.warning( "No email auth SMTP settings provided, email registration will not be enabled." diff --git a/pydatalab/src/pydatalab/main.py b/pydatalab/src/pydatalab/main.py index d8ae01bba..ff05c95ed 100644 --- a/pydatalab/src/pydatalab/main.py +++ b/pydatalab/src/pydatalab/main.py @@ -226,6 +226,10 @@ def register_endpoints(app: Flask): versions = ["", f"v{major}", f"v{major}.{minor}", f"v{major}.{minor}.{patch}"] for bp in BLUEPRINTS: + if bp.name == "notifications" and not CONFIG.ENABLE_NOTIFICATIONS: + LOGGER.info("Skipping notification routes because notifications are disabled.") + continue + for ver in versions: app.register_blueprint( bp, url_prefix=f"{CONFIG.ROOT_PATH}{ver}", name=f"{ver}/{bp.name}" diff --git a/pydatalab/src/pydatalab/models/__init__.py b/pydatalab/src/pydatalab/models/__init__.py index 92db5f475..5544368f7 100644 --- a/pydatalab/src/pydatalab/models/__init__.py +++ b/pydatalab/src/pydatalab/models/__init__.py @@ -4,6 +4,11 @@ from pydatalab.models.collections import Collection from pydatalab.models.equipment import Equipment from pydatalab.models.files import File +from pydatalab.models.notifications import ( + Notification, + NotificationGrouping, + NotificationOccurrence, +) from pydatalab.models.people import Person from pydatalab.models.samples import Sample from pydatalab.models.starting_materials import StartingMaterial @@ -25,5 +30,8 @@ "Collection", "Equipment", "ItemVersion", + "Notification", + "NotificationGrouping", + "NotificationOccurrence", "ITEM_MODELS", ) diff --git a/pydatalab/src/pydatalab/models/notifications.py b/pydatalab/src/pydatalab/models/notifications.py new file mode 100644 index 000000000..017953771 --- /dev/null +++ b/pydatalab/src/pydatalab/models/notifications.py @@ -0,0 +1,116 @@ +from datetime import datetime, timezone +from enum import Enum + +from pydantic import BaseModel, Field, root_validator + +from pydatalab.models.entries import Entry +from pydatalab.models.utils import JSON_ENCODERS, PyObjectId + + +class NotificationLevel(str, Enum): + LOW = "low" + NORMAL = "normal" + IMPORTANT = "important" + URGENT = "urgent" + CRITICAL = "critical" + + @property + def priority(self) -> int: + return { + NotificationLevel.LOW: 10, + NotificationLevel.NORMAL: 20, + NotificationLevel.IMPORTANT: 30, + NotificationLevel.URGENT: 40, + NotificationLevel.CRITICAL: 50, + }[self] + + +class NotificationGroupPolicy(str, Enum): + ONCE = "once" + WINDOW = "window" + + +class NotificationGrouping(BaseModel): + """Rules for grouping repeated notification occurrences.""" + + key: str = Field( + ..., + min_length=1, + max_length=500, + description="Stable key used to group repeated notifications from the same thing.", + ) + policy: NotificationGroupPolicy = Field( + NotificationGroupPolicy.WINDOW, + description="How repeated notification occurrences should be grouped.", + ) + window_seconds: int | None = Field( + 86400, + ge=1, + description=( + "Minimum interval before a grouped notification can create a new notification document." + ), + ) + max_occurrences: int = Field( + 100, + ge=1, + description="Maximum number of occurrences before a new notification document is created.", + ) + + @root_validator + def validate_grouping_policy(cls, values): + policy = NotificationGroupPolicy(values.get("policy")) + if policy == NotificationGroupPolicy.WINDOW and values.get("window_seconds") is None: + raise ValueError("window_seconds must be provided for window grouping.") + if policy == NotificationGroupPolicy.ONCE: + values["window_seconds"] = None + return values + + class Config: + json_encoders = JSON_ENCODERS + use_enum_values = True + + +class NotificationOccurrence(BaseModel): + """A single occurrence represented by a grouped notification.""" + + occurred_at: datetime = Field(default_factory=lambda: datetime.now(tz=timezone.utc)) + summary: str | None = Field(None, max_length=1000) + message: str | None = Field(None, max_length=5000) + level: NotificationLevel = Field(NotificationLevel.NORMAL) + is_new: bool = Field( + False, + description="Whether this occurrence arrived since the notification was last read.", + ) + + class Config: + json_encoders = JSON_ENCODERS + use_enum_values = True + + +class Notification(Entry): + """A notification addressed to one user.""" + + type: str = Field("notifications", const=True) + recipient_id: PyObjectId = Field(..., description="ID of the user receiving the notification") + title: str = Field(..., min_length=1, max_length=200) + summary: str | None = Field( + None, + max_length=1000, + description="Short text shown in compact notification lists.", + ) + message: str | None = Field(None, max_length=5000) + level: NotificationLevel = Field(NotificationLevel.NORMAL) + created_at: datetime = Field(default_factory=lambda: datetime.now(tz=timezone.utc)) + created_by: PyObjectId | None = Field(None, description="User ID that created the notification") + read_at: datetime | None = None + archived_at: datetime | None = None + grouping: NotificationGrouping | None = None + occurrence_count: int = Field(1, ge=1) + last_occurred_at: datetime = Field(default_factory=lambda: datetime.now(tz=timezone.utc)) + occurrences: list[NotificationOccurrence] | None = Field( + None, + description="Individual occurrence details for grouped notifications.", + ) + + class Config(Entry.Config): + use_enum_values = True diff --git a/pydatalab/src/pydatalab/mongo.py b/pydatalab/src/pydatalab/mongo.py index b69b51451..a5ca5274a 100644 --- a/pydatalab/src/pydatalab/mongo.py +++ b/pydatalab/src/pydatalab/mongo.py @@ -374,6 +374,38 @@ def create_group_fts(): "refcode", unique=True, name="unique refcode counter", background=background ) + from pydatalab.config import CONFIG + + if CONFIG.ENABLE_NOTIFICATIONS: + ret += db.notifications.create_index( + [("recipient_id", pymongo.ASCENDING), ("created_at", pymongo.DESCENDING)], + name="notification recipient and created", + background=background, + ) + ret += db.notifications.create_index( + [("recipient_id", pymongo.ASCENDING), ("read_at", pymongo.ASCENDING)], + name="notification recipient and read", + background=background, + ) + ret += db.notifications.create_index( + [("recipient_id", pymongo.ASCENDING), ("archived_at", pymongo.ASCENDING)], + name="notification recipient and archived", + background=background, + ) + ret += db.notifications.create_index( + [ + ("recipient_id", pymongo.ASCENDING), + ("grouping.key", pymongo.ASCENDING), + ("grouping.policy", pymongo.ASCENDING), + ("grouping.window_seconds", pymongo.ASCENDING), + ("grouping.max_occurrences", pymongo.ASCENDING), + ("occurrence_count", pymongo.ASCENDING), + ("last_occurred_at", pymongo.DESCENDING), + ], + name="notification grouping lookup", + background=background, + ) + return ret diff --git a/pydatalab/src/pydatalab/notifications.py b/pydatalab/src/pydatalab/notifications.py new file mode 100644 index 000000000..a4e2d44ae --- /dev/null +++ b/pydatalab/src/pydatalab/notifications.py @@ -0,0 +1,163 @@ +from datetime import datetime, timedelta, timezone +from typing import Any + +from bson import ObjectId +from pymongo import ReturnDocument + +from pydatalab.config import CONFIG +from pydatalab.logger import LOGGER +from pydatalab.models.notifications import ( + Notification, + NotificationGrouping, + NotificationGroupPolicy, + NotificationLevel, + NotificationOccurrence, +) +from pydatalab.models.utils import PyObjectId +from pydatalab.mongo import flask_mongo + + +def _max_level( + current_level: NotificationLevel | str | None, new_level: NotificationLevel | str +) -> str: + new_notification_level = NotificationLevel(new_level) + if current_level is None: + return new_notification_level.value + + current_notification_level = NotificationLevel(current_level) + if new_notification_level.priority > current_notification_level.priority: + return new_notification_level.value + + return current_notification_level.value + + +def _find_grouped_notification( + *, + recipient_id: ObjectId, + title: str, + grouping: NotificationGrouping, + now: datetime, + session: Any | None = None, +) -> dict | None: + query: dict[str, object] = { + "recipient_id": recipient_id, + "title": title, + "grouping.key": grouping.key, + "grouping.policy": grouping.policy, + "$expr": {"$lt": ["$occurrence_count", "$grouping.max_occurrences"]}, + "archived_at": None, + } + if NotificationGroupPolicy(grouping.policy) == NotificationGroupPolicy.WINDOW: + query["grouping.window_seconds"] = grouping.window_seconds + query["last_occurred_at"] = { + "$gte": now - timedelta(seconds=int(grouping.window_seconds or 0)) + } + + return flask_mongo.db.notifications.find_one( + query, + sort=[("last_occurred_at", -1), ("created_at", -1)], + session=session, + ) + + +def _insert_notification(notification: Notification, *, session: Any | None = None) -> Notification: + result = flask_mongo.db.notifications.insert_one( + notification.dict(by_alias=True, exclude_none=True), + session=session, + ) + notification.immutable_id = result.inserted_id + return notification + + +def create_notification_with_result( + *, + recipient_id: str | ObjectId | PyObjectId, + title: str, + message: str | None = None, + summary: str | None = None, + level: NotificationLevel | str = NotificationLevel.NORMAL, + created_by: str | ObjectId | PyObjectId | None = None, + grouping: NotificationGrouping | dict[str, object] | None = None, + session: Any | None = None, +) -> tuple[Notification, bool] | None: + """Create or group an in-app notification if the feature is enabled. + + Returns: + A tuple of ``(notification, created)`` when a notification is created or + grouped. ``created`` is ``True`` when a new notification document was + inserted and ``False`` when the notification was folded into an existing + grouped notification. Returns ``None`` when notifications are disabled. + """ + + if not CONFIG.ENABLE_NOTIFICATIONS: + LOGGER.debug("Notifications are disabled; not creating notification %r", title) + return None + + now = datetime.now(tz=timezone.utc) + recipient_object_id = ObjectId(recipient_id) + + if grouping is None: + notification = Notification( + recipient_id=recipient_object_id, + title=title, + summary=summary, + message=message, + level=level, + created_by=ObjectId(created_by) if created_by else None, + occurrence_count=1, + last_occurred_at=now, + ) + return _insert_notification(notification, session=session), True + + if isinstance(grouping, dict): + grouping = NotificationGrouping(**grouping) + + occurrence = NotificationOccurrence( + occurred_at=now, + message=message, + summary=summary, + level=level, + is_new=True, + ) + grouped_notification = _find_grouped_notification( + recipient_id=recipient_object_id, + title=title, + grouping=grouping, + now=now, + session=session, + ) + + if grouped_notification is not None: + mongo_update = { + "$inc": {"occurrence_count": 1}, + "$set": { + "message": message, + "summary": summary, + "last_occurred_at": now, + "level": _max_level(grouped_notification.get("level"), level), + }, + "$push": {"occurrences": occurrence.dict(exclude_none=True)}, + "$unset": {"read_at": ""}, + } + updated_notification = flask_mongo.db.notifications.find_one_and_update( + {"_id": grouped_notification["_id"]}, + mongo_update, + return_document=ReturnDocument.AFTER, + session=session, + ) + return Notification(**updated_notification), False + + notification = Notification( + recipient_id=recipient_object_id, + title=title, + summary=summary, + message=message, + level=level, + created_by=ObjectId(created_by) if created_by else None, + grouping=grouping, + occurrence_count=1, + last_occurred_at=now, + occurrences=[occurrence], + ) + + return _insert_notification(notification, session=session), True diff --git a/pydatalab/src/pydatalab/permissions.py b/pydatalab/src/pydatalab/permissions.py index 2ae17a0ff..cad4dedf2 100644 --- a/pydatalab/src/pydatalab/permissions.py +++ b/pydatalab/src/pydatalab/permissions.py @@ -113,6 +113,44 @@ def wrapped_route(*args, **kwargs): return wrapped_route +def with_notification_permissions(func): + """Decorator to inject the notification permission filter for the current user.""" + + @wraps(func) + def wrapped_route(*args, **kwargs): + if current_user.is_authenticated and current_user.person is not None: + kwargs["notification_permissions"] = {"recipient_id": current_user.person.immutable_id} + else: + kwargs["notification_permissions"] = {"_id": -1} + return func(*args, **kwargs) + + return wrapped_route + + +def notification_recipient_only(func): + """Decorator to load a notification only if it belongs to the current user.""" + + @wraps(func) + def wrapped_route(*args, **kwargs): + # notifications_permissions are injected by the with_notification_permissions decorator(see below) + notification_permissions = kwargs["notification_permissions"] + try: + notification_object_id = ObjectId(kwargs["notification_id"]) + except Exception: + return {"error": f"Invalid notification_id {kwargs.get('notification_id')!r}."}, 400 + + notification = flask_mongo.db.notifications.find_one( + {"_id": notification_object_id, **notification_permissions} + ) + if notification is None: + return {"error": "Notification not found."}, 404 + + kwargs["notification"] = notification + return func(*args, **kwargs) + + return with_notification_permissions(wrapped_route) + + def check_access_token(refcode: str, token: str | None = None) -> bool: """Check whether the provided access token exists in the get_database and corresponds to the relevant refcode. diff --git a/pydatalab/src/pydatalab/routes/v0_1/__init__.py b/pydatalab/src/pydatalab/routes/v0_1/__init__.py index 93b3a22db..6b2e53269 100644 --- a/pydatalab/src/pydatalab/routes/v0_1/__init__.py +++ b/pydatalab/src/pydatalab/routes/v0_1/__init__.py @@ -12,6 +12,7 @@ from .healthcheck import HEALTHCHECK from .info import INFO from .items import ITEMS +from .notifications import NOTIFICATIONS from .remotes import REMOTES from .users import USERS @@ -27,6 +28,7 @@ FILES, HEALTHCHECK, INFO, + NOTIFICATIONS, GRAPHS, EXPORT, ) diff --git a/pydatalab/src/pydatalab/routes/v0_1/notifications.py b/pydatalab/src/pydatalab/routes/v0_1/notifications.py new file mode 100644 index 000000000..7874dc197 --- /dev/null +++ b/pydatalab/src/pydatalab/routes/v0_1/notifications.py @@ -0,0 +1,305 @@ +from datetime import datetime, timezone +from typing import Any + +from bson import ObjectId +from flask import Blueprint, abort, jsonify, request +from flask_login import current_user +from pydantic import ValidationError +from pymongo import ReturnDocument +from werkzeug.exceptions import BadRequest, NotFound + +from pydatalab.feature_flags import FEATURE_FLAGS +from pydatalab.models.notifications import Notification +from pydatalab.models.people import AccountStatus +from pydatalab.mongo import flask_mongo +from pydatalab.notifications import create_notification_with_result +from pydatalab.permissions import ( + active_users_or_get_only, + admin_only, + notification_recipient_only, + with_notification_permissions, +) + +NOTIFICATIONS = Blueprint("notifications", __name__) + + +@NOTIFICATIONS.before_request +def _require_notifications_feature(): + """Gate the whole blueprint behind the `notifications` feature flag.""" + if not FEATURE_FLAGS.notifications.enabled: + abort(404) + + +@NOTIFICATIONS.before_request +@active_users_or_get_only +def _(): ... + + +def _count_unread_notifications(notification_permissions: dict) -> int: + return flask_mongo.db.notifications.count_documents( + {**notification_permissions, "archived_at": None, "read_at": None} + ) + + +@NOTIFICATIONS.route("/notifications", methods=["POST"]) +@admin_only +def create_notifications(): + request_json = request.get_json() or {} + title = str(request_json.get("title", "")).strip() + message = ( + str(request_json["message"]).strip() if request_json.get("message") is not None else None + ) + summary = ( + str(request_json["summary"]).strip() if request_json.get("summary") is not None else None + ) + + send_all_users = request_json.get("send_all_users", False) is True + if send_all_users: + recipient_ids = [ + user["_id"] + for user in flask_mongo.db.users.find( + {"account_status": AccountStatus.ACTIVE.value}, + {"_id": 1}, + ) + ] + else: + recipient_ids = [] + missing_recipient_ids = [] + raw_recipient_ids = request_json.get("recipient_ids") + if raw_recipient_ids is not None: + if not isinstance(raw_recipient_ids, list): + raise BadRequest("recipient_ids must be a list.") + for recipient_id in raw_recipient_ids: + try: + recipient_object_id = ObjectId(recipient_id) + except Exception as exc: + raise BadRequest(f"Invalid recipient_id {recipient_id!r}.") from exc + + if flask_mongo.db.users.find_one( + {"_id": recipient_object_id}, + {"_id": 1}, + ): + recipient_ids.append(recipient_object_id) + else: + missing_recipient_ids.append(str(recipient_object_id)) + if missing_recipient_ids: + raise NotFound(f"Recipient user(s) not found: {', '.join(missing_recipient_ids)}") + + # deduplicate recipient ids. uses a dict here to preserve order. + recipient_ids = list(dict.fromkeys(recipient_ids)) + if not recipient_ids: + raise BadRequest("At least one recipient or all-users option must be provided.") + + grouping = request_json.get("grouping") + if grouping is not None and grouping != {}: + if not isinstance(grouping, dict): + raise BadRequest("grouping must be a dict.") + grouping = dict(grouping) + else: + grouping = None + + level = request_json.get("level", "normal") + created_by = current_user.person.immutable_id + + def create_for_recipients(session=None): + notification_results = [] + for recipient_id in recipient_ids: + try: + notification_result = create_notification_with_result( + recipient_id=recipient_id, + title=title, + summary=summary, + message=message, + level=level, + created_by=created_by, + grouping=grouping, + session=session, + ) + except (ValidationError, ValueError) as exc: + raise BadRequest(str(exc)) from exc + + if notification_result is not None: + notification_results.append(notification_result) + + return notification_results + + try: + hello = flask_mongo.cx.admin.command("hello") + supports_transactions = hello.get("msg") == "isdbgrid" or "setName" in hello + except Exception: + supports_transactions = False + + if supports_transactions: + with flask_mongo.cx.start_session() as session: + with session.start_transaction(): + notification_results = create_for_recipients(session=session) + else: + notification_results = create_for_recipients() + + created_count = sum(created for _, created in notification_results) + grouped_count = len(notification_results) - created_count + status_code = 201 if created_count else 200 + + return jsonify( + { + "status": "success", + "data": [notification.dict() for notification, _ in notification_results], + "notification_ids": [ + str(notification.immutable_id) for notification, _ in notification_results + ], + "created_count": created_count, + "grouped_count": grouped_count, + } + ), status_code + + +@NOTIFICATIONS.route("/notifications", methods=["GET"]) +@with_notification_permissions +def list_notifications(notification_permissions: dict): + include_archived = request.args.get("include_archived") == "1" + unread_only = request.args.get("unread_only") == "1" + limit = request.args.get("limit", default=50, type=int) + limit = max(1, limit) + + query = dict(notification_permissions) + if not include_archived: + query["archived_at"] = None + if unread_only: + query["read_at"] = None + + notifications = flask_mongo.db.notifications.aggregate( + [ + {"$match": query}, + { + "$addFields": { + "_is_read": {"$ne": ["$read_at", None]}, + "_notification_time": {"$ifNull": ["$last_occurred_at", "$created_at"]}, + } + }, + {"$sort": {"_is_read": 1, "_notification_time": -1, "created_at": -1}}, + {"$limit": limit}, + {"$project": {"_is_read": 0, "_notification_time": 0}}, + ] + ) + return jsonify( + { + "status": "success", + "data": [Notification(**notification).dict() for notification in notifications], + "unread_count": _count_unread_notifications(notification_permissions), + } + ), 200 + + +@NOTIFICATIONS.route("/notifications/unread-count", methods=["GET"]) +@with_notification_permissions +def get_notification_unread_count(notification_permissions: dict): + return jsonify( + { + "status": "success", + "unread_count": _count_unread_notifications(notification_permissions), + } + ), 200 + + +@NOTIFICATIONS.route("/notifications/", methods=["PATCH"]) +@notification_recipient_only +def update_notification( + notification_id: str, + notification: dict, + notification_permissions: dict, +): + notification_query = {"_id": notification["_id"], **notification_permissions} + request_json = request.get_json() or {} + update = {} + unset = {} + now = datetime.now(tz=timezone.utc) + + if "read" in request_json: + if request_json["read"] is True: + update["read_at"] = notification.get("read_at") or now + if notification.get("occurrences"): + update["occurrences.$[].is_new"] = False + else: + unset["read_at"] = "" + + if "archived" in request_json: + if request_json["archived"] is True: + update["archived_at"] = notification.get("archived_at") or now + else: + unset["archived_at"] = "" + + if not update and not unset: + return jsonify( + { + "status": "success", + "data": Notification(**notification).dict(), + "message": "No update to perform.", + } + ), 200 + + mongo_update: dict[str, Any] = {} + if update: + mongo_update["$set"] = update + if unset: + mongo_update["$unset"] = unset + + updated_notification = flask_mongo.db.notifications.find_one_and_update( + notification_query, + mongo_update, + return_document=ReturnDocument.AFTER, + ) + if updated_notification is None: + raise NotFound("Notification not found.") + + return jsonify( + { + "status": "success", + "data": Notification(**updated_notification).dict(), + "unread_count": _count_unread_notifications(notification_permissions), + } + ), 200 + + +@NOTIFICATIONS.route("/notifications/mark-all-read", methods=["POST"]) +@with_notification_permissions +def mark_all_notifications_read(notification_permissions: dict): + now = datetime.now(tz=timezone.utc) + unread_query = {**notification_permissions, "archived_at": None, "read_at": None} + grouped_result = flask_mongo.db.notifications.update_many( + {**unread_query, "occurrences.0": {"$exists": True}}, + {"$set": {"read_at": now, "occurrences.$[].is_new": False}}, + ) + non_grouped_result = flask_mongo.db.notifications.update_many( + {**unread_query, "occurrences.0": {"$exists": False}}, + {"$set": {"read_at": now}}, + ) + + return jsonify( + { + "status": "success", + "modified_count": grouped_result.modified_count + non_grouped_result.modified_count, + "unread_count": 0, + } + ), 200 + + +@NOTIFICATIONS.route("/notifications/", methods=["DELETE"]) +@notification_recipient_only +def delete_notification( + notification_id: str, + notification: dict, + notification_permissions: dict, +): + result = flask_mongo.db.notifications.delete_one( + {"_id": notification["_id"], **notification_permissions} + ) + if result.deleted_count == 0: + raise NotFound("Notification not found.") + + return jsonify( + { + "status": "success", + "deleted_count": result.deleted_count, + "unread_count": _count_unread_notifications(notification_permissions), + } + ), 200 diff --git a/pydatalab/tests/server/conftest.py b/pydatalab/tests/server/conftest.py index 4cc1c7938..c7242cb55 100644 --- a/pydatalab/tests/server/conftest.py +++ b/pydatalab/tests/server/conftest.py @@ -73,6 +73,7 @@ def app_config(secret_key, files_directory): "MAIL_DEBUG": True, "MAIL_SUPPRESS_SEND": True, "MAIL_PASSWORD": "test", + "ENABLE_NOTIFICATIONS": True, # Set to 10 MB to check that larger files fail; this should be larger than all of our example files. # Elsewhere, we can generate an artificial large file to check that it fails. "MAX_CONTENT_LENGTH": 10 * 1000**2, diff --git a/pydatalab/tests/server/test_info_and_health.py b/pydatalab/tests/server/test_info_and_health.py index 614740aab..69ae568a2 100644 --- a/pydatalab/tests/server/test_info_and_health.py +++ b/pydatalab/tests/server/test_info_and_health.py @@ -19,6 +19,7 @@ def test_info_endpoint(client, url_prefix, app): and app.config.get("ORCID_OAUTH_CLIENT_SECRET", None) ) assert auth["email"] is bool(app.config.get("MAIL_PASSWORD", None)) + assert features["notifications"] == {"enabled": True} def test_magic_link_auth_feature_flag_can_be_disabled(app, monkeypatch): diff --git a/pydatalab/tests/server/test_notifications.py b/pydatalab/tests/server/test_notifications.py new file mode 100644 index 000000000..e6642eba9 --- /dev/null +++ b/pydatalab/tests/server/test_notifications.py @@ -0,0 +1,544 @@ +from datetime import datetime, timedelta, timezone + +import pytest +from bson import ObjectId + +from pydatalab.models.people import AccountStatus + + +@pytest.fixture(autouse=True) +def clean_notifications(database): + database.notifications.delete_many({}) + + +def _create_notification(client, recipient_id, **overrides): + payload = { + "recipient_ids": [str(recipient_id)], + "title": "Ingestion warning", + "message": "A generated notification for testing.", + "level": "important", + **overrides, + } + return client.post("/notifications", json=payload) + + +def test_create_and_list( + admin_client, client, another_client, user_id, another_user_id, admin_user_id +): + resp = admin_client.post( + "/notifications", + json={ + "recipient_ids": [str(user_id), str(another_user_id)], + "title": "Batch import warning", + "summary": "Two rows failed.", + "message": "Two rows could not be matched.", + "level": "important", + }, + ) + + assert resp.status_code == 201 + assert len(resp.json["data"]) == 2 + assert resp.json["created_count"] == 2 + assert resp.json["grouped_count"] == 0 + + resp = client.get("/notifications") + + assert resp.status_code == 200 + assert resp.json["unread_count"] == 1 + assert len(resp.json["data"]) == 1 + notification = resp.json["data"][0] + assert notification["recipient_id"] == str(user_id) + assert notification["created_by"] == str(admin_user_id) + assert notification["title"] == "Batch import warning" + assert notification["summary"] == "Two rows failed." + assert notification["message"] == "Two rows could not be matched." + assert notification["level"] == "important" + assert notification["read_at"] is None + assert notification["archived_at"] is None + assert notification["occurrence_count"] == 1 + assert notification.get("occurrences") is None + + resp = admin_client.get("/notifications") + assert resp.status_code == 200 + assert resp.json["unread_count"] == 0 + assert resp.json["data"] == [] + + resp = another_client.get("/notifications") + assert resp.status_code == 200 + assert resp.json["unread_count"] == 1 + assert resp.json["data"][0]["recipient_id"] == str(another_user_id) + + +def test_send_all_users_precedence(admin_client, database): + active_user_ids = { + str(user["_id"]) + for user in database.users.find( + {"account_status": AccountStatus.ACTIVE.value}, + {"_id": 1}, + ) + } + + resp = admin_client.post( + "/notifications", + json={ + "recipient_ids": ["not-an-object-id"], + "send_all_users": True, + "title": "Scheduled maintenance", + }, + ) + + assert resp.status_code == 201 + assert resp.json["created_count"] == len(active_user_ids) + assert resp.json["grouped_count"] == 0 + assert {notification["recipient_id"] for notification in resp.json["data"]} == active_user_ids + assert database.notifications.count_documents({}) == len(active_user_ids) + + +@pytest.mark.parametrize( + ("payload", "message"), + [ + ( + {"recipient_ids": [], "title": "No recipients"}, + "At least one recipient or all-users option must be provided.", + ), + ( + {"recipient_ids": "not-a-list", "title": "Recipient IDs must be a list"}, + "recipient_ids must be a list.", + ), + ( + {"recipient_ids": ["not-an-object-id"], "title": "Malformed recipient ID"}, + "Invalid recipient_id 'not-an-object-id'.", + ), + ], +) +def test_create_validation_errors(admin_client, database, payload, message): + resp = admin_client.post("/notifications", json=payload) + + assert resp.status_code == 400 + assert resp.json["message"] == message + assert database.notifications.count_documents({}) == 0 + + +def test_create_missing_recipient(admin_client, database): + missing_id = str(ObjectId()) + + resp = admin_client.post( + "/notifications", + json={"recipient_ids": [missing_id], "title": "Missing recipient"}, + ) + + assert resp.status_code == 404 + assert resp.json["message"] == f"Recipient user(s) not found: {missing_id}" + assert database.notifications.count_documents({}) == 0 + + +def test_create_requires_admin(client, user_id, database): + resp = _create_notification(client, user_id) + + assert resp.status_code == 403 + assert database.notifications.count_documents({}) == 0 + + +def test_read_archive_filters(admin_client, client, another_client, user_id): + create_resp = _create_notification(admin_client, user_id) + assert create_resp.status_code == 201 + notification_id = create_resp.json["notification_ids"][0] + + resp = another_client.patch(f"/notifications/{notification_id}", json={"read": True}) + assert resp.status_code == 404 + assert resp.json["error"] == "Notification not found." + + before = datetime.now(tz=timezone.utc) + resp = client.patch(f"/notifications/{notification_id}", json={"read": True}) + after = datetime.now(tz=timezone.utc) + + assert resp.status_code == 200 + read_at = datetime.fromisoformat(resp.json["data"]["read_at"]) + if read_at.tzinfo is None: + read_at = read_at.replace(tzinfo=timezone.utc) + assert before <= read_at <= after + assert resp.json["unread_count"] == 0 + + resp = client.patch(f"/notifications/{notification_id}", json={"read": False}) + assert resp.status_code == 200 + assert resp.json["data"]["read_at"] is None + assert resp.json["unread_count"] == 1 + + before = datetime.now(tz=timezone.utc) + resp = client.patch(f"/notifications/{notification_id}", json={"archived": True}) + after = datetime.now(tz=timezone.utc) + + assert resp.status_code == 200 + archived_at = datetime.fromisoformat(resp.json["data"]["archived_at"]) + if archived_at.tzinfo is None: + archived_at = archived_at.replace(tzinfo=timezone.utc) + assert before <= archived_at <= after + assert resp.json["unread_count"] == 0 + + resp = client.get("/notifications") + assert resp.status_code == 200 + assert resp.json["data"] == [] + + resp = client.get("/notifications?include_archived=1") + assert resp.status_code == 200 + assert len(resp.json["data"]) == 1 + + +def test_delete_own_only(admin_client, client, another_client, database, user_id): + create_resp = _create_notification(admin_client, user_id) + assert create_resp.status_code == 201 + notification_id = create_resp.json["notification_ids"][0] + + resp = another_client.delete(f"/notifications/{notification_id}") + assert resp.status_code == 404 + assert resp.json["error"] == "Notification not found." + + resp = client.delete(f"/notifications/{notification_id}") + + assert resp.status_code == 200 + assert resp.json["deleted_count"] == 1 + assert resp.json["unread_count"] == 0 + assert database.notifications.count_documents({"recipient_id": user_id}) == 0 + + +def test_mark_all_read(admin_client, client, another_client, user_id, another_user_id): + first_resp = _create_notification(admin_client, user_id, title="First") + second_resp = _create_notification(admin_client, user_id, title="Second") + other_user_resp = _create_notification(admin_client, another_user_id, title="Other user") + assert first_resp.status_code == 201 + assert second_resp.status_code == 201 + assert other_user_resp.status_code == 201 + grouped_resp = _create_notification( + admin_client, + user_id, + title="Grouped mark all", + grouping={"key": "ingestion:mark-all", "policy": "once"}, + ) + assert grouped_resp.status_code == 201 + grouped_resp = _create_notification( + admin_client, + user_id, + title="Grouped mark all", + grouping={"key": "ingestion:mark-all", "policy": "once"}, + ) + assert grouped_resp.status_code == 200 + assert [occurrence["is_new"] for occurrence in grouped_resp.json["data"][0]["occurrences"]] == [ + True, + True, + ] + + resp = client.get("/notifications/unread-count") + assert resp.status_code == 200 + assert resp.json["unread_count"] == 3 + + resp = client.post("/notifications/mark-all-read") + + assert resp.status_code == 200 + assert resp.json["modified_count"] == 3 + assert resp.json["unread_count"] == 0 + + resp = client.get("/notifications") + grouped_notification = next( + notification + for notification in resp.json["data"] + if notification["title"] == "Grouped mark all" + ) + assert [occurrence["is_new"] for occurrence in grouped_notification["occurrences"]] == [ + False, + False, + ] + + resp = client.get("/notifications?unread_only=1") + assert resp.status_code == 200 + assert resp.json["data"] == [] + + resp = another_client.get("/notifications/unread-count") + assert resp.status_code == 200 + assert resp.json["unread_count"] == 1 + + +def test_list_unread_first(admin_client, client, user_id): + read_resp = _create_notification(admin_client, user_id, title="Read first") + assert read_resp.status_code == 201 + read_notification_id = read_resp.json["notification_ids"][0] + client.patch(f"/notifications/{read_notification_id}", json={"read": True}) + unread_resp = _create_notification(admin_client, user_id, title="Unread second") + assert unread_resp.status_code == 201 + + resp = client.get("/notifications") + + assert resp.status_code == 200 + assert [notification["title"] for notification in resp.json["data"]] == [ + "Unread second", + "Read first", + ] + + +def test_grouping_default_window(admin_client, client, user_id): + first_resp = _create_notification( + admin_client, + user_id, + title="Default grouping", + grouping={"key": "ingestion:default-window"}, + ) + assert first_resp.status_code == 201 + assert first_resp.json["created_count"] == 1 + assert first_resp.json["grouped_count"] == 0 + + second_resp = _create_notification( + admin_client, + user_id, + title="Default grouping", + grouping={"key": "ingestion:default-window"}, + ) + assert second_resp.status_code == 200 + assert second_resp.json["created_count"] == 0 + assert second_resp.json["grouped_count"] == 1 + + resp = client.get("/notifications") + notification = resp.json["data"][0] + assert notification["grouping"] == { + "key": "ingestion:default-window", + "policy": "window", + "window_seconds": 86400, + "max_occurrences": 100, + } + assert [occurrence["is_new"] for occurrence in notification["occurrences"]] == [True, True] + + +def test_grouping_once_updates_existing(admin_client, client, user_id): + first_resp = _create_notification( + admin_client, + user_id, + title="Equipment anomaly", + summary="Equipment drift detected in run 1.", + message="The first ingestion run detected drift.", + level="low", + grouping={"key": "ingestion:equipment:eq-1", "policy": "once"}, + ) + assert first_resp.status_code == 201 + notification_id = first_resp.json["notification_ids"][0] + + resp = client.get("/notifications") + assert len(resp.json["data"]) == 1 + assert resp.json["data"][0]["read_at"] is None + assert [occurrence["is_new"] for occurrence in resp.json["data"][0]["occurrences"]] == [True] + assert resp.json["unread_count"] == 1 + + resp = client.patch(f"/notifications/{notification_id}", json={"read": True}) + assert resp.status_code == 200 + assert resp.json["data"]["read_at"] is not None + assert [occurrence["is_new"] for occurrence in resp.json["data"]["occurrences"]] == [False] + + resp = client.get("/notifications") + assert resp.json["unread_count"] == 0 + assert [occurrence["is_new"] for occurrence in resp.json["data"][0]["occurrences"]] == [False] + + second_resp = _create_notification( + admin_client, + user_id, + title="Equipment anomaly", + summary="Equipment drift detected in run 2.", + message="The second ingestion run detected drift.", + level="critical", + grouping={"key": "ingestion:equipment:eq-1", "policy": "once"}, + ) + third_resp = _create_notification( + admin_client, + user_id, + title="Equipment anomaly", + summary="Equipment drift detected in run 3.", + message="The third ingestion run detected drift.", + level="normal", + grouping={"key": "ingestion:equipment:eq-1", "policy": "once"}, + ) + + assert second_resp.status_code == 200 + assert third_resp.status_code == 200 + assert third_resp.json["notification_ids"] == [notification_id] + + resp = client.get("/notifications") + + assert resp.status_code == 200 + assert resp.json["unread_count"] == 1 + assert len(resp.json["data"]) == 1 + notification = resp.json["data"][0] + assert notification["summary"] == "Equipment drift detected in run 3." + assert notification["message"] == "The third ingestion run detected drift." + assert notification["level"] == "critical" + assert notification["read_at"] is None + assert notification["occurrence_count"] == 3 + assert [occurrence["level"] for occurrence in notification["occurrences"]] == [ + "low", + "critical", + "normal", + ] + assert [occurrence["is_new"] for occurrence in notification["occurrences"]] == [ + False, + True, + True, + ] + assert notification["occurrences"][0]["message"] == "The first ingestion run detected drift." + assert notification["occurrences"][2]["message"] == "The third ingestion run detected drift." + + +def test_grouping_archived_creates_new_document(admin_client, client, database, user_id): + resp = _create_notification( + admin_client, + user_id, + grouping={"key": "ingestion:archived", "policy": "once"}, + ) + assert resp.status_code == 201 + notification_id = resp.json["notification_ids"][0] + client.patch(f"/notifications/{notification_id}", json={"archived": True}) + + resp = _create_notification( + admin_client, + user_id, + grouping={"key": "ingestion:archived", "policy": "once"}, + ) + + assert resp.status_code == 201 + second_notification_id = resp.json["notification_ids"][0] + assert resp.json["notification_ids"] != [notification_id] + assert database.notifications.count_documents({"recipient_id": user_id}) == 2 + + resp = client.get("/notifications?include_archived=1") + assert len(resp.json["data"]) == 2 + + resp = client.get("/notifications") + assert len(resp.json["data"]) == 1 + assert resp.json["data"][0]["immutable_id"] == second_notification_id + + +def test_grouping_title_separates_documents(admin_client, database, user_id): + first_resp = _create_notification( + admin_client, + user_id, + title="First title", + grouping={"key": "ingestion:title", "policy": "once"}, + ) + second_resp = _create_notification( + admin_client, + user_id, + title="Second title", + grouping={"key": "ingestion:title", "policy": "once"}, + ) + + assert first_resp.status_code == 201 + assert second_resp.status_code == 201 + assert database.notifications.count_documents({"recipient_id": user_id}) == 2 + + +def test_grouping_window_expires(admin_client, database, user_id): + first_resp = _create_notification( + admin_client, + user_id, + grouping={"key": "ingestion:pipeline:daily", "policy": "window", "window_seconds": 60}, + ) + assert first_resp.status_code == 201 + first_notification_id = first_resp.json["notification_ids"][0] + + second_resp = _create_notification( + admin_client, + user_id, + grouping={"key": "ingestion:pipeline:daily", "policy": "window", "window_seconds": 60}, + ) + assert second_resp.status_code == 200 + assert second_resp.json["notification_ids"] == [first_notification_id] + + # Manually update the last_occurred_at to be more than 60 seconds ago so that the next notification creates a new document. + database.notifications.update_one( + {"_id": ObjectId(first_notification_id)}, + {"$set": {"last_occurred_at": datetime.now(tz=timezone.utc) - timedelta(seconds=120)}}, + ) + + third_resp = _create_notification( + admin_client, + user_id, + grouping={"key": "ingestion:pipeline:daily", "policy": "window", "window_seconds": 60}, + ) + + assert third_resp.status_code == 201 + assert third_resp.json["created_count"] == 1 + assert database.notifications.count_documents({"recipient_id": user_id}) == 2 + + +def test_grouping_once_uses_stored_cap(admin_client, database, user_id): + first_resp = _create_notification( + admin_client, + user_id, + grouping={"key": "ingestion:max-once", "policy": "once", "max_occurrences": 2}, + ) + assert first_resp.status_code == 201 + first_notification_id = first_resp.json["notification_ids"][0] + + second_resp = _create_notification( + admin_client, + user_id, + grouping={"key": "ingestion:max-once", "policy": "once", "max_occurrences": 100}, + ) + assert second_resp.status_code == 200 + assert second_resp.json["notification_ids"] == [first_notification_id] + + first_notification = database.notifications.find_one({"_id": ObjectId(first_notification_id)}) + assert first_notification["occurrence_count"] == 2 + assert first_notification["grouping"]["max_occurrences"] == 2 + + third_resp = _create_notification( + admin_client, + user_id, + grouping={"key": "ingestion:max-once", "policy": "once", "max_occurrences": 100}, + ) + assert third_resp.status_code == 201 + assert third_resp.json["notification_ids"] != [first_notification_id] + assert third_resp.json["data"][0]["grouping"]["max_occurrences"] == 100 + assert database.notifications.count_documents({"recipient_id": user_id}) == 2 + + +def test_grouping_window_uses_stored_cap(admin_client, database, user_id): + first_resp = _create_notification( + admin_client, + user_id, + grouping={ + "key": "ingestion:pipeline:max-window", + "policy": "window", + "window_seconds": 3600, + "max_occurrences": 2, + }, + ) + assert first_resp.status_code == 201 + first_notification_id = first_resp.json["notification_ids"][0] + + second_resp = _create_notification( + admin_client, + user_id, + grouping={ + "key": "ingestion:pipeline:max-window", + "policy": "window", + "window_seconds": 3600, + "max_occurrences": 100, + }, + ) + assert second_resp.status_code == 200 + assert second_resp.json["notification_ids"] == [first_notification_id] + + first_notification = database.notifications.find_one({"_id": ObjectId(first_notification_id)}) + assert first_notification["occurrence_count"] == 2 + assert first_notification["grouping"]["max_occurrences"] == 2 + + third_resp = _create_notification( + admin_client, + user_id, + grouping={ + "key": "ingestion:pipeline:max-window", + "policy": "window", + "window_seconds": 3600, + "max_occurrences": 100, + }, + ) + + assert third_resp.status_code == 201 + assert third_resp.json["notification_ids"] != [first_notification_id] + assert third_resp.json["data"][0]["grouping"]["max_occurrences"] == 100 + assert database.notifications.count_documents({"recipient_id": user_id}) == 2 diff --git a/pydatalab/tests/test_config.py b/pydatalab/tests/test_config.py index 0637c5794..b29d13201 100644 --- a/pydatalab/tests/test_config.py +++ b/pydatalab/tests/test_config.py @@ -10,6 +10,18 @@ def test_default_settings(): assert config.MONGO_URI == "mongodb://localhost:27017/datalabvue" assert config.SECRET_KEY assert Path(config.FILE_DIRECTORY).name == "files" + assert config.ENABLE_NOTIFICATIONS is False + + +def test_notification_feature_flags(monkeypatch): + from pydatalab.config import CONFIG + from pydatalab.feature_flags import NotificationFeatures + + monkeypatch.setattr(CONFIG, "ENABLE_NOTIFICATIONS", False) + assert NotificationFeatures(enabled=CONFIG.ENABLE_NOTIFICATIONS).dict() == {"enabled": False} + + monkeypatch.setattr(CONFIG, "ENABLE_NOTIFICATIONS", True) + assert NotificationFeatures(enabled=CONFIG.ENABLE_NOTIFICATIONS).dict() == {"enabled": True} def test_update_settings(): diff --git a/pydatalab/tests/test_notifications.py b/pydatalab/tests/test_notifications.py new file mode 100644 index 000000000..fdc18a974 --- /dev/null +++ b/pydatalab/tests/test_notifications.py @@ -0,0 +1,31 @@ +from flask import Flask + + +def test_notification_level_ordering(): + from pydatalab.models.notifications import NotificationLevel + + assert ( + NotificationLevel.LOW.priority + < NotificationLevel.NORMAL.priority + < NotificationLevel.IMPORTANT.priority + < NotificationLevel.URGENT.priority + < NotificationLevel.CRITICAL.priority + ) + + +def test_notification_occurrence_new_state_defaults_false(): + from pydatalab.models.notifications import NotificationOccurrence + + assert NotificationOccurrence().is_new is False + + +def test_notifications_routes_not_registered_when_disabled(monkeypatch): + from pydatalab.config import CONFIG + from pydatalab.main import register_endpoints + + monkeypatch.setattr(CONFIG, "ENABLE_NOTIFICATIONS", False) + + app = Flask("notifications-disabled") + register_endpoints(app) + + assert not any("/notifications" in str(rule) for rule in app.url_map.iter_rules())