218 lines
7.6 KiB
Python
218 lines
7.6 KiB
Python
"""Soft delete рецептов: unlink в периоде, global delete mill, группы, transfer revive."""
|
|
|
|
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,
|
|
Ingredient,
|
|
PeriodRecipe,
|
|
Recipe,
|
|
SyncQueue,
|
|
UnloadingGroup,
|
|
)
|
|
from app.services.recipe_update_service import update_recipe_from_payload
|
|
from config import TestingConfig
|
|
from tests.helpers.mill_recipe_fixtures import (
|
|
create_recipe_with_children,
|
|
group_payload_from_row,
|
|
recipe_update_payload,
|
|
)
|
|
|
|
|
|
class RecipesSoftDeleteConfig(TestingConfig):
|
|
_TMP_DIR = tempfile.mkdtemp(prefix="wesp-recipes-soft-delete-")
|
|
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 = "soft-del-admin"
|
|
AUTH_PASSWORD = "soft-del-secret"
|
|
|
|
|
|
class RecipesSoftDeleteTests(unittest.TestCase):
|
|
def setUp(self) -> None:
|
|
self.app = create_app(RecipesSoftDeleteConfig)
|
|
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": "soft-del-admin", "password": "soft-del-secret"},
|
|
)
|
|
|
|
def tearDown(self) -> None:
|
|
db.session.remove()
|
|
db.drop_all()
|
|
self.ctx.pop()
|
|
|
|
def _seed_dispenser_link(self) -> tuple[str, str, str]:
|
|
disp = FeedDispenser(
|
|
id="sd-disp", name="D", farm="F", operator="O", type="dispenser"
|
|
)
|
|
period = FeedingPeriod(id="sd-period", name="P", dispenser_id=disp.id)
|
|
recipe = Recipe(id="sd-recipe", name="R", heads_per_trip=5, mixing_time=2)
|
|
db.session.add_all([disp, period, recipe])
|
|
db.session.flush()
|
|
db.session.add(
|
|
PeriodRecipe(
|
|
period_id=period.id,
|
|
recipe_id=recipe.id,
|
|
order=0,
|
|
created_at=datetime.now(),
|
|
created_by="system",
|
|
updated_by="system",
|
|
)
|
|
)
|
|
db.session.commit()
|
|
return disp.id, period.id, recipe.id
|
|
|
|
def test_dispenser_delete_unlinks_period_only(self) -> None:
|
|
d_id, p_id, r_id = self._seed_dispenser_link()
|
|
resp = self.client.delete(
|
|
f"/api/feed_dispensers/{d_id}/periods/{p_id}/recipes/{r_id}"
|
|
)
|
|
self.assertEqual(resp.status_code, 200)
|
|
|
|
link = db.session.execute(
|
|
select(PeriodRecipe).where(
|
|
PeriodRecipe.period_id == p_id, PeriodRecipe.recipe_id == r_id
|
|
)
|
|
).scalar_one()
|
|
self.assertTrue(link.is_deleted)
|
|
recipe = db.session.get(Recipe, r_id)
|
|
self.assertFalse(recipe.is_deleted)
|
|
|
|
def test_dispenser_unlink_sync_period_recipes(self) -> None:
|
|
d_id, p_id, r_id = self._seed_dispenser_link()
|
|
self.client.delete(f"/api/feed_dispensers/{d_id}/periods/{p_id}/recipes/{r_id}")
|
|
|
|
tasks = db.session.execute(
|
|
select(SyncQueue)
|
|
.where(
|
|
SyncQueue.table_name == "period_recipes",
|
|
SyncQueue.record_id == f"{p_id}:{r_id}",
|
|
SyncQueue.action.in_(("delete", "update")),
|
|
)
|
|
.order_by(SyncQueue.created_at.desc())
|
|
).scalars().all()
|
|
self.assertGreaterEqual(len(tasks), 1)
|
|
|
|
def test_mill_global_delete_soft_deletes_recipe(self) -> None:
|
|
recipe_id, ing_ids, grp_ids = create_recipe_with_children()
|
|
resp = self.client.delete(f"/api/recipes/{recipe_id}")
|
|
self.assertEqual(resp.status_code, 200)
|
|
|
|
recipe = db.session.get(Recipe, recipe_id)
|
|
self.assertTrue(recipe.is_deleted)
|
|
for ing_id in ing_ids:
|
|
self.assertTrue(db.session.get(Ingredient, ing_id).is_deleted)
|
|
for grp_id in grp_ids:
|
|
self.assertTrue(db.session.get(UnloadingGroup, grp_id).is_deleted)
|
|
|
|
def test_mill_delete_sync_cascade_ingredients_update(self) -> None:
|
|
recipe_id, ing_ids, _ = create_recipe_with_children(with_groups=False)
|
|
self.client.delete(f"/api/recipes/{recipe_id}")
|
|
|
|
ing_tasks = db.session.execute(
|
|
select(SyncQueue).where(
|
|
SyncQueue.table_name == "ingredient",
|
|
SyncQueue.record_id.in_(ing_ids),
|
|
)
|
|
).scalars().all()
|
|
self.assertGreaterEqual(len(ing_tasks), 1)
|
|
|
|
def test_update_deleted_group_ids(self) -> None:
|
|
recipe_id, _, grp_ids = create_recipe_with_children()
|
|
recipe = db.session.get(Recipe, recipe_id)
|
|
keep = db.session.get(UnloadingGroup, grp_ids[1])
|
|
update_recipe_from_payload(
|
|
recipe_id,
|
|
recipe_update_payload(
|
|
recipe,
|
|
ingredients=[],
|
|
groups=[group_payload_from_row(keep, order=1)],
|
|
deleted_group_ids=[grp_ids[0]],
|
|
),
|
|
)
|
|
db.session.commit()
|
|
|
|
self.assertTrue(db.session.get(UnloadingGroup, grp_ids[0]).is_deleted)
|
|
delete_tasks = db.session.execute(
|
|
select(SyncQueue).where(
|
|
SyncQueue.table_name == "unloading_group",
|
|
SyncQueue.record_id == grp_ids[0],
|
|
SyncQueue.action == "delete",
|
|
)
|
|
).scalars().all()
|
|
self.assertGreaterEqual(len(delete_tasks), 1)
|
|
|
|
def test_transfer_revives_soft_deleted_period_recipe(self) -> None:
|
|
d = FeedDispenser(id="sd-xfer-d", name="D", farm="F", operator="O", type="dispenser")
|
|
p1 = FeedingPeriod(id="sd-xfer-p1", name="P1", dispenser_id=d.id)
|
|
p2 = FeedingPeriod(id="sd-xfer-p2", name="P2", dispenser_id=d.id)
|
|
r_move = Recipe(id="sd-xfer-r", name="M", heads_per_trip=1, mixing_time=0)
|
|
db.session.add_all([d, p1, p2, 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.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)
|
|
|
|
revived = db.session.execute(
|
|
select(PeriodRecipe).where(
|
|
PeriodRecipe.period_id == p2.id,
|
|
PeriodRecipe.recipe_id == r_move.id,
|
|
)
|
|
).scalar_one()
|
|
self.assertFalse(revived.is_deleted)
|
|
|
|
def test_deleted_recipe_not_in_period_list(self) -> None:
|
|
d_id, p_id, r_id = self._seed_dispenser_link()
|
|
self.client.delete(f"/api/feed_dispensers/{d_id}/periods/{p_id}/recipes/{r_id}")
|
|
|
|
listed = self.client.get(f"/api/periods/{p_id}/recipes")
|
|
self.assertEqual(listed.status_code, 200)
|
|
self.assertEqual([x["id"] for x in listed.get_json()], [])
|
|
|
|
get_recipe = self.client.get(f"/api/recipes/{r_id}")
|
|
self.assertEqual(get_recipe.status_code, 200)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|