"""AgroStar XML import — parser, matching, API.""" from __future__ import annotations import unittest from pathlib import Path from app import create_app, db from app.lab.etl.agrostar_pdf_import import parse_agrostar_pdf from app.lab.etl.agrostar_xml_import import ( apply_agrostar_import, build_sample_preview, build_storage_report, enrich_parse_with_matches, parse_agrostar_xml, suggest_canonical_feed_type, suggest_component_matches, ) from app.models.base import default_uuid from app.models.component import Component from app.services.setup_state import mark_setup_complete from tests.helpers.zootech_test_helpers import ZootechTestConfig _FIXTURE = Path(__file__).resolve().parent / "fixtures" / "lab" / "agrostar_yama5.xml" _REF = Path(__file__).resolve().parents[3] / "для разбора и внедрения " class AgrostarXmlImportTests(unittest.TestCase): def test_parse_yama5_fixture(self) -> None: text = _FIXTURE.read_text(encoding="utf-8") result = parse_agrostar_xml(text) self.assertFalse(result.errors, result.errors) self.assertEqual(result.lab_name, "АгроСтар") self.assertEqual(len(result.samples), 1) sample = result.samples[0] self.assertEqual(sample.sample_no, "471505270426") self.assertAlmostEqual(sample.dry_matter_pct or 0, 25.06, places=2) self.assertAlmostEqual(sample.nutrients["Сыр. Протеин"], 107.5, places=1) self.assertAlmostEqual(sample.nutrients["Сырая клетч"], 645.5, places=1) self.assertAlmostEqual(sample.nutrients["ВРХ Орг Вещ"], 53.01, places=2) self.assertIn("omd_from_tdn", sample.warnings) def test_build_sample_preview(self) -> None: text = _FIXTURE.read_text(encoding="utf-8") sample = parse_agrostar_xml(text).samples[0] preview = build_sample_preview(sample) self.assertIn("Смешанный сенаж", preview["feedTypeRu"]) self.assertGreaterEqual(preview["recognizedCount"], 10) labels = [row["label"] for row in preview["willWrite"]] self.assertIn("Сухое вещество (поле компонента)", labels) self.assertIn("Сырой протеин", labels) protein = next(row for row in preview["willWrite"] if row["label"] == "Сырой протеин") self.assertEqual(protein["sourceValue"], "10.75") self.assertEqual(protein["sourceUnit"], "%DM") self.assertEqual(protein["value"], "107.5") self.assertTrue(any("TDN" in n or "ВРХ" in n for n in preview["notes"])) @unittest.skipUnless(_REF.is_dir(), "reference folder missing") def test_pdf_structural_fiber_from_andfom(self) -> None: pdf = next(_REF.glob("5LDBW*.pdf"), None) if pdf is None: self.skipTest("no agrostar pdf") sample = parse_agrostar_pdf(pdf.read_bytes()).samples[0] self.assertAlmostEqual(sample.nutrients["КДК"], 444.4, places=1) self.assertAlmostEqual(sample.nutrients["Структур. клетч"], 621.8, places=1) preview = build_sample_preview(sample) struct = next(row for row in preview["willWrite"] if row["label"] == "Структурная клетчатка") self.assertEqual(struct["sourceValue"], "62.18") self.assertEqual(struct["value"], "621.8") def test_suggest_canonical_feed_type(self) -> None: text = _FIXTURE.read_text(encoding="utf-8") sample = parse_agrostar_xml(text).samples[0] self.assertEqual(suggest_canonical_feed_type(sample), "Сочные корма") self.assertEqual(build_sample_preview(sample)["suggestedType"], "Сочные корма") def test_storage_report_flags_unsupported(self) -> None: text = _FIXTURE.read_text(encoding="utf-8") sample = parse_agrostar_xml(text).samples[0] report = build_storage_report(sample) self.assertGreaterEqual(report["nutrientCount"], 15) self.assertTrue(any(u["agroKey"] == "NDFDom_IV_30hr" for u in report["unsupported"])) self.assertIn("переваримость NDF", report["unsupportedMessage"]) self.assertIn("останутся только в AgroStar", report["unsupportedMessage"]) def test_enrich_parse_includes_preview(self) -> None: app = create_app(ZootechTestConfig) with app.app_context(): db.create_all() text = _FIXTURE.read_text(encoding="utf-8") body = enrich_parse_with_matches(parse_agrostar_xml(text)) self.assertIn("preview", body["samples"][0]) self.assertTrue(body["samples"][0]["preview"]["willWrite"]) db.drop_all() def test_parse_invalid_xml(self) -> None: result = parse_agrostar_xml("valid") self.assertTrue(result.errors) self.assertEqual(len(result.samples), 0) def test_suggest_matches_silos(self) -> None: app = create_app(ZootechTestConfig) with app.app_context(): db.create_all() comp = Component( id=default_uuid(), name="Силос ПЗ Мельниково Тр №5. СВ-251", type="Сочные", dry_matter=25.0, is_active=True, ) db.session.add(comp) db.session.commit() matches = suggest_component_matches("Силос разнотравье яма 5 закрытая") self.assertTrue(matches) self.assertGreater(matches[0].score, 0.25) db.drop_all() class AgrostarImportApiTests(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.comp = Component( id=default_uuid(), name="Силос яма 5 тест", type="Сочные", dry_matter=20.0, is_active=True, ) db.session.add(self.comp) db.session.commit() self.client.post( "/api/auth/login", json={"login": "zootech-test-admin", "password": "zootech-test-secret"}, ) def tearDown(self) -> None: db.session.remove() db.drop_all() self.ctx.pop() def test_parse_api_json(self) -> None: xml = _FIXTURE.read_text(encoding="utf-8") r = self.client.post("/api/lab/import/agrostar-xml", json={"xml": xml}) self.assertEqual(r.status_code, 200, r.get_data(as_text=True)) body = r.get_json() self.assertEqual(body["sampleCount"], 1) self.assertTrue(body["samples"][0]["nutrients"]) self.assertIn("willWrite", body["samples"][0]["preview"]) def test_parse_api_multipart(self) -> None: with _FIXTURE.open("rb") as fh: r = self.client.post( "/api/lab/import/agrostar-xml", data={"file": (fh, "sample.xml")}, content_type="multipart/form-data", ) self.assertEqual(r.status_code, 200, r.get_data(as_text=True)) def test_create_component_from_agrostar_sample(self) -> None: xml = _FIXTURE.read_text(encoding="utf-8") sample = parse_agrostar_xml(xml).samples[0] preview = build_sample_preview(sample) r = self.client.post( "/api/components", json={ "name": preview["suggestedName"], "type": preview["suggestedType"], "dry_matter": sample.dry_matter_pct, "nutrients": sample.nutrients, }, ) self.assertEqual(r.status_code, 201, r.get_data(as_text=True)) cid = r.get_json()["id"] one = self.client.get(f"/api/components/{cid}") self.assertEqual(one.status_code, 200) body = one.get_json() self.assertAlmostEqual(body["dryMatter"], 25.06, places=2) self.assertAlmostEqual(body["nutrients"]["Сыр. Протеин"], 107.5, places=1) self.assertAlmostEqual(body["nutrients"]["Лизин"], 4.4, places=1) self.assertAlmostEqual(body["nutrients"]["Ca"], 7.3, places=1) def test_apply_dry_run_and_write(self) -> None: xml = _FIXTURE.read_text(encoding="utf-8") parsed = parse_agrostar_xml(xml) sample = parsed.samples[0] assignment = { "sampleNo": sample.sample_no, "componentId": self.comp.id, "dryMatterPct": sample.dry_matter_pct, "nutrients": sample.nutrients, "warnings": sample.warnings, } r = self.client.post( "/api/lab/import/agrostar-xml/apply", json={"assignments": [assignment], "dryRun": True}, ) self.assertEqual(r.status_code, 200) self.assertEqual(r.get_json()["applied"], 1) r2 = self.client.post( "/api/lab/import/agrostar-xml/apply", json={"assignments": [assignment], "dryRun": False}, ) self.assertEqual(r2.status_code, 200) db.session.refresh(self.comp) self.assertAlmostEqual(self.comp.dry_matter, 25.06, places=2) from app.lab.services.component_nutrients import nutrients_api_dict nutrients = nutrients_api_dict(self.comp.id) self.assertAlmostEqual(nutrients["Сыр. Протеин"], 107.5, places=1) if __name__ == "__main__": unittest.main()