"""Unit-тесты порогов feed_quality.rules.""" from __future__ import annotations import unittest from app import create_app, db from app.services.feed_quality.rules import ( EVENT_LEFT_IN_MIXER, EVENT_LOADING_FAST, EVENT_LOADING_TIME, EVENT_MIX_TIME, EVENT_OVERLOAD, EVENT_UNDERLOAD, SEVERITY_ERROR, SEVERITY_WARNING, TIME_ISSUE_FAST, classify_component_loading_time, classify_left_in_mixer, classify_loading_component, classify_mix_time, classify_unloading_group, merge_component_loading_alerts, ) from app.services.feed_quality.settings_store import save_feed_quality_settings from tests.helpers.zootech_test_helpers import ZootechTestConfig class FeedQualityRulesTests(unittest.TestCase): def setUp(self) -> None: self.app = create_app(ZootechTestConfig) self.ctx = self.app.app_context() self.ctx.push() db.create_all() def tearDown(self) -> None: db.session.remove() db.drop_all() self.ctx.pop() def test_loading_overload_at_threshold(self) -> None: alerts = classify_loading_component( component_name="Силос", target_kg=100.0, actual_kg=111.0, price_rub_per_kg=10.0, ) self.assertEqual(len(alerts), 1) self.assertEqual(alerts[0].event_type, EVENT_OVERLOAD) self.assertAlmostEqual(alerts[0].deviation_pct or 0, 11.0, places=1) self.assertEqual(alerts[0].cost_deviation_rub, 110.0) def test_loading_under_threshold_no_alert(self) -> None: alerts = classify_loading_component( component_name="Силос", target_kg=100.0, actual_kg=105.0, ) self.assertEqual(alerts, []) def test_loading_underload(self) -> None: alerts = classify_loading_component( component_name="Силос", target_kg=100.0, actual_kg=85.0, ) self.assertEqual(len(alerts), 1) self.assertEqual(alerts[0].event_type, EVENT_UNDERLOAD) def test_overload_error_severity(self) -> None: alerts = classify_loading_component( component_name="X", target_kg=100.0, actual_kg=120.0, ) self.assertEqual(alerts[0].severity, SEVERITY_ERROR) def test_loading_disabled(self) -> None: save_feed_quality_settings({"loading": {"enabled": False}}) alerts = classify_loading_component( component_name="X", target_kg=100.0, actual_kg=50.0, ) self.assertEqual(alerts, []) def test_custom_loading_thresholds(self) -> None: save_feed_quality_settings({"loading": {"warning_pct": 20, "critical_pct": 25}}) alerts = classify_loading_component( component_name="X", target_kg=100.0, actual_kg=115.0, ) self.assertEqual(alerts, []) alerts2 = classify_loading_component( component_name="X", target_kg=100.0, actual_kg=122.0, ) self.assertEqual(len(alerts2), 1) self.assertEqual(alerts2[0].severity, SEVERITY_WARNING) def test_mix_time_alert(self) -> None: alerts = classify_mix_time(target_sec=300, actual_sec=340) self.assertEqual(len(alerts), 1) self.assertEqual(alerts[0].event_type, EVENT_MIX_TIME) self.assertEqual(alerts[0].severity, SEVERITY_WARNING) def test_mix_time_critical(self) -> None: alerts = classify_mix_time(target_sec=300, actual_sec=400) self.assertEqual(alerts[0].severity, SEVERITY_ERROR) def test_mix_time_within_tolerance(self) -> None: self.assertEqual(classify_mix_time(target_sec=300, actual_sec=320), []) def test_mix_time_disabled(self) -> None: save_feed_quality_settings({"mix_time": {"enabled": False}}) self.assertEqual(classify_mix_time(target_sec=300, actual_sec=400), []) def test_unloading_group_overload(self) -> None: alerts = classify_unloading_group( group_name="Группа 1", target_kg=100.0, unloaded_kg=108.0, ) self.assertEqual(len(alerts), 1) self.assertEqual(alerts[0].event_type, EVENT_OVERLOAD) def test_left_in_mixer_warning(self) -> None: alerts = classify_left_in_mixer(remaining_kg=8.0, total_kg=200.0) self.assertEqual(len(alerts), 1) self.assertEqual(alerts[0].event_type, EVENT_LEFT_IN_MIXER) self.assertEqual(alerts[0].severity, SEVERITY_WARNING) def test_left_in_mixer_critical(self) -> None: alerts = classify_left_in_mixer(remaining_kg=16.0, total_kg=200.0) self.assertEqual(alerts[0].severity, SEVERITY_ERROR) def test_left_in_mixer_below_threshold(self) -> None: self.assertEqual(classify_left_in_mixer(remaining_kg=2.0, total_kg=500.0), []) def test_loading_time_too_fast_merged_not_standalone(self) -> None: time_issue = classify_component_loading_time(component_name="Сено", duration_sec=0.9) self.assertIsNotNone(time_issue) self.assertEqual(time_issue.kind, TIME_ISSUE_FAST) weight = classify_loading_component( component_name="Сено", target_kg=62.0, actual_kg=0.0 )[0] merged = merge_component_loading_alerts( component_name="Сено", weight_alert=weight, time_issue=time_issue, ) self.assertEqual(len(merged), 1) self.assertEqual(merged[0].event_type, EVENT_UNDERLOAD) self.assertIn("загрузка 0.9 с", merged[0].detail) def test_loading_time_fast_standalone_when_weight_ok(self) -> None: time_issue = classify_component_loading_time(component_name="Сено", duration_sec=0.9) alerts = merge_component_loading_alerts( component_name="Сено", weight_alert=None, time_issue=time_issue, ) self.assertEqual(len(alerts), 1) self.assertEqual(alerts[0].event_type, EVENT_LOADING_FAST) self.assertIn("0.9 с", alerts[0].detail) def test_loading_time_slow_standalone_when_weight_ok(self) -> None: save_feed_quality_settings({"loading": {"warning_max_sec": 60, "critical_max_sec": 120}}) time_issue = classify_component_loading_time(component_name="Силос", duration_sec=90.0) self.assertIsNotNone(time_issue) alerts = merge_component_loading_alerts( component_name="Силос", weight_alert=None, time_issue=time_issue, ) self.assertEqual(len(alerts), 1) self.assertEqual(alerts[0].event_type, EVENT_LOADING_TIME) self.assertEqual(alerts[0].severity, SEVERITY_WARNING) def test_loading_time_slow_merged_when_weight_bad(self) -> None: save_feed_quality_settings({"loading": {"warning_max_sec": 60, "critical_max_sec": 120}}) time_issue = classify_component_loading_time(component_name="X", duration_sec=200.0) weight = classify_loading_component( component_name="X", target_kg=100.0, actual_kg=111.0 )[0] merged = merge_component_loading_alerts( component_name="X", weight_alert=weight, time_issue=time_issue, ) self.assertEqual(len(merged), 1) self.assertEqual(merged[0].event_type, EVENT_OVERLOAD) self.assertEqual(merged[0].severity, SEVERITY_ERROR) if __name__ == "__main__": unittest.main()