"""Harness: центральный сервер + два терминала (отдельные SQLite) с SyncClient pull/apply/confirm.""" from __future__ import annotations import os import socket import tempfile import threading import time from contextlib import contextmanager from datetime import datetime from typing import Any, Dict, Generator, List, Optional, Tuple from sqlalchemy import select from werkzeug.serving import make_server from app import create_app, db from app.models import ( FeedDispenser, FeedingPeriod, Ingredient, PeriodRecipe, Recipe, SyncClient as SyncClientModel, SyncEngineState, SyncQueue, ) from app.services.sync_manager import SNAPSHOT_MODELS, enqueue_sync_queue_task from config import TestingConfig from sqlalchemy import func from sync_client import SyncClient, _attach_local_db_apply, _attach_local_db_push def _free_port() -> int: sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) sock.bind(("127.0.0.1", 0)) port = sock.getsockname()[1] sock.close() return port def _config_for(tmp_dir: str, db_name: str, *, login: str, password: str) -> type: class _Cfg(TestingConfig): SQLALCHEMY_DATABASE_URI = f"sqlite:///{os.path.join(tmp_dir, db_name)}" SQLALCHEMY_BINDS = {"reports": f"sqlite:///{os.path.join(tmp_dir, db_name.replace('.db', '_reports.db'))}"} AUTH_LOGIN = login AUTH_PASSWORD = password TESTING = True return _Cfg class SyncDualInstanceHarness: """Сервер (master DB) + terminal A/B (client DB) в одном тестовом цикле.""" NODE_A = "sync-dual-term-a" NODE_B = "sync-dual-term-b" AUTH_LOGIN = "sync-dual-admin" AUTH_PASSWORD = "sync-dual-secret" def __init__(self) -> None: self._tmpdir = tempfile.mkdtemp(prefix="wesp-sync-dual-") self._server: Any = None self._server_thread: Optional[threading.Thread] = None self.server_url = "" self.server_app = None self.term_a_app = None self.term_b_app = None self.client_a: Optional[SyncClient] = None self.client_b: Optional[SyncClient] = None self._ctx_stack: List[Any] = [] def start(self) -> None: server_cfg = _config_for( self._tmpdir, "server.db", login=self.AUTH_LOGIN, password=self.AUTH_PASSWORD ) term_a_cfg = _config_for( self._tmpdir, "terminal_a.db", login=self.AUTH_LOGIN, password=self.AUTH_PASSWORD ) term_b_cfg = _config_for( self._tmpdir, "terminal_b.db", login=self.AUTH_LOGIN, password=self.AUTH_PASSWORD ) self.server_app = create_app(server_cfg) self.term_a_app = create_app(term_a_cfg) self.term_b_app = create_app(term_b_cfg) for app in (self.server_app, self.term_a_app, self.term_b_app): ctx = app.app_context() ctx.push() self._ctx_stack.append(ctx) db.create_all() port = _free_port() self.server_url = f"http://127.0.0.1:{port}" self._server = make_server("127.0.0.1", port, self.server_app, threaded=True) self._server_thread = threading.Thread(target=self._server.serve_forever, daemon=True) self._server_thread.start() time.sleep(0.15) self._register_nodes([self.NODE_A, self.NODE_B]) self._mark_bootstrap_ready([self.NODE_A, self.NODE_B]) self.client_a = self._make_sync_client( self.term_a_app, self.NODE_A, "Terminal A", attach_push=True ) self.client_b = self._make_sync_client( self.term_b_app, self.NODE_B, "Terminal B", attach_push=True ) def stop(self) -> None: if self._server is not None: self._server.shutdown() self._server = None for app in (self.server_app, self.term_a_app, self.term_b_app): if app is not None: with app.app_context(): db.session.remove() db.drop_all() while self._ctx_stack: self._ctx_stack.pop().pop() def _register_nodes(self, node_ids: List[str]) -> None: import requests for nid in node_ids: resp = requests.post( f"{self.server_url}/api/sync/register", json={"client_id": nid, "client_name": nid}, timeout=10, ) resp.raise_for_status() def _mark_bootstrap_ready(self, node_ids: List[str]) -> None: with self.server_app.app_context(): state = db.session.get(SyncEngineState, 1) if state is None: state = SyncEngineState(id=1) db.session.add(state) now = datetime.now() state.universal_bootstrap_completed_at = now state.universal_bootstrap_cursor = len(SNAPSHOT_MODELS) for nid in node_ids: sc = db.session.execute( select(SyncClientModel).where(SyncClientModel.node_id == nid) ).scalar_one() sc.personal_snapshot_cursor = len(SNAPSHOT_MODELS) sc.personal_snapshot_completed_at = now db.session.commit() def _make_sync_client( self, app: Any, node_id: str, name: str, *, attach_push: bool = False, ) -> SyncClient: sc = SyncClient() sc.server_url = self.server_url sc.client_id = node_id sc.client_name = name sc.role = "client" sc.config["role"] = "client" sc._initial_sync_active = False _attach_local_db_apply(sc, app) if attach_push: _attach_local_db_push(sc, app) return sc def terminal_test_client(self, which: str): app = self.term_a_app if which == "a" else self.term_b_app client = app.test_client() client.post( "/api/auth/login", json={"login": self.AUTH_LOGIN, "password": self.AUTH_PASSWORD}, ) return client def push_from_terminal(self, which: str) -> Dict[str, Any]: return self.sync_terminal(which) def assert_no_duplicate_sync_queue( self, table_name: str, record_id: str, action: str ) -> None: with self.server_ctx(): n = int( db.session.scalar( select(func.count()) .select_from(SyncQueue) .where( SyncQueue.table_name == table_name, SyncQueue.record_id == record_id, SyncQueue.action == action, SyncQueue.status.in_(("pending", "processing")), ) ) or 0 ) if n > 1: raise AssertionError( f"duplicate sync_queue {table_name}/{record_id}/{action}: {n}" ) def server_test_client(self): client = self.server_app.test_client() client.post( "/api/auth/login", json={"login": self.AUTH_LOGIN, "password": self.AUTH_PASSWORD}, ) return client def sync_terminal(self, which: str) -> Dict[str, Any]: sc = self.client_a if which == "a" else self.client_b assert sc is not None return sc.sync_cycle() def sync_both(self, rounds: int = 3, *, pause_sec: float = 0.05) -> None: for _ in range(rounds): self.sync_terminal("a") self.sync_terminal("b") if pause_sec: time.sleep(pause_sec) def drain_sync(self, *, max_rounds: int = 12, pause_sec: float = 0.08) -> None: for _ in range(max_rounds): ra = self.sync_terminal("a") rb = self.sync_terminal("b") if ( int(ra.get("pulled") or 0) == 0 and int(rb.get("pulled") or 0) == 0 and not self._pending_universal_tasks() ): break time.sleep(pause_sec) def _pending_universal_tasks(self) -> bool: with self.server_app.app_context(): row = db.session.scalar( select(SyncQueue.id) .where( SyncQueue.status.in_(("pending", "processing")), SyncQueue.target_node_id.is_(None), ) .limit(1) ) return row is not None @contextmanager def server_ctx(self) -> Generator[None, None, None]: with self.server_app.app_context(): yield @contextmanager def terminal_ctx(self, which: str) -> Generator[None, None, None]: app = self.term_a_app if which == "a" else self.term_b_app with app.app_context(): yield def recipe_on_terminal(self, which: str, recipe_id: str) -> Optional[Recipe]: with self.terminal_ctx(which): return db.session.get(Recipe, recipe_id) def active_ingredient_ids(self, which: str, recipe_id: str) -> List[str]: with self.terminal_ctx(which): rows = db.session.execute( select(Ingredient.id).where( Ingredient.recipe_id == recipe_id, Ingredient.is_deleted.is_(False), ) ).scalars().all() return [str(x) for x in rows] def sync_queue_tasks(self, table_name: str, record_id: str) -> List[SyncQueue]: with self.server_ctx(): return list( db.session.execute( select(SyncQueue) .where( SyncQueue.table_name == table_name, SyncQueue.record_id == record_id, ) .order_by(SyncQueue.created_at.desc()) ) .scalars() .all() ) def period_recipe_link_deleted( self, which: str, period_id: str, recipe_id: str ) -> bool: with self.terminal_ctx(which): row = db.session.get( PeriodRecipe, {"period_id": period_id, "recipe_id": recipe_id} ) return row is None or bool(row.is_deleted) def seed_dispenser_two_periods_two_recipes( self, ) -> Tuple[str, str, str, str, str]: """1 dispenser, period A/B, recipe r1/r2 в A. Sync на оба терминала.""" disp_id = "dual-disp-1" period_a = "dual-period-a" period_b = "dual-period-b" r1, r2 = "dual-r1", "dual-r2" with self.server_ctx(): db.session.add_all( [ FeedDispenser( id=disp_id, name="Dual D", farm="F", operator="O", type="dispenser", content_hash="", ), FeedingPeriod(id=period_a, name="A", dispenser_id=disp_id), FeedingPeriod(id=period_b, name="B", 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 ord_, rid in enumerate((r1, r2)): db.session.add( PeriodRecipe( period_id=period_a, 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) for rid in (r1, r2): enqueue_sync_queue_task( "period_recipes", f"{period_a}:{rid}", "create", priority=2, target_node_id=None, ) db.session.commit() self.drain_sync() return disp_id, period_a, period_b, r1, r2 def period_recipe_order(self, which: str, period_id: str) -> List[str]: with self.terminal_ctx(which): rows = db.session.execute( select(PeriodRecipe.recipe_id) .where( PeriodRecipe.period_id == period_id, PeriodRecipe.is_deleted.is_(False), ) .order_by(PeriodRecipe.order.asc()) ).scalars().all() return [str(x) for x in rows] def task_status(self, task_id: str) -> Optional[str]: with self.server_ctx(): row = db.session.get(SyncQueue, task_id) return row.status if row else None def latest_task( self, table_name: str, record_id: str, action: str ) -> Optional[SyncQueue]: with self.server_ctx(): return db.session.execute( select(SyncQueue) .where( SyncQueue.table_name == table_name, SyncQueue.record_id == record_id, SyncQueue.action == action, ) .order_by(SyncQueue.created_at.desc()) .limit(1) ).scalar_one_or_none() def confirm_only(self, which: str, task_ids: List[str]) -> bool: sc = self.client_a if which == "a" else self.client_b assert sc is not None return sc.confirm_tasks(task_ids) def pull_only(self, which: str, limit: int = 50) -> List[Dict[str, Any]]: sc = self.client_a if which == "a" else self.client_b assert sc is not None return sc.pull_changes(limit)