298 lines
11 KiB
Python
298 lines
11 KiB
Python
"""Авторацион: пул кандидатов → рацион из 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)
|