import os import tempfile import unittest import uuid from datetime import datetime, timedelta from unittest.mock import patch from sqlalchemy import func, select from app import create_app, db from app.models import ( Component, FeedDispenser, FeedingPeriod, Ingredient, LoadingReport, LoadingReportComponent, PeriodRecipe, Recipe, SyncClient, SyncEngineState, SyncDelivery, SyncQueue, WESP_SUPPRESS_SYNC_ENQUEUE, ) from app.services.sync_manager import ( SNAPSHOT_MODELS, SyncManager, apply_sync_change, enqueue_personal_snapshot, enqueue_sync_queue_task, requeue_stuck_processing, ) from app.timeutil import utc_now_naive from config import TestingConfig class SyncTestConfig(TestingConfig): _TMP_DIR = tempfile.mkdtemp(prefix="wesp-sync-tests-") 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')}", } class SyncManagerTests(unittest.TestCase): def setUp(self) -> None: self.app = create_app(SyncTestConfig) self.ctx = self.app.app_context() self.ctx.push() db.create_all() def tearDown(self) -> None: db.session.remove() db.drop_all() self.ctx.pop() def test_apply_sync_change_strips_unknown_keys_from_get_object_data(self) -> None: """get_object_data добавляет legacy-ключи soft-delete без колонок в модели — create не должен падать.""" cid = str(uuid.uuid4()) data = { "id": cid, "name": "K1", "type": "grain", "is_active": True, "dry_matter": 1.0, "protein": 2.0, "energy": 3.0, "price": 4.0, "version": 1, "content_hash": "", "created_by": "system", "updated_by": "system", "is_deleted": False, "deleted_at": None, "deleted_by": None, "deleted_reason": None, "restored_at": None, "restored_by": None, "restored_reason": None, "delete_restore_count": 0, } res = apply_sync_change("component", cid, "create", data) self.assertTrue(res.get("success"), msg=res.get("error")) db.session.commit() row = db.session.get(Component, cid) self.assertIsNotNone(row) self.assertEqual(row.name, "K1") def test_apply_period_recipes_coerces_iso_datetime_strings(self) -> None: """После JSON pull даты приходят строками; ORM + SQLite DateTime ожидают datetime.""" d_id = str(uuid.uuid4()) fp_id = str(uuid.uuid4()) r_id = str(uuid.uuid4()) db.session.add_all( [ FeedDispenser( id=d_id, name="D", farm="F", operator="O", content_hash="", ), FeedingPeriod( id=fp_id, name="Per", dispenser_id=d_id, content_hash="", ), Recipe(id=r_id, name="Rec", content_hash=""), ] ) db.session.commit() rid_key = f"{fp_id}:{r_id}" data = { "period_id": fp_id, "recipe_id": r_id, "order": 0, "version": 1, "created_by": "system", "updated_by": "system", "sync_status": "pending", "content_hash": "ab" * 32, "created_at": "2026-04-20T01:48:57.123456", "updated_at": "2026-04-21T02:00:00Z", "is_deleted": False, "deleted_at": None, "deleted_by": None, } res = apply_sync_change("period_recipes", rid_key, "update", data) self.assertTrue(res.get("success"), msg=res.get("error")) db.session.commit() row = db.session.execute( select(PeriodRecipe).where( PeriodRecipe.period_id == fp_id, PeriodRecipe.recipe_id == r_id, ) ).scalar_one() self.assertIsInstance(row.created_at, datetime) self.assertIsInstance(row.updated_at, datetime) def test_apply_sync_change_failed_flush_does_not_poison_session(self) -> None: """После ошибки flush одной записи следующая в той же сессии всё ещё коммитится (SAVEPOINT).""" orig_flush = db.session.flush boom = [0] def side_effect(*args, **kwargs): boom[0] += 1 if boom[0] == 1: raise RuntimeError("simulated flush failure") return orig_flush(*args, **kwargs) def cmp_data(name: str) -> dict: return { "name": name, "type": "grain", "is_active": True, "dry_matter": 1.0, "protein": 0.0, "energy": 0.0, "price": 0.0, "version": 1, "content_hash": "", "created_by": "system", "updated_by": "system", } c1 = str(uuid.uuid4()) self.assertTrue(apply_sync_change("component", c1, "create", cmp_data("ok-first"))["success"]) db.session.commit() c_bad = str(uuid.uuid4()) c_ok = str(uuid.uuid4()) with patch.object(db.session, "flush", side_effect=side_effect): r_bad = apply_sync_change("component", c_bad, "create", cmp_data("bad")) self.assertFalse(r_bad.get("success")) r_ok = apply_sync_change("component", c_ok, "create", cmp_data("ok-after-fail")) self.assertTrue(r_ok.get("success"), msg=r_ok.get("error")) db.session.commit() self.assertIsNotNone(db.session.get(Component, c1)) self.assertIsNone(db.session.get(Component, c_bad)) self.assertIsNotNone(db.session.get(Component, c_ok)) def test_remote_apply_suppress_flag_skips_sync_queue_enqueue(self) -> None: """Как на клиенте при pull: не плодить sync_queue на каждый применённый объект.""" cid = str(uuid.uuid4()) db.session.info[WESP_SUPPRESS_SYNC_ENQUEUE] = True try: r = apply_sync_change( "component", cid, "create", { "name": "NoQueue", "type": "grain", "is_active": True, "dry_matter": 1.0, "protein": 0.0, "energy": 0.0, "price": 0.0, "version": 1, "content_hash": "", "created_by": "system", "updated_by": "system", }, ) self.assertTrue(r.get("success"), r) db.session.commit() finally: db.session.info.pop(WESP_SUPPRESS_SYNC_ENQUEUE, None) n = db.session.scalar( select(func.count()).select_from(SyncQueue).where( SyncQueue.table_name == "component", SyncQueue.record_id == cid, ) ) self.assertEqual(int(n or 0), 0) def test_process_push_server_master_skips_stale_version_applies_rest(self) -> None: """Серверный master: несовпадение версии — пропуск строки без 409; остальные изменения применяются.""" existing = Component(id="cmp-1", name="server-name", version=2) db.session.add(existing) db.session.commit() changes = [ { "table_name": "component", "record_id": "cmp-1", "action": "update", "data": {"id": "cmp-1", "name": "client-name", "version": 1}, }, { "table_name": "component", "record_id": "cmp-2", "action": "create", "data": {"id": "cmp-2", "name": "new-component"}, }, ] result = SyncManager.process_push(client_id="client-a", changes=changes) self.assertEqual(result["status_code"], 200) self.assertEqual(result["payload"]["total_applied"], 1) cmp1 = db.session.get(Component, "cmp-1") cmp2 = db.session.get(Component, "cmp-2") self.assertIsNotNone(cmp1) self.assertEqual(cmp1.name, "server-name") self.assertIsNotNone(cmp2) self.assertEqual(cmp2.name, "new-component") def test_process_push_unknown_table_returns_409(self) -> None: changes = [ { "table_name": "unknown_table_xyz", "record_id": "x-1", "action": "update", "data": {"id": "x-1"}, }, ] result = SyncManager.process_push(client_id="client-a", changes=changes) self.assertEqual(result["status_code"], 409) self.assertTrue(result["payload"]["error"]) def test_check_for_sync_conflict_detects_version_mismatch(self) -> None: from app.services.sync_manager import _check_for_sync_conflict cid = str(uuid.uuid4()) db.session.add( Component( id=cid, name="Server", type="grain", dry_matter=50.0, protein=0.0, energy=0.0, price=0.0, version=3, ) ) db.session.commit() conflict = _check_for_sync_conflict( "component", cid, {"id": cid, "name": "Client", "version": 1}, ) self.assertIsNotNone(conflict) self.assertEqual(conflict.get("conflict_type"), "version_mismatch") def test_process_confirm_universal_task_completes_only_after_all_active_clients(self) -> None: now = datetime.now() client1 = SyncClient(node_id="node-1", client_name="client-1", status="active", is_enabled=True) client2 = SyncClient(node_id="node-2", client_name="client-2", status="active", is_enabled=True) db.session.add_all([client1, client2]) db.session.flush() task = SyncQueue( id="task-1", table_name="component", record_id="cmp-x", action="update", status="processing", target_node_id=None, created_at=now, processed_at=now, ) db.session.add(task) # Delivered only to client1 for now db.session.add(SyncDelivery(client_id=client1.id, task_id=task.id, delivered_at=now)) db.session.commit() first = SyncManager.process_confirm(client_id="node-1", task_ids=[task.id]) self.assertEqual(first["status_code"], 200) self.assertEqual(first["payload"]["updated"], 0) self.assertEqual(db.session.get(SyncQueue, task.id).status, "processing") # After delivery to all active clients, universal task can be completed db.session.add(SyncDelivery(client_id=client2.id, task_id=task.id, delivered_at=now)) db.session.commit() second = SyncManager.process_confirm(client_id="node-1", task_ids=[task.id]) self.assertEqual(second["status_code"], 200) self.assertEqual(second["payload"]["updated"], 1) self.assertEqual(db.session.get(SyncQueue, task.id).status, "completed") def test_requeue_does_not_fire_on_fresh_utc_processed_at(self) -> None: """processed_at в UTC; requeue не должен срабатывать на свежих задачах (раньше ломалось с datetime.now()).""" now = utc_now_naive() task = SyncQueue( id="task-fresh", table_name="component", record_id="cmp-fresh", action="update", status="processing", target_node_id=None, created_at=now, processed_at=now, ) db.session.add(task) db.session.commit() n = requeue_stuck_processing(timeout_minutes=5) self.assertEqual(n, 0) self.assertEqual(db.session.get(SyncQueue, task.id).status, "processing") def test_process_pull_delivers_processing_universal_task_to_second_client(self) -> None: """Второй клиент получает универсальную задачу в processing, если delivery ещё не было.""" now = utc_now_naive() db.session.add( SyncEngineState(id=1, universal_bootstrap_completed_at=now, universal_bootstrap_cursor=0) ) client1 = SyncClient( node_id="node-a", client_name="a", status="active", is_enabled=True, personal_snapshot_cursor=len(SNAPSHOT_MODELS), personal_snapshot_completed_at=now, ) client2 = SyncClient( node_id="node-b", client_name="b", status="active", is_enabled=True, personal_snapshot_cursor=len(SNAPSHOT_MODELS), personal_snapshot_completed_at=now, ) db.session.add_all([client1, client2]) db.session.flush() cid = str(uuid.uuid4()) from app.models import WESP_SUPPRESS_SYNC_ENQUEUE db.session.info[WESP_SUPPRESS_SYNC_ENQUEUE] = True try: db.session.add( Component( id=cid, name="Shared", type="grain", dry_matter=1.0, protein=1.0, energy=1.0, price=0.0, content_hash="0" * 64, ) ) db.session.flush() finally: db.session.info.pop(WESP_SUPPRESS_SYNC_ENQUEUE, None) task = SyncQueue( id="task-shared", table_name="component", record_id=cid, action="create", status="processing", target_node_id=None, priority=3, created_at=now, processed_at=now, ) db.session.add(task) db.session.add(SyncDelivery(client_id=client1.id, task_id=task.id, delivered_at=now)) db.session.commit() res = SyncManager.process_pull(client_id="node-b", limit=10) self.assertEqual(res["status_code"], 200) self.assertEqual(len(res["payload"]["changes"]), 1) deliveries = db.session.execute( select(func.count()).select_from(SyncDelivery).where(SyncDelivery.task_id == task.id) ).scalar() self.assertEqual(int(deliveries or 0), 2) def test_loading_report_component_insert_enqueues_sync(self) -> None: rid = str(uuid.uuid4()) report = LoadingReport( id=rid, recipe_id=str(uuid.uuid4()), recipe_name="R", start_time=datetime.now(), dispenser_type="dispenser", ) db.session.add(report) db.session.flush() lrc_id = str(uuid.uuid4()) lrc = LoadingReportComponent( id=lrc_id, report_id=rid, component_name="C", target_weight=1.0, actual_weight=1.0, loading_order=1, ) db.session.add(lrc) db.session.commit() row = db.session.execute( select(SyncQueue).where( SyncQueue.table_name == "loading_report_component", SyncQueue.record_id == lrc_id, SyncQueue.action == "create", ) ).scalar_one_or_none() self.assertIsNotNone(row) self.assertEqual(row.status, "pending") def test_enqueue_personal_snapshot_includes_reports(self) -> None: rid = str(uuid.uuid4()) report = LoadingReport( id=rid, recipe_id=str(uuid.uuid4()), recipe_name="R2", start_time=datetime.now(), dispenser_type="dispenser", ) db.session.add(report) db.session.commit() n = enqueue_personal_snapshot("new-node-1") self.assertGreaterEqual(n, 1) hit = db.session.execute( select(SyncQueue).where( SyncQueue.table_name == "loading_report", SyncQueue.record_id == rid, SyncQueue.target_node_id == "new-node-1", ) ).scalar_one_or_none() self.assertIsNotNone(hit) def test_process_pull_does_not_run_universal_step_when_flag_is_set(self) -> None: now = datetime.now() db.session.add( SyncEngineState(id=1, universal_bootstrap_completed_at=now, universal_bootstrap_cursor=0) ) db.session.add( SyncClient( node_id="node-ready", client_name="ready", status="active", is_enabled=True, personal_snapshot_completed_at=now, personal_snapshot_cursor=len(SNAPSHOT_MODELS), ) ) db.session.commit() with patch("app.services.sync_manager._bootstrap_universal_step") as boot_step: first = SyncManager.process_pull(client_id="node-ready", limit=10) second = SyncManager.process_pull(client_id="node-ready", limit=10) self.assertEqual(first["status_code"], 200) self.assertEqual(second["status_code"], 200) boot_step.assert_not_called() def test_process_pull_reactivates_soft_deleted_sync_client(self) -> None: """Повторный pull после soft-delete в админке не должен падать по UNIQUE(node_id).""" now = datetime.now() db.session.add( SyncEngineState(id=1, universal_bootstrap_completed_at=now, universal_bootstrap_cursor=0) ) db.session.add( SyncClient( node_id="node-sd", client_name="sd", status="disabled", is_enabled=False, is_deleted=True, personal_snapshot_completed_at=now, personal_snapshot_cursor=len(SNAPSHOT_MODELS), ) ) db.session.commit() res = SyncManager.process_pull( client_id="node-sd", limit=10, client_name="back", client_ip="192.168.0.13" ) self.assertEqual(res["status_code"], 200) row = db.session.execute( select(SyncClient).where(SyncClient.node_id == "node-sd") ).scalar_one() self.assertFalse(bool(row.is_deleted)) self.assertTrue(row.is_enabled) self.assertEqual(row.client_name, "back") self.assertEqual(row.ip_address, "192.168.0.13") def test_process_pull_personal_snapshot_runs_once_with_client_flag(self) -> None: now = datetime.now() db.session.add( SyncEngineState(id=1, universal_bootstrap_completed_at=now, universal_bootstrap_cursor=0) ) db.session.add( SyncClient( node_id="node-new", client_name="new", status="active", is_enabled=True, ) ) db.session.commit() step_payload = {"done": True, "created": 0, "cursor": 0, "total": 1} def _snapshot_once(client, *, max_models): client.personal_snapshot_cursor = len(SNAPSHOT_MODELS) return step_payload with patch( "app.services.sync_manager._enqueue_personal_snapshot_step", side_effect=_snapshot_once, ) as snapshot_step: first = SyncManager.process_pull(client_id="node-new", limit=10) second = SyncManager.process_pull(client_id="node-new", limit=10) self.assertEqual(first["status_code"], 200) self.assertEqual(second["status_code"], 200) self.assertEqual(snapshot_step.call_count, 1) def test_enqueue_redelivery_clears_sync_delivery_for_completed_universal(self) -> None: """Повторный enqueue update после completed должен снова выдать задачу клиентам.""" old = utc_now_naive() - timedelta(minutes=5) client = SyncClient(node_id="node-enq", client_name="c", status="active", is_enabled=True) db.session.add(client) db.session.flush() task = SyncQueue( id="task-enq", table_name="period_recipes", record_id="period-a:recipe-a", action="update", status="completed", target_node_id=None, created_at=old, processed_at=old, completed_at=old, ) db.session.add(task) db.session.add(SyncDelivery(client_id=client.id, task_id=task.id, delivered_at=old)) db.session.commit() enqueue_sync_queue_task("period_recipes", "period-a:recipe-a", "update", priority=4) db.session.commit() self.assertEqual(db.session.get(SyncQueue, task.id).status, "pending") d = db.session.execute( select(SyncDelivery).where(SyncDelivery.task_id == task.id) ).scalar_one_or_none() self.assertIsNone(d) def test_requeue_stuck_clears_sync_delivery_for_universal_retry(self) -> None: """После requeue универсальная задача снова может быть выдана (иначе блокирует NOT EXISTS delivery).""" old = utc_now_naive() - timedelta(minutes=60) client = SyncClient(node_id="node-rq", client_name="c", status="active", is_enabled=True) db.session.add(client) db.session.flush() task = SyncQueue( id="task-rq", table_name="component", record_id="cmp-rq", action="create", status="processing", target_node_id=None, created_at=old, processed_at=old, ) db.session.add(task) db.session.add(SyncDelivery(client_id=client.id, task_id=task.id, delivered_at=old)) db.session.commit() n = requeue_stuck_processing(timeout_minutes=30) self.assertGreaterEqual(n, 1) self.assertEqual(db.session.get(SyncQueue, task.id).status, "pending") d = db.session.execute( select(SyncDelivery).where(SyncDelivery.task_id == task.id) ).scalar_one_or_none() self.assertIsNone(d) def test_process_pull_delivers_when_universal_bootstrap_incomplete_but_queue_has_tasks(self) -> None: """Живые задачи в очереди не должны ждать 202 от universal bootstrap.""" db.session.add(SyncEngineState(id=1)) cid = "cmp-ub-1" db.session.add( Component( id=cid, name="Ub", type="grain", dry_matter=1.0, protein=1.0, energy=1.0, price=0.0, content_hash="0" * 64, ) ) db.session.add( SyncClient( node_id="node-ub", client_name="ub", status="active", is_enabled=True, ) ) now = datetime.now() db.session.add( SyncQueue( id="task-ub", table_name="component", record_id=cid, action="create", status="pending", target_node_id=None, priority=3, created_at=now, ) ) db.session.commit() res = SyncManager.process_pull(client_id="node-ub", limit=10) self.assertEqual(res["status_code"], 200) self.assertGreater(len(res["payload"]["changes"]), 0) def test_initial_sync_active_while_personal_tasks_pending(self) -> None: """Enqueue завершён, но personal задачи ещё pending — initial_sync_active остаётся True.""" now = datetime.now() total = len(SNAPSHOT_MODELS) db.session.add( SyncEngineState(id=1, universal_bootstrap_completed_at=now, universal_bootstrap_cursor=0) ) db.session.add( SyncClient( node_id="node-pending", client_name="pending", status="active", is_enabled=True, personal_snapshot_cursor=total, ) ) db.session.add( SyncQueue( id="task-pend", table_name="component", record_id=str(uuid.uuid4()), action="create", status="pending", target_node_id="node-pending", priority=2, created_at=now, ) ) db.session.commit() res = SyncManager.process_pull(client_id="node-pending", limit=10) self.assertEqual(res["status_code"], 200) self.assertTrue(res["payload"]["initial_sync_active"]) row = db.session.execute( select(SyncClient).where(SyncClient.node_id == "node-pending") ).scalar_one() self.assertIsNone(row.personal_snapshot_completed_at) def test_confirm_all_personal_marks_snapshot_completed(self) -> None: now = datetime.now() total = len(SNAPSHOT_MODELS) client = SyncClient( node_id="node-done", client_name="done", status="active", is_enabled=True, personal_snapshot_cursor=total, ) db.session.add(client) db.session.flush() task = SyncQueue( id="task-done", table_name="component", record_id=str(uuid.uuid4()), action="create", status="processing", target_node_id="node-done", priority=2, created_at=now, processed_at=now, ) db.session.add(task) db.session.commit() res = SyncManager.process_confirm(client_id="node-done", task_ids=[task.id]) self.assertEqual(res["status_code"], 200) self.assertFalse(res["payload"]["initial_sync_active"]) row = db.session.execute( select(SyncClient).where(SyncClient.node_id == "node-done") ).scalar_one() self.assertIsNotNone(row.personal_snapshot_completed_at) def test_pull_initial_sync_orders_parent_before_child(self) -> None: now = datetime.now() total = len(SNAPSHOT_MODELS) db.session.add( SyncEngineState(id=1, universal_bootstrap_completed_at=now, universal_bootstrap_cursor=0) ) db.session.add( SyncClient( node_id="node-order", client_name="order", status="active", is_enabled=True, personal_snapshot_cursor=total, ) ) disp_id = str(uuid.uuid4()) period_id = str(uuid.uuid4()) db.session.add( SyncQueue( id="t-period", table_name="feeding_period", record_id=period_id, action="create", status="pending", target_node_id="node-order", priority=2, created_at=now + timedelta(seconds=1), ) ) db.session.add( SyncQueue( id="t-disp", table_name="feed_dispenser", record_id=disp_id, action="create", status="pending", target_node_id="node-order", priority=2, created_at=now + timedelta(seconds=2), ) ) db.session.add(FeedDispenser(id=disp_id, name="D", farm="F", operator="O", content_hash="")) db.session.add( FeedingPeriod(id=period_id, name="P", dispenser_id=disp_id, content_hash="") ) db.session.commit() res = SyncManager.process_pull(client_id="node-order", limit=10) self.assertEqual(res["status_code"], 200) tables = [c["table_name"] for c in res["payload"]["changes"]] if len(tables) >= 2: self.assertLess( tables.index("feed_dispenser"), tables.index("feeding_period"), ) def test_cascade_feed_dispenser_marks_period_deleted_and_enqueues(self) -> None: from app.services.sync_cascade import cascade_soft_delete disp_id = str(uuid.uuid4()) period_id = str(uuid.uuid4()) db.session.add(FeedDispenser(id=disp_id, name="D", farm="F", operator="O", content_hash="")) db.session.add( FeedingPeriod(id=period_id, name="P", dispenser_id=disp_id, content_hash="") ) db.session.commit() cascade_soft_delete("feed_dispenser", disp_id, deleted_by="test") db.session.commit() period = db.session.get(FeedingPeriod, period_id) self.assertIsNotNone(period) self.assertTrue(period.is_deleted) q = db.session.execute( select(SyncQueue) .where( SyncQueue.table_name == "feeding_period", SyncQueue.record_id == period_id, SyncQueue.action == "update", ) .order_by(SyncQueue.created_at.desc()) ).scalars().first() self.assertIsNotNone(q) def test_one_completed_personal_task_does_not_finish_initial_sync_on_pull(self) -> None: """Reconnect: одна completed personal-задача не должна завершать initial sync.""" now = datetime.now() total = len(SNAPSHOT_MODELS) db.session.add( SyncEngineState(id=1, universal_bootstrap_completed_at=now, universal_bootstrap_cursor=0) ) db.session.add( SyncClient( node_id="node-one-done", client_name="one", status="active", is_enabled=True, personal_snapshot_cursor=total, ) ) db.session.add( SyncQueue( id="task-done-one", table_name="component", record_id=str(uuid.uuid4()), action="create", status="completed", target_node_id="node-one-done", priority=2, created_at=now, completed_at=now, ) ) pending_id = str(uuid.uuid4()) db.session.add( SyncQueue( id="task-still-pending", table_name="component", record_id=pending_id, action="create", status="pending", target_node_id="node-one-done", priority=2, created_at=now, ) ) db.session.commit() res = SyncManager.process_pull(client_id="node-one-done", limit=10) self.assertEqual(res["status_code"], 200) self.assertTrue(res["payload"]["initial_sync_active"]) row = db.session.execute( select(SyncClient).where(SyncClient.node_id == "node-one-done") ).scalar_one() self.assertIsNone(row.personal_snapshot_completed_at) def test_initial_sync_header_uses_larger_snapshot_step(self) -> None: now = datetime.now() db.session.add( SyncEngineState( id=1, universal_bootstrap_completed_at=now, universal_bootstrap_cursor=0, ) ) db.session.add( SyncClient( node_id="node-hdr", client_name="hdr", status="active", is_enabled=True, ) ) db.session.commit() seen: list[int] = [] def _capture_step(client, *, max_models): seen.append(max_models) client.personal_snapshot_cursor = len(SNAPSHOT_MODELS) return {"done": True, "created": 0, "cursor": len(SNAPSHOT_MODELS), "total": len(SNAPSHOT_MODELS)} with patch( "app.services.sync_manager._enqueue_personal_snapshot_step", side_effect=_capture_step, ): SyncManager.process_pull( client_id="node-hdr", limit=10, initial_sync_header=True ) self.assertTrue(seen) self.assertGreaterEqual(seen[0], 5) def test_cascade_recipe_via_soft_delete_enqueues_ingredients(self) -> None: rid = str(uuid.uuid4()) ing_id = str(uuid.uuid4()) db.session.add(Recipe(id=rid, name="R", content_hash="")) db.session.add( Ingredient( id=ing_id, recipe_id=rid, name="Ing", amount=1.0, dry_matter=1.0, content_hash="", ) ) db.session.commit() recipe = db.session.get(Recipe, rid) recipe.soft_delete(deleted_by_user="test") db.session.commit() ing = db.session.get(Ingredient, ing_id) self.assertIsNotNone(ing) self.assertTrue(ing.is_deleted) q = db.session.execute( select(SyncQueue) .where( SyncQueue.table_name == "ingredient", SyncQueue.record_id == ing_id, SyncQueue.action == "update", ) .order_by(SyncQueue.created_at.desc()) ).scalars().first() self.assertIsNotNone(q) def test_repair_legacy_sync_client_snapshot_cursor(self) -> None: from alembic.operations import Operations from alembic.runtime.migration import MigrationContext from app.schema_bootstrap import repair_legacy_sync_client_snapshot_cursor now = datetime.now() total = len(SNAPSHOT_MODELS) db.session.add( SyncClient( node_id="node-legacy-cursor", client_name="legacy", status="active", is_enabled=True, personal_snapshot_completed_at=now, personal_snapshot_cursor=0, ) ) db.session.commit() with db.engine.begin() as conn: ctx = MigrationContext.configure(conn) repair_legacy_sync_client_snapshot_cursor(Operations(ctx)) row = db.session.execute( select(SyncClient).where(SyncClient.node_id == "node-legacy-cursor") ).scalar_one() self.assertEqual(int(row.personal_snapshot_cursor or 0), total) if __name__ == "__main__": unittest.main()