"""API /api/feed-quality.""" from __future__ import annotations import unittest from datetime import datetime from app import create_app, db from app.models import ComponentLoadingTime, LoadingReport, LoadingReportComponent, Recipe from app.services.feed_quality.evaluator import evaluate_loading_report from app.services.setup_state import mark_setup_complete from tests.helpers.zootech_test_helpers import ZootechTestConfig class FeedQualityApiTests(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"}, ) db.session.add( Recipe(id="api-r1", name="API рейс", heads_per_trip=1, content_hash="") ) report = LoadingReport( id="api-lr-1", recipe_id="api-r1", recipe_name="API рейс", start_time=datetime.now(), target_mixing_time=60, actual_mixing_time=60, total_weight=50.0, created_by="system", updated_by="system", ) db.session.add(report) db.session.flush() db.session.add( LoadingReportComponent( report_id=report.id, component_name="C1", target_weight=50.0, actual_weight=60.0, overload=10.0, loading_order=1, created_by="system", updated_by="system", ) ) db.session.add( ComponentLoadingTime( report_id=report.id, component_name="C1", start_time=datetime.now(), end_time=datetime.now(), loading_duration=42.5, loading_order=1, created_by="system", updated_by="system", ) ) db.session.commit() evaluate_loading_report(report.id, send_notifications=False) def tearDown(self) -> None: db.session.remove() db.drop_all() self.ctx.pop() def test_list_alerts(self) -> None: today = datetime.now().strftime("%Y-%m-%d") resp = self.client.get(f"/api/feed-quality/alerts?date_from={today}&date_to={today}") self.assertEqual(resp.status_code, 200) body = resp.get_json() self.assertGreaterEqual(body["total"], 1) self.assertIn("bySeverity", body) self.assertEqual(body["items"][0]["linkKind"], "report_loading") component_alert = next( (x for x in body["items"] if x.get("componentName") == "C1"), body["items"][0], ) self.assertAlmostEqual(float(component_alert["actualKg"]), 60.0, places=1) self.assertAlmostEqual(float(component_alert["durationSec"]), 42.5, places=1) def test_summary(self) -> None: today = datetime.now().strftime("%Y-%m-%d") resp = self.client.get(f"/api/feed-quality/alerts/summary?date_from={today}&date_to={today}") self.assertEqual(resp.status_code, 200) self.assertGreaterEqual(resp.get_json()["total"], 1) def test_requires_auth(self) -> None: self.client.post("/api/auth/logout") resp = self.client.get("/api/feed-quality/alerts") self.assertEqual(resp.status_code, 401) def test_backfill_on_list_for_unevaluated_report(self) -> None: from sqlalchemy import delete from app.models.feed_alert import FeedAlert db.session.execute(delete(FeedAlert)) db.session.commit() today = datetime.now().strftime("%Y-%m-%d") resp = self.client.get(f"/api/feed-quality/alerts?date_from={today}&date_to={today}") self.assertEqual(resp.status_code, 200) self.assertGreaterEqual(resp.get_json()["total"], 1) if __name__ == "__main__": unittest.main()