175 lines
5.0 KiB
Python
175 lines
5.0 KiB
Python
"""Фабрики данных для тестов кормоцеха и рецептов."""
|
||
|
||
from __future__ import annotations
|
||
|
||
import uuid
|
||
from datetime import datetime
|
||
from typing import Any, Dict, List, Optional, Tuple
|
||
|
||
from sqlalchemy import func, select
|
||
|
||
from app import db
|
||
from app.models import Component, Ingredient, Recipe, SyncQueue, UnloadingGroup
|
||
|
||
|
||
def create_component(*, name: str = "Компонент А", dry_matter: float = 50.0) -> str:
|
||
comp_id = str(uuid.uuid4())
|
||
db.session.add(
|
||
Component(
|
||
id=comp_id,
|
||
name=name,
|
||
type="grain",
|
||
dry_matter=dry_matter,
|
||
protein=0.0,
|
||
energy=0.0,
|
||
price=0.0,
|
||
)
|
||
)
|
||
db.session.flush()
|
||
return comp_id
|
||
|
||
|
||
def create_recipe_with_children(
|
||
*,
|
||
name: str = "Тестовый рецепт",
|
||
heads: int = 10,
|
||
component_ids: Optional[List[str]] = None,
|
||
with_groups: bool = True,
|
||
) -> Tuple[str, List[str], List[str]]:
|
||
"""Рецепт + ингредиенты (+ опционально группы). Возвращает recipe_id, ingredient_ids, group_ids."""
|
||
if component_ids is None:
|
||
component_ids = [create_component(name=f"К{i + 1}") for i in range(2)]
|
||
|
||
recipe_id = str(uuid.uuid4())
|
||
recipe = Recipe(
|
||
id=recipe_id,
|
||
name=name,
|
||
heads_per_trip=heads,
|
||
mixing_time=5,
|
||
trip_percent=100.0,
|
||
target_component_id=component_ids[0],
|
||
content_hash="",
|
||
)
|
||
db.session.add(recipe)
|
||
db.session.flush()
|
||
|
||
ingredient_ids: List[str] = []
|
||
for idx, comp_id in enumerate(component_ids, 1):
|
||
ing = Ingredient(
|
||
name=f"Ing {idx}",
|
||
weight_per_head=float(idx),
|
||
amount=float(idx * heads),
|
||
dry_matter=50.0 + idx,
|
||
component_id=comp_id,
|
||
order=idx,
|
||
recipe_id=recipe_id,
|
||
created_by="system",
|
||
updated_by="system",
|
||
)
|
||
db.session.add(ing)
|
||
db.session.flush()
|
||
ingredient_ids.append(str(ing.id))
|
||
|
||
group_ids: List[str] = []
|
||
if with_groups:
|
||
for idx in range(1, 3):
|
||
grp = UnloadingGroup(
|
||
name=f"Г{idx}",
|
||
distribution_type="percent",
|
||
value=50.0,
|
||
weight=float(idx * 10),
|
||
order=idx,
|
||
recipe_id=recipe_id,
|
||
created_by="system",
|
||
updated_by="system",
|
||
)
|
||
db.session.add(grp)
|
||
db.session.flush()
|
||
group_ids.append(str(grp.id))
|
||
|
||
db.session.commit()
|
||
return recipe_id, ingredient_ids, group_ids
|
||
|
||
|
||
def recipe_update_payload(
|
||
recipe: Recipe,
|
||
*,
|
||
ingredients: List[Dict[str, Any]],
|
||
groups: Optional[List[Dict[str, Any]]] = None,
|
||
deleted_ingredient_ids: Optional[List[str]] = None,
|
||
deleted_group_ids: Optional[List[str]] = None,
|
||
target_component_id: Optional[str] = None,
|
||
) -> Dict[str, Any]:
|
||
payload: Dict[str, Any] = {
|
||
"name": recipe.name,
|
||
"heads_count": recipe.heads_per_trip,
|
||
"mixing_time": recipe.mixing_time,
|
||
"trip_percent": recipe.trip_percent,
|
||
"target_component_id": target_component_id or recipe.target_component_id,
|
||
"dry_matter_locked": False,
|
||
"unloading_link_broken": False,
|
||
"ingredients": ingredients,
|
||
"unloading_groups": groups or [],
|
||
}
|
||
if deleted_ingredient_ids:
|
||
payload["deleted_ingredient_ids"] = deleted_ingredient_ids
|
||
if deleted_group_ids:
|
||
payload["deleted_unloading_group_ids"] = deleted_group_ids
|
||
return payload
|
||
|
||
|
||
def sync_task_exists(
|
||
table_name: str,
|
||
record_id: str,
|
||
action: str,
|
||
) -> bool:
|
||
row = db.session.execute(
|
||
select(SyncQueue).where(
|
||
SyncQueue.table_name == table_name,
|
||
SyncQueue.record_id == record_id,
|
||
SyncQueue.action == action,
|
||
)
|
||
).scalar_one_or_none()
|
||
return row is not None
|
||
|
||
|
||
def count_sync_tasks(
|
||
table_name: str,
|
||
record_id: str,
|
||
action: str,
|
||
) -> int:
|
||
return int(
|
||
db.session.scalar(
|
||
select(func.count())
|
||
.select_from(SyncQueue)
|
||
.where(
|
||
SyncQueue.table_name == table_name,
|
||
SyncQueue.record_id == record_id,
|
||
SyncQueue.action == action,
|
||
)
|
||
)
|
||
or 0
|
||
)
|
||
|
||
|
||
def ingredient_payload_from_row(ing: Ingredient, *, order: Optional[int] = None) -> Dict[str, Any]:
|
||
return {
|
||
"id": str(ing.id),
|
||
"component_id": str(ing.component_id),
|
||
"weight_per_head": ing.weight_per_head,
|
||
"amount": ing.amount,
|
||
"dry_matter": ing.dry_matter,
|
||
"order": order if order is not None else ing.order,
|
||
}
|
||
|
||
|
||
def group_payload_from_row(grp: UnloadingGroup, *, order: Optional[int] = None) -> Dict[str, Any]:
|
||
return {
|
||
"id": str(grp.id),
|
||
"name": grp.name,
|
||
"distribution_type": grp.distribution_type,
|
||
"value": grp.value,
|
||
"weight": grp.weight,
|
||
"order": order if order is not None else grp.order,
|
||
}
|