"""Sync с двумя независимыми терминалами: pull → apply → confirm на обоих.""" from __future__ import annotations import time import unittest import uuid from datetime import datetime from sqlalchemy import select from app import db from app.models import ( FeedDispenser, FeedingPeriod, Ingredient, PeriodRecipe, Recipe, SyncQueue, ) from app.services.recipe_update_service import update_recipe_from_payload from app.services.sync_manager import enqueue_sync_queue_task from tests.helpers.mill_recipe_fixtures import ( create_recipe_with_children, ingredient_payload_from_row, recipe_update_payload, ) from tests.helpers.sync_dual_harness import SyncDualInstanceHarness class SyncDualInstanceTests(unittest.TestCase): def setUp(self) -> None: self.harness = SyncDualInstanceHarness() self.harness.start() def tearDown(self) -> None: self.harness.stop() def test_recipe_create_reaches_both_terminals(self) -> None: with self.harness.server_ctx(): recipe_id, _, _ = create_recipe_with_children(name="Dual create") self.harness.drain_sync() for term in ("a", "b"): row = self.harness.recipe_on_terminal(term, recipe_id) self.assertIsNotNone(row, msg=f"terminal {term}") self.assertEqual(row.name, "Dual create") def test_ingredient_delete_reaches_both_terminals(self) -> None: with self.harness.server_ctx(): recipe_id, ing_ids, _ = create_recipe_with_children(with_groups=False) recipe = db.session.get(Recipe, recipe_id) remove_id, keep_id = ing_ids[0], ing_ids[1] keep_row = db.session.get(Ingredient, keep_id) update_recipe_from_payload( recipe_id, recipe_update_payload( recipe, deleted_ingredient_ids=[remove_id], ingredients=[ingredient_payload_from_row(keep_row, order=1)], groups=[], ), ) db.session.commit() self.harness.drain_sync() for term in ("a", "b"): active = self.harness.active_ingredient_ids(term, recipe_id) self.assertEqual(active, [keep_id], msg=f"terminal {term}") with self.harness.terminal_ctx(term): deleted = db.session.get(Ingredient, remove_id) self.assertTrue(deleted.is_deleted) def test_recipe_update_reaches_both_terminals(self) -> None: with self.harness.server_ctx(): recipe_id, _, _ = create_recipe_with_children(name="Before update") recipe = db.session.get(Recipe, recipe_id) update_recipe_from_payload( recipe_id, recipe_update_payload( recipe, ingredients=[], groups=[], ) | {"name": "After dual sync"}, ) db.session.commit() self.harness.drain_sync() for term in ("a", "b"): row = self.harness.recipe_on_terminal(term, recipe_id) self.assertEqual(row.name, "After dual sync", msg=f"terminal {term}") def test_mill_global_delete_reaches_both_terminals(self) -> None: with self.harness.server_ctx(): recipe_id, ing_ids, _ = create_recipe_with_children(with_groups=False) client = self.harness.server_test_client() resp = client.delete(f"/api/recipes/{recipe_id}") self.assertEqual(resp.status_code, 200) self.harness.drain_sync() for term in ("a", "b"): with self.harness.terminal_ctx(term): recipe = db.session.get(Recipe, recipe_id) self.assertIsNotNone(recipe) self.assertTrue(recipe.is_deleted) for ing_id in ing_ids: ing = db.session.get(Ingredient, ing_id) self.assertTrue(ing.is_deleted, msg=f"ing {ing_id} term {term}") def test_period_recipes_reorder_reaches_both_terminals(self) -> None: disp_id = "dual-disp-1" period_id = "dual-period-1" r1, r2 = "dual-r1", "dual-r2" with self.harness.server_ctx(): db.session.add_all( [ FeedDispenser( id=disp_id, name="D", farm="F", operator="O", type="dispenser", content_hash="", ), FeedingPeriod(id=period_id, name="P", dispenser_id=disp_id), Recipe(id=r1, name="R1", heads_per_trip=1, mixing_time=0, content_hash=""), Recipe(id=r2, name="R2", heads_per_trip=1, mixing_time=0, content_hash=""), ] ) db.session.flush() now = datetime.now() for rid, ord_ in ((r1, 1), (r2, 2)): db.session.add( PeriodRecipe( period_id=period_id, recipe_id=rid, order=ord_, created_at=now, created_by="system", updated_by="system", ) ) db.session.commit() for rid in (r1, r2): enqueue_sync_queue_task("recipe", rid, "create", priority=2, target_node_id=None) enqueue_sync_queue_task( "period_recipes", f"{period_id}:{r1}", "create", priority=2, target_node_id=None ) enqueue_sync_queue_task( "period_recipes", f"{period_id}:{r2}", "create", priority=2, target_node_id=None ) db.session.commit() self.harness.drain_sync() client = self.harness.server_test_client() move = client.put( f"/api/recipes/{r1}/move", json={"period_id": period_id, "from_index": 0, "to_index": 1}, ) self.assertEqual(move.status_code, 200, move.get_json()) time.sleep(0.25) self.harness.drain_sync() for term in ("a", "b"): order = self.harness.period_recipe_order(term, period_id) self.assertEqual(order, [r2, r1], msg=f"terminal {term}") def test_confirm_waits_for_both_terminals(self) -> None: with self.harness.server_ctx(): recipe_id, _, _ = create_recipe_with_children(name="Confirm gate") task = self.harness.latest_task("recipe", recipe_id, "create") self.assertIsNotNone(task) task_id = task.id self.harness.sync_terminal("a") self.assertEqual(self.harness.task_status(task_id), "processing") self.harness.sync_terminal("b") self.assertEqual(self.harness.task_status(task_id), "completed") def test_late_join_terminal_catches_up(self) -> None: with self.harness.server_ctx(): recipe_id, _, _ = create_recipe_with_children(name="Late join") self.harness.sync_terminal("a") self.harness.sync_terminal("a") row_b_before = self.harness.recipe_on_terminal("b", recipe_id) self.assertIsNone(row_b_before) self.harness.sync_terminal("b") self.harness.sync_terminal("b") row_b_after = self.harness.recipe_on_terminal("b", recipe_id) self.assertIsNotNone(row_b_after) self.assertEqual(row_b_after.name, "Late join") def test_offline_terminal_catches_up_after_changes(self) -> None: with self.harness.server_ctx(): recipe_id, ing_ids, _ = create_recipe_with_children(with_groups=False) recipe = db.session.get(Recipe, recipe_id) self.harness.sync_both(rounds=2) with self.harness.server_ctx(): remove_id = ing_ids[0] keep_row = db.session.get(Ingredient, ing_ids[1]) update_recipe_from_payload( recipe_id, recipe_update_payload( recipe, deleted_ingredient_ids=[remove_id], ingredients=[ingredient_payload_from_row(keep_row, order=1)], groups=[], ), ) db.session.commit() self.harness.drain_sync() for term in ("a", "b"): self.assertEqual( len(self.harness.active_ingredient_ids(term, recipe_id)), 1, msg=f"terminal {term}", ) if __name__ == "__main__": unittest.main()