"""Авторацион: пул кандидатов → рацион из 3 компонентов.""" from __future__ import annotations import unittest from app import create_app, db from app.lab.calc.formulate import FormulateRequest, formulate from app.lab.calc.formulate_validate import validate_component from app.lab.commands.seed_lab_math_ration import seed_lab_math_ration from app.lab.commands.seed_math_test_profiles import seed_math_test_profiles from app.lab.models import LabAnimalProfile from app.lab.services.component_nutrients import save_component_nutrients from app.models import Component from app.models.base import default_uuid from app.services.setup_state import mark_setup_complete from tests.helpers.zootech_test_helpers import ZootechTestConfig class LabFormulateTests(unittest.TestCase): @classmethod def setUpClass(cls) -> None: cls.app = create_app(ZootechTestConfig) cls.ctx = cls.app.app_context() cls.ctx.push() db.create_all() mark_setup_complete(cls.app) seed_math_test_profiles() seed_lab_math_ration(force=True) @classmethod def tearDownClass(cls) -> None: db.session.remove() db.drop_all() cls.ctx.pop() def setUp(self) -> None: self.client = self.app.test_client() self.client.post( "/api/auth/login", json={"login": "zootech-test-admin", "password": "zootech-test-secret"}, ) profile = LabAnimalProfile.query.filter_by(profile_key="lab_math_dairy_01").first() self.assertIsNotNone(profile) self.profile_id = profile.id self.lab_components = Component.query.filter( Component.name.like("LAB тест —%"), Component.is_deleted.is_(False), ).all() self.assertGreaterEqual(len(self.lab_components), 3) self.lab_ids = [c.id for c in self.lab_components] def _bad_component(self) -> str: comp = Component( id=default_uuid(), name="LAB formulate — без нутриентов", type="тест", dry_matter=50.0, price=10.0, is_active=True, ) db.session.add(comp) db.session.commit() return comp.id def test_validate_ineligible_without_protein(self) -> None: cid = self._bad_component() v = validate_component(cid) self.assertFalse(v["eligible"]) self.assertIn("nutrients_empty", v["missing"]) def test_validate_lab_seed_component(self) -> None: v = validate_component(self.lab_ids[0]) self.assertTrue(v["eligible"]) self.assertEqual(v["missing"], []) def test_formulate_small_pool_three_ids(self) -> None: result = formulate( FormulateRequest( profile_id=self.profile_id, candidate_ids=self.lab_ids[:3], total_kg_per_head=7.3, heads_per_trip=10, grid_step=0.2, ) ) self.assertEqual(len(result["lines"]), 3) self.assertGreater(result["costTotal"], 0) self.assertGreater(result["searchStats"]["evaluations"], 0) self.assertFalse(result["searchStats"]["prefilterApplied"]) line_ids = {line["componentId"] for line in result["lines"]} self.assertTrue(line_ids.issubset(set(self.lab_ids[:3]))) def test_formulate_respects_mass_milk_on_repeat(self) -> None: """Профиль с фикс. min уСП/ЧЭЛ в БД — масса/удой всё равно влияют (force_dynamic).""" pool = self.lab_ids[:3] low = formulate( FormulateRequest( profile_id=self.profile_id, candidate_ids=pool, mass_kg=500, milk_yield_kg=15, total_kg_per_head=7.3, grid_step=0.2, ) ) high = formulate( FormulateRequest( profile_id=self.profile_id, candidate_ids=pool, mass_kg=700, milk_yield_kg=35, total_kg_per_head=7.3, grid_step=0.2, ) ) self.assertIn("usp", (low.get("dynamicNorms") or {})) self.assertIn("usp", (high.get("dynamicNorms") or {})) self.assertNotAlmostEqual( low["dynamicNorms"]["usp"]["min"], high["dynamicNorms"]["usp"]["min"], places=0, ) self.assertNotAlmostEqual(low["score"], high["score"], places=3) def test_formulate_pool_ten_no_prefilter(self) -> None: pool = self.lab_ids[:3] extras = Component.query.filter( Component.is_deleted.is_(False), ~Component.id.in_(pool), ).limit(7).all() for comp in extras: save_component_nutrients( comp.id, { "Сыр. Протеин": 50, "Сырая клетч": 200, "Сырой жир": 20, }, ) if not comp.dry_matter: comp.dry_matter = 80.0 db.session.commit() candidate_ids = pool + [c.id for c in extras] self.assertGreaterEqual(len(candidate_ids), 3) result = formulate( FormulateRequest( profile_id=self.profile_id, candidate_ids=candidate_ids, prefilter_k=18, grid_step=0.2, ) ) self.assertEqual(len(result["lines"]), 3) self.assertFalse(result["searchStats"]["prefilterApplied"]) chosen = {line["componentId"] for line in result["lines"]} self.assertTrue(chosen.issubset(set(candidate_ids))) def test_formulate_pool_triggers_prefilter(self) -> None: pool = list(self.lab_ids) for i in range(22): comp = Component( id=default_uuid(), name=f"LAB formulate filler {i}", type="тест", dry_matter=80.0, price=5.0 + i, is_active=True, ) db.session.add(comp) db.session.flush() save_component_nutrients( comp.id, { "Сыр. Протеин": 40 + i, "Сырая клетч": 150, "Сырой жир": 15, }, ) pool.append(comp.id) db.session.commit() self.assertGreater(len(pool), 18) result = formulate( FormulateRequest( profile_id=self.profile_id, candidate_ids=pool, prefilter_k=12, grid_step=0.2, ) ) self.assertTrue(result["searchStats"]["prefilterApplied"]) self.assertLessEqual(len(result["shortlistedIds"]), 12) self.assertEqual(len(result["lines"]), 3) chosen = {line["componentId"] for line in result["lines"]} self.assertTrue(chosen.issubset(set(pool))) def _default_group_payload(self) -> dict: hay = next(c for c in self.lab_components if "сено" in c.name) sil = next(c for c in self.lab_components if "силос" in c.name) conc = next(c for c in self.lab_components if "комбикорм" in c.name) return { "rough": [hay.id], "succulent": [sil.id], "concentrate": [conc.id], "other": [], } def test_api_formulate_repeat_with_different_params(self) -> None: """Два POST подряд: второй запрос с другими mass/milk/total — другой ответ.""" base = { "profileId": self.profile_id, "groupSelections": self._default_group_payload(), "gridStep": 0.2, } r1 = self.client.post( "/api/lab/formulate", json={**base, "massKg": 500, "milkYieldKg": 15, "totalKgPerHead": 7.3}, ) self.assertEqual(r1.status_code, 200, r1.get_data(as_text=True)) out1 = r1.get_json() r2 = self.client.post( "/api/lab/formulate", json={**base, "massKg": 700, "milkYieldKg": 35, "totalKgPerHead": 8.5}, ) self.assertEqual(r2.status_code, 200, r2.get_data(as_text=True)) out2 = r2.get_json() self.assertEqual(out1["searchStats"].get("scoreEngine"), "fast") self.assertEqual(out2["searchStats"].get("optimizer"), "slsqp") self.assertIn("usp", out1.get("dynamicNorms") or {}) self.assertIn("usp", out2.get("dynamicNorms") or {}) self.assertNotAlmostEqual( out1["dynamicNorms"]["usp"]["min"], out2["dynamicNorms"]["usp"]["min"], places=0, ) self.assertNotAlmostEqual(out1["score"], out2["score"], places=3) total1 = sum(line["dailyKg"] for line in out1["lines"]) total2 = sum(line["dailyKg"] for line in out2["lines"]) self.assertNotAlmostEqual(total1, total2, places=2) def test_api_formulate_validate_and_post(self) -> None: r = self.client.get(f"/api/lab/formulate/validate?ids={','.join(self.lab_ids[:2])}") self.assertEqual(r.status_code, 200) body = r.get_json() self.assertEqual(len(body["components"]), 2) hay = next(c for c in self.lab_components if "сено" in c.name) sil = next(c for c in self.lab_components if "силос" in c.name) conc = next(c for c in self.lab_components if "комбикорм" in c.name) r2 = self.client.post( "/api/lab/formulate", json={ "profileId": self.profile_id, "groupSelections": { "rough": [hay.id], "succulent": [sil.id], "concentrate": [conc.id], "other": [], }, "totalKgPerHead": 7.3, "headsPerTrip": 10, "gridStep": 0.2, }, ) self.assertEqual(r2.status_code, 200, r2.get_data(as_text=True)) out = r2.get_json() self.assertEqual(len(out["lines"]), 3) chosen = {line["componentId"] for line in out["lines"]} self.assertIn(hay.id, chosen) keys = {i["key"] for i in out.get("indicators") or []} self.assertIn("fat_pct_per_kg_dm", keys) def test_formulate_requires_rough_in_triplet(self) -> None: hay = next(c for c in self.lab_components if "сено" in c.name) sil = next(c for c in self.lab_components if "силос" in c.name) conc = next(c for c in self.lab_components if "комбикорм" in c.name) result = formulate( FormulateRequest( profile_id=self.profile_id, group_selections={ "rough": [hay.id], "succulent": [sil.id], "concentrate": [conc.id], "other": [], }, grid_step=0.2, ) ) chosen = {line["componentId"] for line in result["lines"]} self.assertIn(hay.id, chosen) def test_api_rejects_small_pool(self) -> None: r = self.client.post( "/api/lab/formulate", json={"profileId": self.profile_id, "candidateIds": self.lab_ids[:2]}, ) self.assertEqual(r.status_code, 400)