116 lines
4.0 KiB
Python
116 lines
4.0 KiB
Python
"""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()
|