96 lines
3.0 KiB
Python
96 lines
3.0 KiB
Python
"""feed_quality после sync push."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import unittest
|
|
import uuid
|
|
from datetime import datetime
|
|
|
|
from sqlalchemy import select
|
|
|
|
from app import create_app, db
|
|
from app.models import Recipe
|
|
from app.models.feed_alert import FeedAlert
|
|
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 FeedQualitySyncTests(unittest.TestCase):
|
|
def setUp(self) -> None:
|
|
self.app = create_app(ZootechTestConfig)
|
|
self.ctx = self.app.app_context()
|
|
self.ctx.push()
|
|
db.create_all()
|
|
mark_setup_complete(self.app)
|
|
recipe = Recipe(
|
|
name="Sync FQ",
|
|
created_by="system",
|
|
updated_by="system",
|
|
created_at=datetime.utcnow(),
|
|
updated_at=datetime.utcnow(),
|
|
)
|
|
db.session.add(recipe)
|
|
db.session.commit()
|
|
self.recipe_id = recipe.id
|
|
|
|
def tearDown(self) -> None:
|
|
db.session.remove()
|
|
db.drop_all()
|
|
self.ctx.pop()
|
|
|
|
def test_push_overload_report_creates_alert(self) -> None:
|
|
report_id = str(uuid.uuid4())
|
|
changes = [
|
|
{
|
|
"table_name": "loading_report",
|
|
"record_id": report_id,
|
|
"action": "create",
|
|
"data": {
|
|
"id": report_id,
|
|
"recipe_id": self.recipe_id,
|
|
"recipe_name": "Sync FQ",
|
|
"start_time": "2026-06-07T10:00:00",
|
|
"target_mixing_time": 300,
|
|
"actual_mixing_time": 300,
|
|
"dispenser_type": "dispenser",
|
|
"version": 1,
|
|
"created_by": "terminal",
|
|
"updated_by": "terminal",
|
|
"content_hash": "abc",
|
|
},
|
|
},
|
|
{
|
|
"table_name": "loading_report_component",
|
|
"record_id": str(uuid.uuid4()),
|
|
"action": "create",
|
|
"data": {
|
|
"report_id": report_id,
|
|
"component_name": "C1",
|
|
"target_weight": 100.0,
|
|
"actual_weight": 112.0,
|
|
"overload": 12.0,
|
|
"loading_order": 1,
|
|
"version": 1,
|
|
"created_by": "terminal",
|
|
"updated_by": "terminal",
|
|
"content_hash": "def",
|
|
},
|
|
},
|
|
]
|
|
result = SyncManager.process_push(client_id="node-fq-1", changes=changes)
|
|
self.assertEqual(result["status_code"], 200, result)
|
|
|
|
alerts = (
|
|
db.session.execute(
|
|
select(FeedAlert).where(FeedAlert.loading_report_id == report_id)
|
|
)
|
|
.scalars()
|
|
.all()
|
|
)
|
|
self.assertGreaterEqual(len(alerts), 1)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|