"""Unit-тесты analytics services.""" from __future__ import annotations import unittest from datetime import datetime from app import create_app, db from app.models import ( Component, LoadingReport, LoadingReportComponent, Recipe, UnloadingReport, UnloadingReportGroup, ) from app.services.analytics.finance_summary import build_finance_summary from app.services.analytics.plan_fact_layers import build_plan_fact_rows from app.services.daily_plan.skips import skip_ingredient from app.services.setup_state import mark_setup_complete from tests.helpers.dispenser_recipe_fixtures import ( E2E_DISP_COMP, E2E_DISP_RECIPE_1, seed_dispenser_period_recipes, ) from tests.helpers.zootech_test_helpers import ZootechTestConfig class AnalyticsFinanceSummaryTests(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) db.session.add( Component( id="an-comp-1", name="Премикс", type="premix", dry_matter=90.0, protein=0.0, energy=0.0, price=150.0, ) ) db.session.add( Recipe(id="an-r1", name="Рейс A", heads_per_trip=10, content_hash="") ) report = LoadingReport( id="an-lr-1", recipe_id="an-r1", recipe_name="Рейс A", start_time=datetime(2026, 6, 7, 10, 0, 0), target_mixing_time=60, actual_mixing_time=60, total_weight=100.0, created_by="system", updated_by="system", ) db.session.add(report) db.session.flush() db.session.add( LoadingReportComponent( report_id=report.id, component_id="an-comp-1", component_name="Премикс", target_weight=100.0, actual_weight=110.0, overload=10.0, loading_order=1, created_by="system", updated_by="system", ) ) db.session.add( LoadingReportComponent( report_id=report.id, component_id="an-comp-1", component_name="Премикс", target_weight=50.0, actual_weight=40.0, overload=0.0, loading_order=2, created_by="system", updated_by="system", ) ) db.session.commit() def tearDown(self) -> None: db.session.remove() db.drop_all() self.ctx.pop() def test_overload_and_underload_rub(self) -> None: summary = build_finance_summary(date_from="2026-06-07", date_to="2026-06-07") # +10 kg * 150 = 1500 overload; -10 kg * 150 = 1500 underload self.assertEqual(summary["overloadRub"], 1500.0) self.assertEqual(summary["underloadRub"], 1500.0) self.assertEqual(summary["netRub"], 0.0) self.assertEqual(summary["dominantIssue"], "balanced") self.assertEqual(len(summary["topComponents"]), 1) self.assertEqual(summary["topComponents"][0]["name"], "Премикс") class AnalyticsPlanFactTests(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) seed_dispenser_period_recipes() report = LoadingReport( id="pf-lr-1", recipe_id=E2E_DISP_RECIPE_1, recipe_name="E2E Рейс 1", start_time=datetime(2026, 6, 7, 8, 0, 0), target_mixing_time=60, actual_mixing_time=60, total_weight=25.0, created_by="system", updated_by="system", ) db.session.add(report) db.session.flush() db.session.add( LoadingReportComponent( report_id=report.id, component_id=E2E_DISP_COMP, component_name="E2E Disp компонент", target_weight=20.0, actual_weight=45.0, overload=25.0, loading_order=1, created_by="system", updated_by="system", ) ) db.session.commit() def tearDown(self) -> None: db.session.remove() db.drop_all() self.ctx.pop() def test_plan_fact_layers_and_notes(self) -> None: data = build_plan_fact_rows(date_from="2026-06-07", date_to="2026-06-07") self.assertEqual(data["reportCount"], 1) item = data["items"][0] self.assertEqual(item["recipeId"], E2E_DISP_RECIPE_1) comps = item["components"] self.assertEqual(len(comps), 1) c = comps[0] self.assertEqual(c["baseKg"], 20.0) self.assertEqual(c["planTodayKg"], 20.0) self.assertEqual(c["actualKg"], 45.0) exec_notes = [n for n in c["notes"] if n.get("kind") == "execution"] self.assertEqual(len(exec_notes), 1) self.assertIn("перегруз", exec_notes[0]["text"]) self.assertEqual(c["fault"], "execution") def test_zootech_skip_without_report_row(self) -> None: """Исключённый ингредиент без строки в отчёте терминала — всё равно в сравнении.""" skip_ingredient(E2E_DISP_RECIPE_1, "e2e-disp-ing-1", "2026-06-07") report = LoadingReport( id="pf-lr-skip-only-other", recipe_id=E2E_DISP_RECIPE_1, recipe_name="E2E Рейс 1", start_time=datetime(2026, 6, 7, 11, 0, 0), target_mixing_time=60, actual_mixing_time=60, total_weight=100.0, created_by="system", updated_by="system", ) db.session.add(report) db.session.flush() db.session.add( LoadingReportComponent( report_id=report.id, component_id="e2e-disp-comp-other", component_name="Другой компонент", target_weight=100.0, actual_weight=100.0, overload=0.0, loading_order=2, created_by="system", updated_by="system", ) ) db.session.commit() data = build_plan_fact_rows(date_from="2026-06-07", date_to="2026-06-07") item = next( (i for i in data["items"] if i["loadingReportId"] == "pf-lr-skip-only-other"), None, ) self.assertIsNotNone(item) skipped_rows = [c for c in item["components"] if c.get("skippedToday")] self.assertEqual(len(skipped_rows), 1) self.assertEqual(skipped_rows[0]["fault"], "excluded") self.assertEqual(skipped_rows[0]["actualKg"], 0.0) self.assertGreater(skipped_rows[0]["baseKg"], 0.0) def test_zootech_skip_no_execution_fault(self) -> None: skip_ingredient(E2E_DISP_RECIPE_1, "e2e-disp-ing-1", "2026-06-07") report = LoadingReport( id="pf-lr-skip", recipe_id=E2E_DISP_RECIPE_1, recipe_name="E2E Рейс 1", start_time=datetime(2026, 6, 7, 9, 0, 0), target_mixing_time=60, actual_mixing_time=60, total_weight=0.0, created_by="system", updated_by="system", ) db.session.add(report) db.session.flush() db.session.add( LoadingReportComponent( report_id=report.id, component_id=E2E_DISP_COMP, component_name="E2E Disp компонент", target_weight=0.0, actual_weight=0.0, overload=0.0, loading_order=1, created_by="system", updated_by="system", ) ) db.session.commit() data = build_plan_fact_rows(date_from="2026-06-07", date_to="2026-06-07") skip_item = next( (i for i in data["items"] if i["loadingReportId"] == "pf-lr-skip"), None, ) self.assertIsNotNone(skip_item) c = skip_item["components"][0] self.assertEqual(c["planTodayKg"], 0.0) self.assertEqual(c["actualKg"], 0.0) self.assertEqual(c["fault"], "excluded") self.assertTrue(c["skippedToday"]) zootech_notes = [n for n in c["notes"] if n.get("kind") == "zootech"] self.assertEqual(len(zootech_notes), 1) self.assertIn("исключено из плана", zootech_notes[0]["text"]) self.assertNotIn("execution", [n.get("kind") for n in c["notes"]]) def test_operator_underload_is_execution_not_excluded(self) -> None: """Полный недогруз при плане 20 кг — в пределах допуска ±20 кг, без ошибки.""" report = LoadingReport( id="pf-lr-hay", recipe_id=E2E_DISP_RECIPE_1, recipe_name="E2E Рейс 1", start_time=datetime(2026, 6, 7, 10, 0, 0), target_mixing_time=60, actual_mixing_time=60, total_weight=0.0, created_by="system", updated_by="system", ) db.session.add(report) db.session.flush() db.session.add( LoadingReportComponent( report_id=report.id, component_id=E2E_DISP_COMP, component_name="E2E Disp компонент", target_weight=20.0, actual_weight=0.0, overload=0.0, loading_order=1, created_by="system", updated_by="system", ) ) db.session.commit() data = build_plan_fact_rows(date_from="2026-06-07", date_to="2026-06-07") hay_item = next( (i for i in data["items"] if i["loadingReportId"] == "pf-lr-hay"), None, ) self.assertIsNotNone(hay_item) c = hay_item["components"][0] self.assertEqual(c["planTodayKg"], 20.0) self.assertEqual(c["actualKg"], 0.0) self.assertFalse(c["skippedToday"]) self.assertEqual(c["fault"], "none") self.assertEqual([n for n in c["notes"] if n.get("kind") == "execution"], []) def test_plan_fact_unloading_and_mixer_remainder(self) -> None: unloading = UnloadingReport( id="pf-ur-1", recipe_id=E2E_DISP_RECIPE_1, recipe_name="E2E Рейс 1", loading_report_id="pf-lr-1", start_time=datetime(2026, 6, 7, 8, 30, 0), end_time=datetime(2026, 6, 7, 8, 45, 0), total_weight=25.0, total_unloaded_weight=22.0, remaining_weight=3.0, created_by="system", updated_by="system", ) db.session.add(unloading) db.session.flush() db.session.add( UnloadingReportGroup( report_id=unloading.id, name="Г1", target_weight=20.0, unloaded_weight=45.0, remaining_weight=0.0, distribution_type="percent", distribution_value=100.0, order=1, created_by="system", updated_by="system", ) ) db.session.commit() data = build_plan_fact_rows(date_from="2026-06-07", date_to="2026-06-07") item = data["items"][0] self.assertEqual(len(item["unloadingGroups"]), 1) grp = item["unloadingGroups"][0] self.assertEqual(grp["name"], "Г1") self.assertEqual(grp["actualKg"], 45.0) self.assertEqual(grp["fault"], "execution") self.assertIsNotNone(item["mixerRemainder"]) self.assertEqual(item["mixerRemainder"]["actualKg"], 3.0) self.assertEqual(item["mixerRemainder"]["fault"], "none") def test_plan_fact_mixer_remainder_can_be_negative(self) -> None: unloading = UnloadingReport( id="pf-ur-neg", recipe_id=E2E_DISP_RECIPE_1, recipe_name="E2E Рейс 1", loading_report_id="pf-lr-1", start_time=datetime(2026, 6, 7, 9, 0, 0), end_time=datetime(2026, 6, 7, 9, 15, 0), total_weight=25.0, total_unloaded_weight=30.0, remaining_weight=-5.0, created_by="system", updated_by="system", ) db.session.add(unloading) db.session.commit() data = build_plan_fact_rows(date_from="2026-06-07", date_to="2026-06-07") item = next(i for i in data["items"] if i["loadingReportId"] == "pf-lr-1") self.assertIsNotNone(item["mixerRemainder"]) self.assertEqual(item["mixerRemainder"]["actualKg"], -5.0) self.assertEqual(item["mixerRemainder"]["fault"], "none") class AnalyticsReportsDetailTests(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) db.session.add( Recipe(id="an-r-detail", name="Рейс detail", heads_per_trip=10, content_hash="") ) report = LoadingReport( id="an-lr-detail", recipe_id="an-r-detail", recipe_name="Рейс detail", start_time=datetime(2026, 6, 7, 17, 43, 0), target_mixing_time=60, actual_mixing_time=60, total_weight=100.0, created_by="system", updated_by="system", ) db.session.add(report) db.session.add( LoadingReportComponent( report_id=report.id, component_name="БУРТ", target_weight=394.1, actual_weight=0.0, overload=-394.1, loading_order=1, created_by="system", updated_by="system", ) ) db.session.add( LoadingReportComponent( report_id=report.id, component_name="БУРТ", target_weight=394.1, actual_weight=0.0, overload=-394.1, loading_order=2, created_by="system", updated_by="system", ) ) db.session.add( LoadingReportComponent( report_id=report.id, component_name="Кукуруза", target_weight=400.0, actual_weight=450.0, overload=50.0, loading_order=3, created_by="system", updated_by="system", ) ) db.session.commit() def tearDown(self) -> None: db.session.remove() db.drop_all() self.ctx.pop() def test_overload_only_when_fact_exceeds_plan(self) -> None: from app.services.analytics.reports_detail import build_reports_detail_rows rows = build_reports_detail_rows(date_from="2026-06-07", date_to="2026-06-07") by_name = {r["feedComponent"]: r for r in rows if r["kind"] == "component"} self.assertEqual(by_name["БУРТ"]["deviationKg"], -394.1) self.assertEqual(by_name["БУРТ"]["overloadKg"], 0.0) self.assertEqual(by_name["Кукуруза"]["deviationKg"], 50.0) self.assertEqual(by_name["Кукуруза"]["overloadKg"], 50.0) def test_dedupe_identical_component_rows(self) -> None: from app.services.analytics.reports_detail import build_reports_detail_rows rows = build_reports_detail_rows(date_from="2026-06-07", date_to="2026-06-07") burt_rows = [r for r in rows if r.get("feedComponent") == "БУРТ"] self.assertEqual(len(burt_rows), 1) class AnalyticsReportsDetailUnloadingTests(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) from app.models import UnloadingReport, UnloadingReportGroup db.session.add( Recipe(id="an-r-unload", name="Рейс unload", heads_per_trip=10, content_hash="") ) report = LoadingReport( id="an-lr-unload", recipe_id="an-r-unload", recipe_name="Рейс unload", start_time=datetime(2026, 6, 7, 18, 0, 0), target_mixing_time=300, actual_mixing_time=300, total_weight=3340.0, created_by="system", updated_by="system", ) db.session.add(report) db.session.flush() unloading = UnloadingReport( id="an-ur-1", recipe_id="an-r-unload", recipe_name="Рейс unload", loading_report_id=report.id, start_time=datetime(2026, 6, 7, 18, 30, 0), total_weight=3340.0, total_unloaded_weight=3300.0, remaining_weight=40.0, created_by="system", updated_by="system", ) db.session.add(unloading) db.session.flush() db.session.add( UnloadingReportGroup( report_id=unloading.id, name="выгрузка: 4 гр 1 зам", target_weight=3340.0, unloaded_weight=3300.0, remaining_weight=40.0, distribution_type="percent", distribution_value=100.0, order=1, created_by="system", updated_by="system", ) ) db.session.add( UnloadingReportGroup( report_id=unloading.id, name="все", target_weight=1710.0, unloaded_weight=1710.0, remaining_weight=0.0, distribution_type="percent", distribution_value=100.0, order=2, created_by="system", updated_by="system", ) ) db.session.commit() def tearDown(self) -> None: db.session.remove() db.drop_all() self.ctx.pop() def test_unloading_row_schema(self) -> None: from app.services.analytics.reports_detail import ( UNLOADING_FEED_COMPONENT_LABEL, build_reports_detail_rows, ) rows = [r for r in build_reports_detail_rows(date_from="2026-06-07", date_to="2026-06-07") if r["kind"] == "unloading"] self.assertEqual(len(rows), 2) by_group = {r["animalGroup"]: r for r in rows} self.assertEqual(by_group["4 гр 1 зам"]["feedComponent"], UNLOADING_FEED_COMPONENT_LABEL) self.assertEqual(by_group["4 гр 1 зам"]["targetKg"], 3340.0) self.assertEqual(by_group["4 гр 1 зам"]["actualKg"], 3300.0) self.assertEqual(by_group["4 гр 1 зам"]["deviationKg"], -40.0) self.assertEqual(by_group["все"]["feedComponent"], UNLOADING_FEED_COMPONENT_LABEL) self.assertEqual(by_group["все"]["targetKg"], 1710.0) self.assertEqual(by_group["все"]["actualKg"], 1710.0) self.assertEqual(by_group["все"]["deviationKg"], 0.0) def test_detail_xlsx_unloading_column_alignment(self) -> None: from io import BytesIO from openpyxl import load_workbook from app.services.analytics.reports_detail import UNLOADING_FEED_COMPONENT_LABEL from app.services.analytics.xlsx_summary import build_finance_summary_xlsx xlsx = build_finance_summary_xlsx( date_from="2026-06-07", date_to="2026-06-07", section="reports", ) ws = load_workbook(BytesIO(xlsx))["Подробно"] self.assertEqual(ws.cell(1, 8).value, "Группа") self.assertGreaterEqual(ws.column_dimensions["B"].width, 14) self.assertGreaterEqual(ws.column_dimensions["I"].width, 8) farm_cell = ws.cell(row=2, column=2) self.assertTrue(farm_cell.alignment.wrap_text) unloading_rows = [ row for row in ws.iter_rows(min_row=2, values_only=True) if row[5] == "Выгрузка" ] self.assertEqual(len(unloading_rows), 2) for row in unloading_rows: self.assertEqual(row[6], UNLOADING_FEED_COMPONENT_LABEL) self.assertIsInstance(row[8], (int, float)) self.assertIsInstance(row[9], (int, float)) self.assertEqual(round(row[10], 2), round(row[9] - row[8], 2)) if __name__ == "__main__": unittest.main()