import os import tempfile import unittest from datetime import datetime from app import create_app, db from app.models import FeedDispenser, FeedingPeriod, PeriodRecipe, Recipe from config import TestingConfig class LegacySecondBatchConfig(TestingConfig): _TMP_DIR = tempfile.mkdtemp(prefix="wesp-legacy-batch2-tests-") SQLALCHEMY_DATABASE_URI = f"sqlite:///{os.path.join(_TMP_DIR, 'recipes_test.db')}" SQLALCHEMY_BINDS = {"reports": f"sqlite:///{os.path.join(_TMP_DIR, 'reports_test.db')}"} AUTH_LOGIN = "legacy2-admin" AUTH_PASSWORD = "legacy2-secret" class LegacyCompatSecondBatchTests(unittest.TestCase): def setUp(self) -> None: self.app = create_app(LegacySecondBatchConfig) self.client = self.app.test_client() self.ctx = self.app.app_context() self.ctx.push() db.create_all() login = self.client.post( "/api/auth/login", json={"login": "legacy2-admin", "password": "legacy2-secret"} ) self.assertEqual(login.status_code, 200) def tearDown(self) -> None: db.session.remove() db.drop_all() self.ctx.pop() def test_feed_dispenser_periods_and_recipes_endpoints(self) -> None: disp = FeedDispenser(id="d-1", name="D1", farm="F", operator="O", type="dispenser") period = FeedingPeriod(id="p-1", name="Morning", dispenser_id=disp.id) r1 = Recipe(id="r-1", name="Route 1", heads_per_trip=10, mixing_time=5, trip_percent=100) db.session.add_all([disp, period, r1]) db.session.flush() db.session.add( PeriodRecipe( period_id=period.id, recipe_id=r1.id, order=1, created_at=datetime.now() ) ) db.session.commit() periods_resp = self.client.get("/api/feed_dispensers/d-1/periods") self.assertEqual(periods_resp.status_code, 200) periods = periods_resp.get_json() self.assertIsInstance(periods, list) self.assertEqual(periods[0]["id"], "p-1") self.assertEqual(len(periods[0]["recipes"]), 1) recipes_resp = self.client.get("/api/feed_dispensers/d-1/recipes") self.assertEqual(recipes_resp.status_code, 200) recipes = recipes_resp.get_json() self.assertIsInstance(recipes, list) self.assertEqual(recipes[0]["id"], "r-1") def test_period_recipes_get_post_and_move(self) -> None: disp = FeedDispenser(id="d-2", name="D2", farm="F", operator="O", type="dispenser") period = FeedingPeriod(id="p-2", name="Evening", dispenser_id=disp.id) r1 = Recipe(id="r-21", name="Route 21", heads_per_trip=11, mixing_time=6, trip_percent=90) r2 = Recipe(id="r-22", name="Route 22", heads_per_trip=12, mixing_time=7, trip_percent=95) db.session.add_all([disp, period, r1, r2]) db.session.flush() db.session.add_all( [ PeriodRecipe( period_id=period.id, recipe_id=r1.id, order=1, created_at=datetime.now(), ), PeriodRecipe( period_id=period.id, recipe_id=r2.id, order=2, created_at=datetime.now(), ), ] ) db.session.commit() list_before = self.client.get("/api/periods/p-2/recipes") self.assertEqual(list_before.status_code, 200) before = list_before.get_json() self.assertEqual([x["id"] for x in before], ["r-21", "r-22"]) move = self.client.put( "/api/recipes/r-22/move", json={"period_id": "p-2", "from_index": 1, "to_index": 0}, ) self.assertEqual(move.status_code, 200) list_after = self.client.get("/api/periods/p-2/recipes") self.assertEqual(list_after.status_code, 200) after = list_after.get_json() self.assertEqual([x["id"] for x in after], ["r-22", "r-21"]) # POST attach existing recipe r3 = Recipe(id="r-23", name="Route 23", heads_per_trip=13, mixing_time=8, trip_percent=100) db.session.add(r3) db.session.commit() attach = self.client.post("/api/periods/p-2/recipes", json={"recipe_id": "r-23"}) self.assertIn(attach.status_code, (200, 201)) final_list = self.client.get("/api/periods/p-2/recipes").get_json() self.assertEqual([x["id"] for x in final_list], ["r-22", "r-21", "r-23"]) def test_period_recipes_soft_deleted_link_hidden_from_dispenser_periods(self) -> None: disp = FeedDispenser(id="d-sd", name="D", farm="F", operator="O", type="dispenser") period = FeedingPeriod(id="p-sd", name="P", dispenser_id=disp.id, is_active=True) r_ok = Recipe( id="r-ok", name="OK", heads_per_trip=1, mixing_time=1, trip_percent=100 ) r_gone = Recipe( id="r-gone", name="Gone", heads_per_trip=1, mixing_time=1, trip_percent=100 ) db.session.add_all([disp, period, r_ok, r_gone]) db.session.flush() db.session.add_all( [ PeriodRecipe( period_id=period.id, recipe_id=r_ok.id, order=1, created_at=datetime.now(), ), PeriodRecipe( period_id=period.id, recipe_id=r_gone.id, order=2, created_at=datetime.now(), is_deleted=True, ), ] ) db.session.commit() periods_resp = self.client.get("/api/feed_dispensers/d-sd/periods") self.assertEqual(periods_resp.status_code, 200) pdata = periods_resp.get_json() self.assertEqual(len(pdata), 1) self.assertEqual([x["id"] for x in pdata[0]["recipes"]], ["r-ok"]) one = self.client.get("/api/feed_dispensers/d-sd") self.assertEqual(one.status_code, 200) body = one.get_json() self.assertEqual(len(body["periods"]), 1) self.assertEqual([x["id"] for x in body["periods"][0]["recipes"]], ["r-ok"]) def test_inactive_period_hidden_from_dispenser_periods(self) -> None: disp = FeedDispenser(id="d-in", name="D", farm="F", operator="O", type="dispenser") p_on = FeedingPeriod(id="p-on", name="On", dispenser_id=disp.id, is_active=True) p_off = FeedingPeriod( id="p-off", name="Off", dispenser_id=disp.id, is_active=False ) r1 = Recipe(id="r-in1", name="A", heads_per_trip=1, mixing_time=1, trip_percent=100) db.session.add_all([disp, p_on, p_off, r1]) db.session.flush() db.session.add( PeriodRecipe( period_id=p_on.id, recipe_id=r1.id, order=1, created_at=datetime.now() ) ) db.session.commit() periods_resp = self.client.get("/api/feed_dispensers/d-in/periods") self.assertEqual(periods_resp.status_code, 200) ids = [p["id"] for p in periods_resp.get_json()] self.assertEqual(ids, ["p-on"]) def test_mill_dispenser_recipes_excludes_period_linked_recipes(self) -> None: """Как legacy: GET .../recipes для кормоцеха — только рецепты без period_recipes.""" mill = FeedDispenser( id="d-mill", name="Mill", farm="F", operator="O", type="mill" ) disp = FeedDispenser( id="d-disp", name="D", farm="F", operator="O", type="dispenser" ) period = FeedingPeriod(id="p-mill", name="P", dispenser_id=disp.id) r_orphan = Recipe( id="r-orphan", name="Orphan", heads_per_trip=1, mixing_time=1, trip_percent=100 ) r_linked = Recipe( id="r-linked", name="Linked", heads_per_trip=10, mixing_time=5, trip_percent=100 ) db.session.add_all([mill, disp, period, r_orphan, r_linked]) db.session.flush() db.session.add( PeriodRecipe( period_id=period.id, recipe_id=r_linked.id, order=1, created_at=datetime.now(), ) ) db.session.commit() mill_resp = self.client.get("/api/feed_dispensers/d-mill/recipes") self.assertEqual(mill_resp.status_code, 200) mill_ids = [x["id"] for x in mill_resp.get_json()] self.assertEqual(mill_ids, ["r-orphan"]) disp_resp = self.client.get("/api/feed_dispensers/d-disp/recipes") self.assertEqual(disp_resp.status_code, 200) disp_ids = [x["id"] for x in disp_resp.get_json()] self.assertEqual(disp_ids, ["r-linked"]) if __name__ == "__main__": unittest.main()