404 lines
14 KiB
Python
404 lines
14 KiB
Python
"""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)
|