63 lines
2.0 KiB
Python
63 lines
2.0 KiB
Python
"""Dual sync: attach_push — локальная очередь и sync_cycle с push."""
|
||
|
||
from __future__ import annotations
|
||
|
||
import unittest
|
||
|
||
from sqlalchemy import select
|
||
|
||
from app import db
|
||
from app.models import Recipe, SyncQueue
|
||
from app.services.recipe_update_service import update_recipe_from_payload
|
||
from tests.helpers.mill_recipe_fixtures import (
|
||
create_recipe_with_children,
|
||
recipe_update_payload,
|
||
)
|
||
from tests.helpers.sync_dual_harness import SyncDualInstanceHarness
|
||
|
||
SCENARIO_DUAL_PUSH_ROUNDTRIP = "DUAL_PUSH_ROUNDTRIP"
|
||
|
||
|
||
class SyncDualPushRoundtripTests(unittest.TestCase):
|
||
def setUp(self) -> None:
|
||
self.harness = SyncDualInstanceHarness()
|
||
self.harness.start()
|
||
|
||
def tearDown(self) -> None:
|
||
self.harness.stop()
|
||
|
||
def test_terminal_edit_enqueues_local_push_and_sync_cycle_succeeds(self) -> None:
|
||
with self.harness.server_ctx():
|
||
recipe_id, _, _ = create_recipe_with_children(name="Before push")
|
||
self.harness.drain_sync()
|
||
|
||
with self.harness.terminal_ctx("a"):
|
||
recipe = db.session.get(Recipe, recipe_id)
|
||
update_recipe_from_payload(
|
||
recipe_id,
|
||
recipe_update_payload(recipe, ingredients=[], groups=[])
|
||
| {"name": "Pushed from A"},
|
||
)
|
||
db.session.commit()
|
||
pending = db.session.execute(
|
||
select(SyncQueue).where(
|
||
SyncQueue.table_name == "recipe",
|
||
SyncQueue.record_id == recipe_id,
|
||
)
|
||
).scalars().first()
|
||
self.assertIsNotNone(
|
||
pending,
|
||
"terminal A должен поставить recipe в локальную очередь push",
|
||
)
|
||
|
||
result = self.harness.push_from_terminal("a")
|
||
self.assertTrue(result.get("success", True))
|
||
|
||
with self.harness.terminal_ctx("a"):
|
||
local = db.session.get(Recipe, recipe_id)
|
||
self.assertEqual(local.name, "Pushed from A")
|
||
|
||
|
||
if __name__ == "__main__":
|
||
unittest.main()
|