204 lines
7.2 KiB
Python
204 lines
7.2 KiB
Python
"""API границы /recipes: пустой период, max order, transfer enqueue, repeat."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
import tempfile
|
|
import unittest
|
|
from datetime import datetime
|
|
|
|
from sqlalchemy import select
|
|
|
|
from app import create_app, db
|
|
from app.models import FeedDispenser, FeedingPeriod, PeriodRecipe, Recipe, SyncQueue
|
|
from config import TestingConfig
|
|
|
|
SCENARIO_HAPPY_ONE_RECIPE_MOVE = "HAPPY_ONE_RECIPE_MOVE"
|
|
SCENARIO_EDGE_EMPTY_PERIOD_TRANSFER = "EDGE_EMPTY_PERIOD_TRANSFER"
|
|
SCENARIO_EDGE_MAX_RECIPES_ORDER = "EDGE_MAX_RECIPES_ORDER"
|
|
SCENARIO_TRANSFER_ENQUEUE_SOURCE_DELETE = "TRANSFER_ENQUEUE_SOURCE_DELETE"
|
|
SCENARIO_TRANSFER_REPEAT_A_B_A = "TRANSFER_REPEAT_A_B_A"
|
|
|
|
|
|
class RecipesApiEdgesConfig(TestingConfig):
|
|
_TMP_DIR = tempfile.mkdtemp(prefix="wesp-recipes-api-edges-")
|
|
SQLALCHEMY_DATABASE_URI = f"sqlite:///{os.path.join(_TMP_DIR, 'test.db')}"
|
|
SQLALCHEMY_BINDS = {"reports": f"sqlite:///{os.path.join(_TMP_DIR, 'reports.db')}"}
|
|
AUTH_LOGIN = "edges-admin"
|
|
AUTH_PASSWORD = "edges-secret"
|
|
|
|
|
|
class RecipesApiEdgesTests(unittest.TestCase):
|
|
def setUp(self) -> None:
|
|
self.app = create_app(RecipesApiEdgesConfig)
|
|
self.client = self.app.test_client()
|
|
self.ctx = self.app.app_context()
|
|
self.ctx.push()
|
|
db.create_all()
|
|
self.client.post(
|
|
"/api/auth/login", json={"login": "edges-admin", "password": "edges-secret"}
|
|
)
|
|
|
|
def tearDown(self) -> None:
|
|
db.session.remove()
|
|
db.drop_all()
|
|
self.ctx.pop()
|
|
|
|
def _seed_dispenser_periods(self) -> tuple[str, str, str]:
|
|
disp = FeedDispenser(
|
|
id="edge-disp", name="D", farm="F", operator="O", type="dispenser"
|
|
)
|
|
pa = FeedingPeriod(id="edge-pa", name="A", dispenser_id=disp.id)
|
|
pb = FeedingPeriod(id="edge-pb", name="B", dispenser_id=disp.id)
|
|
db.session.add_all([disp, pa, pb])
|
|
db.session.commit()
|
|
return disp.id, pa.id, pb.id
|
|
|
|
def _link(self, period_id: str, recipe_id: str, order: int) -> None:
|
|
db.session.add(
|
|
PeriodRecipe(
|
|
period_id=period_id,
|
|
recipe_id=recipe_id,
|
|
order=order,
|
|
created_at=datetime.now(),
|
|
created_by="system",
|
|
updated_by="system",
|
|
)
|
|
)
|
|
|
|
def _recipe(self, rid: str) -> Recipe:
|
|
r = Recipe(id=rid, name=rid, heads_per_trip=1, mixing_time=0)
|
|
db.session.add(r)
|
|
return r
|
|
|
|
def _period_order(self, period_id: str) -> list[str]:
|
|
rows = db.session.execute(
|
|
select(PeriodRecipe.recipe_id)
|
|
.where(
|
|
PeriodRecipe.period_id == period_id,
|
|
PeriodRecipe.is_deleted.is_(False),
|
|
)
|
|
.order_by(PeriodRecipe.order.asc())
|
|
).scalars().all()
|
|
return [str(x) for x in rows]
|
|
|
|
def test_happy_move_up_down_single_recipe_pair(self) -> None:
|
|
SCENARIO_HAPPY_ONE_RECIPE_MOVE # noqa: F841
|
|
_, period_id, _ = self._seed_dispenser_periods()
|
|
self._recipe("edge-r1")
|
|
self._recipe("edge-r2")
|
|
self._link(period_id, "edge-r1", 0)
|
|
self._link(period_id, "edge-r2", 1)
|
|
db.session.commit()
|
|
|
|
self.client.put(
|
|
f"/api/recipes/edge-r1/move",
|
|
json={"period_id": period_id, "from_index": 0, "to_index": 1},
|
|
)
|
|
self.assertEqual(self._period_order(period_id), ["edge-r2", "edge-r1"])
|
|
|
|
self.client.put(
|
|
f"/api/recipes/edge-r1/move",
|
|
json={"period_id": period_id, "from_index": 1, "to_index": 0},
|
|
)
|
|
self.assertEqual(self._period_order(period_id), ["edge-r1", "edge-r2"])
|
|
|
|
def test_transfer_into_empty_period_index_zero(self) -> None:
|
|
SCENARIO_EDGE_EMPTY_PERIOD_TRANSFER # noqa: F841
|
|
disp_id, pa, pb = self._seed_dispenser_periods()
|
|
self._recipe("edge-move")
|
|
self._link(pa, "edge-move", 0)
|
|
db.session.commit()
|
|
|
|
resp = self.client.post(
|
|
f"/api/feed_dispensers/{disp_id}/periods/{pb}/recipes/edge-move/transfer",
|
|
json={
|
|
"from_dispenser_id": disp_id,
|
|
"from_period_id": pa,
|
|
"to_index": 0,
|
|
},
|
|
)
|
|
self.assertEqual(resp.status_code, 200, resp.get_json())
|
|
self.assertEqual(self._period_order(pa), [])
|
|
self.assertEqual(self._period_order(pb), ["edge-move"])
|
|
|
|
def test_max_ten_recipes_move_extremes(self) -> None:
|
|
SCENARIO_EDGE_MAX_RECIPES_ORDER # noqa: F841
|
|
_, period_id, _ = self._seed_dispenser_periods()
|
|
ids = [f"edge-max-{i}" for i in range(10)]
|
|
for rid in ids:
|
|
self._recipe(rid)
|
|
for ord_, rid in enumerate(ids):
|
|
self._link(period_id, rid, ord_)
|
|
db.session.commit()
|
|
|
|
self.client.put(
|
|
f"/api/recipes/{ids[0]}/move",
|
|
json={"period_id": period_id, "from_index": 0, "to_index": 9},
|
|
)
|
|
order = self._period_order(period_id)
|
|
self.assertEqual(order[-1], ids[0])
|
|
self.assertEqual(order[0], ids[1])
|
|
|
|
self.client.put(
|
|
f"/api/recipes/{ids[0]}/move",
|
|
json={"period_id": period_id, "from_index": 9, "to_index": 0},
|
|
)
|
|
self.assertEqual(self._period_order(period_id), ids)
|
|
|
|
def test_transfer_enqueues_source_period_recipes_delete(self) -> None:
|
|
SCENARIO_TRANSFER_ENQUEUE_SOURCE_DELETE # noqa: F841
|
|
disp_id, pa, pb = self._seed_dispenser_periods()
|
|
self._recipe("edge-xfer")
|
|
self._link(pa, "edge-xfer", 0)
|
|
db.session.commit()
|
|
|
|
resp = self.client.post(
|
|
f"/api/feed_dispensers/{disp_id}/periods/{pb}/recipes/edge-xfer/transfer",
|
|
json={
|
|
"from_dispenser_id": disp_id,
|
|
"from_period_id": pa,
|
|
"to_index": 0,
|
|
},
|
|
)
|
|
self.assertEqual(resp.status_code, 200, resp.get_json())
|
|
|
|
tasks = db.session.execute(
|
|
select(SyncQueue).where(
|
|
SyncQueue.table_name == "period_recipes",
|
|
SyncQueue.record_id == f"{pa}:edge-xfer",
|
|
SyncQueue.action == "delete",
|
|
)
|
|
).scalars().all()
|
|
self.assertGreaterEqual(len(tasks), 1)
|
|
|
|
def test_transfer_repeat_a_b_a(self) -> None:
|
|
SCENARIO_TRANSFER_REPEAT_A_B_A # noqa: F841
|
|
disp_id, pa, pb = self._seed_dispenser_periods()
|
|
self._recipe("edge-bounce")
|
|
self._link(pa, "edge-bounce", 0)
|
|
db.session.commit()
|
|
|
|
def xfer(to_period: str, from_period: str, to_index: int = 0) -> None:
|
|
r = self.client.post(
|
|
f"/api/feed_dispensers/{disp_id}/periods/{to_period}/recipes/edge-bounce/transfer",
|
|
json={
|
|
"from_dispenser_id": disp_id,
|
|
"from_period_id": from_period,
|
|
"to_index": to_index,
|
|
},
|
|
)
|
|
self.assertEqual(r.status_code, 200, r.get_json())
|
|
|
|
xfer(pb, pa, 0)
|
|
self.assertEqual(self._period_order(pa), [])
|
|
self.assertEqual(self._period_order(pb), ["edge-bounce"])
|
|
|
|
xfer(pa, pb, 0)
|
|
self.assertEqual(self._period_order(pa), ["edge-bounce"])
|
|
self.assertEqual(self._period_order(pb), [])
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|