194 lines
7.4 KiB
Python
194 lines
7.4 KiB
Python
"""Перенос рейса между периодами (POST .../transfer)."""
|
||
|
||
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 RecipeTransferConfig(TestingConfig):
|
||
_TMP_DIR = tempfile.mkdtemp(prefix="wesp-recipe-transfer-")
|
||
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 = "xfer-admin"
|
||
AUTH_PASSWORD = "xfer-secret"
|
||
|
||
|
||
class RecipePeriodTransferTests(unittest.TestCase):
|
||
def setUp(self) -> None:
|
||
self.app = create_app(RecipeTransferConfig)
|
||
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": "xfer-admin", "password": "xfer-secret"}
|
||
)
|
||
self.assertEqual(login.status_code, 200)
|
||
|
||
def tearDown(self) -> None:
|
||
db.session.remove()
|
||
db.drop_all()
|
||
self.ctx.pop()
|
||
|
||
def _seed_two_periods(self) -> tuple[str, str, str, str, str]:
|
||
d1 = FeedDispenser(
|
||
id="d-xfer-1", name="D1", farm="F", operator="O", type="dispenser"
|
||
)
|
||
d2 = FeedDispenser(
|
||
id="d-xfer-2", name="D2", farm="F", operator="O", type="dispenser"
|
||
)
|
||
p1 = FeedingPeriod(id="p-xfer-1", name="P1", dispenser_id=d1.id)
|
||
p2 = FeedingPeriod(id="p-xfer-2", name="P2", dispenser_id=d2.id)
|
||
r_a = Recipe(id="r-xfer-a", name="A", heads_per_trip=1, mixing_time=0)
|
||
r_b = Recipe(id="r-xfer-b", name="B", heads_per_trip=1, mixing_time=0)
|
||
r_move = Recipe(id="r-xfer-move", name="Move", heads_per_trip=1, mixing_time=0)
|
||
db.session.add_all([d1, d2, p1, p2, r_a, r_b, r_move])
|
||
db.session.flush()
|
||
now = datetime.now()
|
||
for pid, rid, ord_ in (
|
||
("p-xfer-1", "r-xfer-a", 1),
|
||
("p-xfer-1", "r-xfer-move", 2),
|
||
("p-xfer-2", "r-xfer-b", 1),
|
||
):
|
||
db.session.add(
|
||
PeriodRecipe(
|
||
period_id=pid,
|
||
recipe_id=rid,
|
||
order=ord_,
|
||
created_at=now,
|
||
created_by="system",
|
||
updated_by="system",
|
||
)
|
||
)
|
||
db.session.commit()
|
||
return "d-xfer-1", "p-xfer-1", "d-xfer-2", "p-xfer-2", "r-xfer-move"
|
||
|
||
def test_transfer_inserts_at_index_and_removes_from_source(self) -> None:
|
||
d1, p1, d2, p2, r_move = self._seed_two_periods()
|
||
|
||
resp = self.client.post(
|
||
f"/api/feed_dispensers/{d2}/periods/{p2}/recipes/{r_move}/transfer",
|
||
json={"from_dispenser_id": d1, "from_period_id": p1, "to_index": 0},
|
||
)
|
||
self.assertEqual(resp.status_code, 200, resp.get_json())
|
||
self.assertTrue(resp.get_json().get("success"))
|
||
|
||
p1_list = self.client.get(f"/api/periods/{p1}/recipes").get_json()
|
||
p2_list = self.client.get(f"/api/periods/{p2}/recipes").get_json()
|
||
self.assertEqual([x["id"] for x in p1_list], ["r-xfer-a"])
|
||
self.assertEqual([x["id"] for x in p2_list], ["r-xfer-move", "r-xfer-b"])
|
||
|
||
def test_transfer_same_period_rejected(self) -> None:
|
||
d1, p1, _, _, r_move = self._seed_two_periods()
|
||
resp = self.client.post(
|
||
f"/api/feed_dispensers/{d1}/periods/{p1}/recipes/{r_move}/transfer",
|
||
json={"from_dispenser_id": d1, "from_period_id": p1, "to_index": 0},
|
||
)
|
||
self.assertEqual(resp.status_code, 400)
|
||
|
||
def test_transfer_duplicate_in_target_409(self) -> None:
|
||
d1, p1, d2, p2, r_move = self._seed_two_periods()
|
||
# Тот же рейс уже привязан к p2 (дубликат цели)
|
||
db.session.add(
|
||
PeriodRecipe(
|
||
period_id=p2,
|
||
recipe_id=r_move,
|
||
order=2,
|
||
created_at=datetime.now(),
|
||
created_by="system",
|
||
updated_by="system",
|
||
)
|
||
)
|
||
db.session.commit()
|
||
|
||
resp = self.client.post(
|
||
f"/api/feed_dispensers/{d2}/periods/{p2}/recipes/{r_move}/transfer",
|
||
json={"from_dispenser_id": d1, "from_period_id": p1, "to_index": 0},
|
||
)
|
||
self.assertEqual(resp.status_code, 409)
|
||
|
||
def test_transfer_revives_soft_deleted_row_same_dispenser(self) -> None:
|
||
"""Раньше в целевом периоде удалили рейс (строка period_recipes с is_deleted) — INSERT давал UNIQUE."""
|
||
d = FeedDispenser(
|
||
id="d-xfer-sd", name="D", farm="F", operator="O", type="dispenser"
|
||
)
|
||
p1 = FeedingPeriod(id="p-xfer-sd-1", name="P1", dispenser_id=d.id)
|
||
p2 = FeedingPeriod(id="p-xfer-sd-2", name="P2", dispenser_id=d.id)
|
||
r_keep = Recipe(id="r-xfer-sd-k", name="K", heads_per_trip=1, mixing_time=0)
|
||
r_move = Recipe(id="r-xfer-sd-m", name="M", heads_per_trip=1, mixing_time=0)
|
||
db.session.add_all([d, p1, p2, r_keep, r_move])
|
||
db.session.flush()
|
||
now = datetime.now()
|
||
db.session.add(
|
||
PeriodRecipe(
|
||
period_id=p1.id,
|
||
recipe_id=r_move.id,
|
||
order=1,
|
||
created_at=now,
|
||
created_by="system",
|
||
updated_by="system",
|
||
)
|
||
)
|
||
db.session.add(
|
||
PeriodRecipe(
|
||
period_id=p2.id,
|
||
recipe_id=r_move.id,
|
||
order=1,
|
||
created_at=now,
|
||
created_by="system",
|
||
updated_by="system",
|
||
is_deleted=True,
|
||
deleted_at=now,
|
||
deleted_by="test",
|
||
)
|
||
)
|
||
db.session.add(
|
||
PeriodRecipe(
|
||
period_id=p2.id,
|
||
recipe_id=r_keep.id,
|
||
order=2,
|
||
created_at=now,
|
||
created_by="system",
|
||
updated_by="system",
|
||
)
|
||
)
|
||
db.session.commit()
|
||
|
||
resp = self.client.post(
|
||
f"/api/feed_dispensers/{d.id}/periods/{p2.id}/recipes/{r_move.id}/transfer",
|
||
json={
|
||
"from_dispenser_id": d.id,
|
||
"from_period_id": p1.id,
|
||
"to_index": 0,
|
||
},
|
||
)
|
||
self.assertEqual(resp.status_code, 200, resp.get_json())
|
||
p1_list = self.client.get(f"/api/periods/{p1.id}/recipes").get_json()
|
||
p2_list = self.client.get(f"/api/periods/{p2.id}/recipes").get_json()
|
||
self.assertEqual(p1_list, [])
|
||
self.assertEqual([x["id"] for x in p2_list], [r_move.id, r_keep.id])
|
||
|
||
def test_transfer_to_mill_rejected(self) -> None:
|
||
d1, p1, _, _, r_move = self._seed_two_periods()
|
||
mill = FeedDispenser(
|
||
id="d-mill-xfer", name="Mill", farm="F", operator="O", type="mill"
|
||
)
|
||
pm = FeedingPeriod(id="p-mill-xfer", name="Px", dispenser_id=mill.id)
|
||
db.session.add_all([mill, pm])
|
||
db.session.commit()
|
||
|
||
resp = self.client.post(
|
||
f"/api/feed_dispensers/d-mill-xfer/periods/p-mill-xfer/recipes/{r_move}/transfer",
|
||
json={"from_dispenser_id": d1, "from_period_id": p1, "to_index": 0},
|
||
)
|
||
self.assertEqual(resp.status_code, 400)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
unittest.main()
|