156 lines
5.2 KiB
Python
156 lines
5.2 KiB
Python
"""Уведомления при синхронизации рейсов и отчётов."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import unittest
|
|
from datetime import datetime
|
|
|
|
from sqlalchemy import select
|
|
|
|
from app import create_app, db
|
|
from app.models import Recipe, SyncClient, SyncQueue
|
|
from app.models.report import LoadingReport
|
|
from app.models.zootech_notification import ZootechNotification
|
|
from app.services.notification_center_service import create_notification
|
|
from app.services.setup_state import mark_setup_complete
|
|
from app.services.sync_manager import SyncManager
|
|
from tests.helpers.zootech_test_helpers import ZootechTestConfig
|
|
|
|
|
|
class SyncNotificationHooksTests(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 _register_sync_client(self, node_id: str, name: str) -> SyncClient:
|
|
row = SyncClient(
|
|
node_id=node_id,
|
|
client_name=name,
|
|
status="active",
|
|
is_enabled=True,
|
|
created_at=datetime.utcnow(),
|
|
updated_at=datetime.utcnow(),
|
|
)
|
|
db.session.add(row)
|
|
db.session.commit()
|
|
return row
|
|
|
|
def test_push_report_creates_notification_with_link(self) -> None:
|
|
self._register_sync_client("node-field-1", "Зал 2")
|
|
recipe = Recipe(
|
|
name="Утренний",
|
|
created_by="system",
|
|
updated_by="system",
|
|
created_at=datetime.utcnow(),
|
|
updated_at=datetime.utcnow(),
|
|
)
|
|
db.session.add(recipe)
|
|
db.session.commit()
|
|
|
|
import uuid
|
|
|
|
report_id = str(uuid.uuid4())
|
|
changes = [
|
|
{
|
|
"table_name": "loading_report",
|
|
"record_id": report_id,
|
|
"action": "create",
|
|
"data": {
|
|
"id": report_id,
|
|
"recipe_id": recipe.id,
|
|
"recipe_name": recipe.name,
|
|
"start_time": "2026-06-07T10:00:00",
|
|
"dispenser_type": "dispenser",
|
|
"version": 1,
|
|
"created_by": "terminal",
|
|
"updated_by": "terminal",
|
|
"content_hash": "abc",
|
|
},
|
|
}
|
|
]
|
|
result = SyncManager.process_push(client_id="node-field-1", changes=changes)
|
|
self.assertEqual(result["status_code"], 200, result)
|
|
|
|
rows = (
|
|
db.session.execute(
|
|
select(ZootechNotification).order_by(ZootechNotification.created_at.desc())
|
|
)
|
|
.scalars()
|
|
.all()
|
|
)
|
|
self.assertGreaterEqual(len(rows), 1)
|
|
latest = rows[0]
|
|
self.assertEqual(latest.title, "Отчёт о загрузке")
|
|
self.assertIn("Зал 2", latest.detail)
|
|
self.assertIn("Утренний", latest.detail)
|
|
self.assertEqual(latest.link_kind, "report_loading")
|
|
self.assertEqual(latest.link_id, report_id)
|
|
|
|
def test_confirm_recipe_delivery_creates_notification(self) -> None:
|
|
self._register_sync_client("node-kiosk-1", "Киоск 1")
|
|
recipe = Recipe(
|
|
name="Вечерний",
|
|
created_by="system",
|
|
updated_by="system",
|
|
created_at=datetime.utcnow(),
|
|
updated_at=datetime.utcnow(),
|
|
)
|
|
db.session.add(recipe)
|
|
db.session.commit()
|
|
|
|
task = SyncQueue(
|
|
table_name="recipe",
|
|
record_id=recipe.id,
|
|
action="update",
|
|
status="processing",
|
|
target_node_id="node-kiosk-1",
|
|
created_at=datetime.utcnow(),
|
|
updated_at=datetime.utcnow(),
|
|
)
|
|
db.session.add(task)
|
|
db.session.commit()
|
|
|
|
result = SyncManager.process_confirm("node-kiosk-1", [task.id])
|
|
self.assertEqual(result["status_code"], 200)
|
|
|
|
rows = (
|
|
db.session.execute(
|
|
select(ZootechNotification).order_by(ZootechNotification.created_at.desc())
|
|
)
|
|
.scalars()
|
|
.all()
|
|
)
|
|
self.assertEqual(len(rows), 1)
|
|
note = rows[0]
|
|
self.assertEqual(note.title, "Рейс на терминале")
|
|
self.assertIn("Вечерний", note.detail)
|
|
self.assertIn("Киоск 1", note.detail)
|
|
self.assertEqual(note.link_kind, "recipe")
|
|
self.assertEqual(note.link_id, recipe.id)
|
|
|
|
def test_notification_api_returns_link_fields(self) -> None:
|
|
create_notification(
|
|
title="Тест",
|
|
detail="Подробности",
|
|
link_kind="recipe",
|
|
link_id="rid-1",
|
|
)
|
|
listed = self.client.get("/api/notifications")
|
|
self.assertEqual(listed.status_code, 200)
|
|
item = listed.get_json()["items"][0]
|
|
self.assertEqual(item["linkKind"], "recipe")
|
|
self.assertEqual(item["linkId"], "rid-1")
|