"""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)