212 lines
9.2 KiB
Python
212 lines
9.2 KiB
Python
"""API /api/notifications — центр уведомлений зоотехника."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import unittest
|
|
from datetime import date, timedelta
|
|
|
|
from sqlalchemy import select
|
|
|
|
from app import create_app, db
|
|
from app.models.zootech_notification import ZootechNotification
|
|
from app.services.notification_center_service import (
|
|
RETENTION_DAYS,
|
|
create_notification,
|
|
mark_read,
|
|
purge_expired,
|
|
)
|
|
from app.services.setup_state import mark_setup_complete
|
|
from app.timeutil import display_calendar_today, utc_now_naive
|
|
from tests.helpers.zootech_test_helpers import ZootechTestConfig
|
|
|
|
|
|
class NotificationsApiTests(unittest.TestCase):
|
|
def setUp(self) -> None:
|
|
self.app = create_app(ZootechTestConfig)
|
|
self.client = self.app.test_client()
|
|
self.ctx = self.app.app_context()
|
|
self.ctx.push()
|
|
db.create_all()
|
|
mark_setup_complete(self.app)
|
|
self.client.post(
|
|
"/api/auth/login",
|
|
json={"login": "zootech-test-admin", "password": "zootech-test-secret"},
|
|
)
|
|
|
|
def tearDown(self) -> None:
|
|
db.session.remove()
|
|
db.drop_all()
|
|
self.ctx.pop()
|
|
|
|
def test_create_list_and_read(self) -> None:
|
|
created = self.client.post(
|
|
"/api/notifications",
|
|
json={
|
|
"title": "Сохранено",
|
|
"detail": "Рейс «Утренний» сохранён — 7 июня 2026, 14:32",
|
|
"kind": "success",
|
|
"category": "recipe",
|
|
"page": "recipes",
|
|
},
|
|
)
|
|
self.assertEqual(created.status_code, 201, created.get_data(as_text=True))
|
|
nid = created.get_json()["id"]
|
|
|
|
listed = self.client.get("/api/notifications")
|
|
self.assertEqual(listed.status_code, 200)
|
|
body = listed.get_json()
|
|
self.assertEqual(body["unreadCount"], 1)
|
|
self.assertEqual(body["items"][0]["id"], nid)
|
|
self.assertFalse(body["items"][0]["read"])
|
|
self.assertIn("date", body)
|
|
|
|
with_link = self.client.post(
|
|
"/api/notifications",
|
|
json={
|
|
"title": "Рейс на терминале",
|
|
"detail": "Рейс «Утренний» передан на терминал «Зал» — 7 июня 2026, 14:32",
|
|
"kind": "info",
|
|
"category": "sync",
|
|
"page": "recipes",
|
|
"linkKind": "recipe",
|
|
"linkId": "recipe-uuid-1",
|
|
},
|
|
)
|
|
self.assertEqual(with_link.status_code, 201)
|
|
body2 = self.client.get("/api/notifications").get_json()
|
|
sync_item = next(x for x in body2["items"] if x.get("linkKind") == "recipe")
|
|
self.assertEqual(sync_item["linkId"], "recipe-uuid-1")
|
|
|
|
read = self.client.patch(f"/api/notifications/{nid}/read")
|
|
self.assertEqual(read.status_code, 200)
|
|
self.assertTrue(read.get_json()["read"])
|
|
|
|
listed2 = self.client.get("/api/notifications")
|
|
self.assertEqual(listed2.get_json()["unreadCount"], 1)
|
|
|
|
read_all = self.client.patch("/api/notifications/read-all")
|
|
self.assertEqual(read_all.status_code, 200)
|
|
self.assertEqual(self.client.get("/api/notifications?summary=1").get_json()["unreadCount"], 0)
|
|
|
|
def test_read_all(self) -> None:
|
|
for i in range(3):
|
|
create_notification(title=f"N{i}", detail=f"D{i}", kind="info")
|
|
resp = self.client.patch("/api/notifications/read-all")
|
|
self.assertEqual(resp.status_code, 200)
|
|
self.assertEqual(resp.get_json()["marked"], 3)
|
|
self.assertEqual(self.client.get("/api/notifications?summary=1").get_json()["unreadCount"], 0)
|
|
|
|
def test_purge_older_than_retention(self) -> None:
|
|
old = create_notification(title="Старое", detail="Удалить", kind="info")
|
|
old.created_at = utc_now_naive() - timedelta(days=RETENTION_DAYS + 1)
|
|
db.session.commit()
|
|
create_notification(title="Свежее", detail="Оставить", kind="info")
|
|
purge_expired()
|
|
rows = db.session.execute(select(ZootechNotification)).scalars().all()
|
|
self.assertEqual(len(rows), 1)
|
|
self.assertEqual(rows[0].title, "Свежее")
|
|
|
|
def test_list_by_day_and_category(self) -> None:
|
|
from datetime import datetime, time, timezone
|
|
from zoneinfo import ZoneInfo
|
|
|
|
today = display_calendar_today()
|
|
yesterday = today - timedelta(days=1)
|
|
create_notification(title="Today plan", detail="d", kind="warning", category="daily_plan")
|
|
old = create_notification(title="Yesterday sync", detail="d", kind="info", category="sync")
|
|
tz = ZoneInfo("Europe/Moscow")
|
|
local = datetime.combine(yesterday, time(12, 0), tzinfo=tz)
|
|
old.created_at = local.astimezone(timezone.utc).replace(tzinfo=None)
|
|
db.session.commit()
|
|
|
|
today_resp = self.client.get("/api/notifications")
|
|
self.assertEqual(today_resp.status_code, 200)
|
|
today_body = today_resp.get_json()
|
|
self.assertEqual(len(today_body["items"]), 1)
|
|
self.assertEqual(today_body["items"][0]["title"], "Today plan")
|
|
self.assertTrue(today_body["hasMore"])
|
|
self.assertEqual(today_body["prevDate"], yesterday.isoformat())
|
|
|
|
y_resp = self.client.get(f"/api/notifications?date={yesterday.isoformat()}")
|
|
self.assertEqual(len(y_resp.get_json()["items"]), 1)
|
|
self.assertEqual(y_resp.get_json()["items"][0]["category"], "sync")
|
|
|
|
plan_resp = self.client.get("/api/notifications?category=daily_plan")
|
|
self.assertEqual(len(plan_resp.get_json()["items"]), 1)
|
|
self.assertEqual(plan_resp.get_json()["items"][0]["category"], "daily_plan")
|
|
|
|
def test_category_filter_prev_date_when_today_empty(self) -> None:
|
|
from datetime import datetime, time, timezone
|
|
from zoneinfo import ZoneInfo
|
|
|
|
today = display_calendar_today()
|
|
yesterday = today - timedelta(days=1)
|
|
old = create_notification(title="Plan old", detail="d", kind="info", category="daily_plan")
|
|
tz = ZoneInfo("Europe/Moscow")
|
|
local = datetime.combine(yesterday, time(12, 0), tzinfo=tz)
|
|
old.created_at = local.astimezone(timezone.utc).replace(tzinfo=None)
|
|
create_notification(title="Sync today", detail="d", kind="info", category="sync")
|
|
db.session.commit()
|
|
|
|
resp = self.client.get("/api/notifications?category=daily_plan")
|
|
body = resp.get_json()
|
|
self.assertEqual(len(body["items"]), 0)
|
|
self.assertTrue(body["hasMore"])
|
|
self.assertEqual(body["prevDate"], yesterday.isoformat())
|
|
|
|
prev = self.client.get(
|
|
f"/api/notifications?category=daily_plan&date={yesterday.isoformat()}"
|
|
).get_json()
|
|
self.assertEqual(len(prev["items"]), 1)
|
|
self.assertEqual(prev["items"][0]["title"], "Plan old")
|
|
|
|
def test_legacy_toast_matches_plan_filter(self) -> None:
|
|
create_notification(
|
|
title="Компонент снова в плане",
|
|
detail="Компонент снова в плане — 7 июня 2026, 20:32",
|
|
kind="success",
|
|
category="recipe",
|
|
page="recipes",
|
|
)
|
|
plan_resp = self.client.get("/api/notifications?category=daily_plan").get_json()
|
|
self.assertEqual(len(plan_resp["items"]), 1)
|
|
recipe_resp = self.client.get("/api/notifications?category=recipe").get_json()
|
|
self.assertEqual(len(recipe_resp["items"]), 0)
|
|
|
|
def test_list_sort_unread_first(self) -> None:
|
|
base = utc_now_naive()
|
|
a = create_notification(title="Read first", detail="d", kind="info")
|
|
a.created_at = base
|
|
second = create_notification(title="Unread second", detail="d", kind="error")
|
|
second.created_at = base + timedelta(seconds=1)
|
|
third = create_notification(title="Unread third", detail="d", kind="warning")
|
|
third.created_at = base + timedelta(seconds=2)
|
|
db.session.commit()
|
|
mark_read(a.id)
|
|
resp = self.client.get("/api/notifications?sort=unread_first")
|
|
self.assertEqual(resp.status_code, 200)
|
|
titles = [x["title"] for x in resp.get_json()["items"]]
|
|
self.assertEqual(titles[:2], ["Unread third", "Unread second"])
|
|
|
|
def test_list_sort_severity(self) -> None:
|
|
create_notification(title="Info", detail="d", kind="info")
|
|
create_notification(title="Error", detail="d", kind="error")
|
|
create_notification(title="Warning", detail="d", kind="warning")
|
|
resp = self.client.get("/api/notifications?sort=severity")
|
|
self.assertEqual(resp.status_code, 200)
|
|
kinds = [x["kind"] for x in resp.get_json()["items"][:3]]
|
|
self.assertEqual(kinds, ["error", "warning", "info"])
|
|
|
|
def test_summary_endpoint(self) -> None:
|
|
create_notification(title="A", detail="d", kind="info")
|
|
resp = self.client.get("/api/notifications?summary=1")
|
|
self.assertEqual(resp.status_code, 200)
|
|
body = resp.get_json()
|
|
self.assertEqual(body["unreadCount"], 1)
|
|
self.assertEqual(body["items"], [])
|
|
|
|
def test_requires_auth(self) -> None:
|
|
self.client.post("/api/auth/logout")
|
|
resp = self.client.get("/api/notifications")
|
|
self.assertEqual(resp.status_code, 401)
|