262 lines
10 KiB
Python
262 lines
10 KiB
Python
import argparse
|
|
import importlib.util
|
|
import subprocess
|
|
import tempfile
|
|
import time
|
|
import unittest
|
|
from pathlib import Path
|
|
from unittest import mock
|
|
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
SPEC = importlib.util.spec_from_file_location(
|
|
"sync_peer_worker", ROOT / "tools/sync_peer_worker.py"
|
|
)
|
|
worker = importlib.util.module_from_spec(SPEC)
|
|
assert SPEC.loader is not None
|
|
SPEC.loader.exec_module(worker)
|
|
|
|
|
|
class SyncPeerWorkerTest(unittest.TestCase):
|
|
def setUp(self):
|
|
self.temporary = tempfile.TemporaryDirectory(prefix="trainlog-peer-worker-test-")
|
|
self.transport = Path(self.temporary.name) / "transport"
|
|
self.transport.mkdir()
|
|
|
|
def tearDown(self):
|
|
self.temporary.cleanup()
|
|
|
|
def test_rejected_android_generation_retains_consumer_diagnostic(self):
|
|
error = worker.rejected_android_generation_error(
|
|
{"diagnostic": "concurrent sleep diary revision"}
|
|
)
|
|
self.assertEqual(
|
|
str(error),
|
|
"desktop rejected Android generation: concurrent sleep diary revision",
|
|
)
|
|
self.assertEqual(
|
|
str(worker.rejected_android_generation_error({"diagnostic": ""})),
|
|
"desktop rejected Android generation",
|
|
)
|
|
|
|
def test_drive_adapter_uses_remaining_conversation_deadline(self):
|
|
completed = subprocess.CompletedProcess([], 0, b"", b"")
|
|
with mock.patch.object(worker.subprocess, "run", return_value=completed) as run:
|
|
worker.run_drive_adapter(
|
|
Path("/fixed/adapter.py"),
|
|
"pull",
|
|
"remote:Trainlog/Sync-Test/run",
|
|
self.transport,
|
|
time.monotonic() + 180,
|
|
)
|
|
timeout = run.call_args.kwargs["timeout"]
|
|
self.assertGreaterEqual(timeout, 178)
|
|
self.assertLessEqual(timeout, 180)
|
|
|
|
def test_mtp_adapter_uses_fixed_operation_budget(self):
|
|
completed = subprocess.CompletedProcess([], 0, b"", b"")
|
|
with mock.patch.object(worker.subprocess, "run", return_value=completed) as run:
|
|
worker.run_adapter(
|
|
Path("/fixed/adapter"), "pull", "peer_fixed",
|
|
self.transport, 50.0 + worker.MTP_OPERATION_TIMEOUT_SECONDS + 10.0,
|
|
clock=lambda: 50.0,
|
|
)
|
|
self.assertEqual(worker.MTP_OPERATION_TIMEOUT_SECONDS, run.call_args.kwargs["timeout"])
|
|
|
|
def test_mtp_adapter_uses_and_releases_shared_transport_lock(self):
|
|
completed = subprocess.CompletedProcess([], 0, b"", b"")
|
|
real_flock = worker.fcntl.flock
|
|
with mock.patch.object(worker.subprocess, "run", return_value=completed), mock.patch.object(
|
|
worker.fcntl, "flock", wraps=real_flock
|
|
) as flock:
|
|
worker.run_adapter(
|
|
Path("/fixed/adapter"), "pull", "peer_fixed",
|
|
self.transport, 100.0, clock=lambda: 50.0,
|
|
)
|
|
self.assertTrue((self.transport.parent / "mtp.lock").is_file())
|
|
operations = [call.args[1] for call in flock.call_args_list]
|
|
self.assertIn(worker.fcntl.LOCK_EX | worker.fcntl.LOCK_NB, operations)
|
|
self.assertEqual(worker.fcntl.LOCK_UN, operations[-1])
|
|
|
|
def test_mtp_pull_pump_is_throttled(self):
|
|
now = [0.0]
|
|
pulls = []
|
|
pump = worker.MtpPullPump(
|
|
lambda: pulls.append(now[0]), interval=3.0, clock=lambda: now[0]
|
|
)
|
|
for timestamp in (0.05, 0.5, 2.99, 3.0, 3.1, 5.99, 6.0):
|
|
now[0] = timestamp
|
|
pump()
|
|
self.assertEqual([3.0, 6.0], pulls)
|
|
|
|
def test_global_deadline_is_not_reported_as_transport_timeout(self):
|
|
with self.assertRaises(worker.ConversationDeadlineExpired):
|
|
worker.run_adapter(
|
|
Path("/fixed/adapter"), "pull", "peer_fixed",
|
|
self.transport, 10.0, clock=lambda: 10.0,
|
|
)
|
|
|
|
def test_deadline_limited_pull_becomes_protocol_timeout(self):
|
|
with mock.patch.object(
|
|
worker.subprocess, "run", side_effect=subprocess.TimeoutExpired(["adapter"], 0.5)
|
|
):
|
|
with self.assertRaises(worker.ConversationDeadlineExpired):
|
|
worker.run_adapter(
|
|
Path("/fixed/adapter"), "pull", "peer_fixed",
|
|
self.transport, 10.5, clock=lambda: 10.0,
|
|
)
|
|
|
|
def test_blocked_full_budget_pull_is_transport_timeout(self):
|
|
with mock.patch.object(
|
|
worker.subprocess, "run", side_effect=subprocess.TimeoutExpired(["adapter"], 30)
|
|
):
|
|
with self.assertRaisesRegex(RuntimeError, "transport_timeout"):
|
|
worker.run_adapter(
|
|
Path("/fixed/adapter"), "pull", "peer_fixed",
|
|
self.transport, 50.0 + worker.MTP_OPERATION_TIMEOUT_SECONDS + 10.0,
|
|
clock=lambda: 50.0,
|
|
)
|
|
|
|
def test_delayed_correlated_generation_survives_multiple_polls(self):
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
reference = Path(directory) / "android-generation-v1.json"
|
|
now = [0.0]
|
|
pulls = []
|
|
|
|
def pump():
|
|
pulls.append(now[0])
|
|
if len(pulls) == 3:
|
|
reference.write_text('{"run_id":"sy_fresh"}', encoding="utf-8")
|
|
|
|
value = worker.wait_json(
|
|
reference, 10.0,
|
|
lambda candidate: candidate.get("run_id") == "sy_fresh",
|
|
pump, clock=lambda: now[0],
|
|
sleeper=lambda duration: now.__setitem__(0, now[0] + duration),
|
|
)
|
|
self.assertEqual("sy_fresh", value["run_id"])
|
|
self.assertEqual(3, len(pulls))
|
|
|
|
def test_stale_reference_is_ignored_until_correlated_replacement(self):
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
reference = Path(directory) / "android-generation-v1.json"
|
|
reference.write_text('{"run_id":"sy_stale"}', encoding="utf-8")
|
|
now = [0.0]
|
|
polls = [0]
|
|
|
|
def pump():
|
|
polls[0] += 1
|
|
if polls[0] == 2:
|
|
reference.write_text('{"run_id":"sy_fresh"}', encoding="utf-8")
|
|
|
|
value = worker.wait_json(
|
|
reference, 1.0,
|
|
lambda candidate: candidate.get("run_id") == "sy_fresh",
|
|
pump, clock=lambda: now[0],
|
|
sleeper=lambda duration: now.__setitem__(0, now[0] + duration),
|
|
)
|
|
self.assertEqual("sy_fresh", value["run_id"])
|
|
|
|
def test_bluetooth_adapter_uses_distinct_transport_lock(self):
|
|
completed = subprocess.CompletedProcess([], 0, b"", b"")
|
|
with mock.patch.object(worker.subprocess, "run", return_value=completed):
|
|
worker.run_adapter(
|
|
Path("/fixed/bt-adapter"),
|
|
"pull",
|
|
"peer_fixed",
|
|
self.transport,
|
|
100.0,
|
|
clock=lambda: 50.0,
|
|
transport_name="Bluetooth",
|
|
lock_name="bt.lock",
|
|
)
|
|
self.assertTrue((self.transport.parent / "bt.lock").is_file())
|
|
|
|
def test_ai_export_delivery_is_best_effort_and_uses_bluetooth(self):
|
|
database = Path(self.temporary.name) / "trainlog.db"
|
|
database.touch()
|
|
args = argparse.Namespace(
|
|
database=database,
|
|
transport_root=self.transport,
|
|
mode="bt",
|
|
bt_adapter=Path("/fixed/bt-adapter"),
|
|
mtp_adapter=None,
|
|
expected_peer="peer_fixed",
|
|
)
|
|
|
|
def execute(command, **kwargs):
|
|
if command[0] == worker.sys.executable:
|
|
output = Path(command[2])
|
|
output.write_text(
|
|
'{"format":"TRAINLOG_AI_EXPORT","version":1}\n',
|
|
encoding="utf-8",
|
|
)
|
|
return subprocess.CompletedProcess(
|
|
command,
|
|
0,
|
|
"TRAINLOG_AI_EXPORT_V1=PASS\n",
|
|
"",
|
|
)
|
|
self.assertEqual("/fixed/bt-adapter", command[0])
|
|
self.assertEqual("push", command[1])
|
|
return subprocess.CompletedProcess(command, 0, b"", b"")
|
|
|
|
with mock.patch.object(worker.subprocess, "run", side_effect=execute):
|
|
result = worker.publish_ai_export_to_android(args, time.monotonic() + 30)
|
|
self.assertEqual({"result": "delivered_to_android"}, result)
|
|
self.assertTrue((self.transport / "trainlog_ai_export_v1.json").is_file())
|
|
|
|
def test_ai_export_transport_failure_does_not_raise(self):
|
|
database = Path(self.temporary.name) / "trainlog.db"
|
|
database.touch()
|
|
args = argparse.Namespace(
|
|
database=database,
|
|
transport_root=self.transport,
|
|
mode="bt",
|
|
bt_adapter=Path("/fixed/bt-adapter"),
|
|
mtp_adapter=None,
|
|
expected_peer="peer_fixed",
|
|
)
|
|
|
|
def execute(command, **kwargs):
|
|
if command[0] == worker.sys.executable:
|
|
Path(command[2]).write_text(
|
|
'{"format":"TRAINLOG_AI_EXPORT","version":1}\n',
|
|
encoding="utf-8",
|
|
)
|
|
return subprocess.CompletedProcess(
|
|
command,
|
|
0,
|
|
"TRAINLOG_AI_EXPORT_V1=PASS\n",
|
|
"",
|
|
)
|
|
return subprocess.CompletedProcess(
|
|
command,
|
|
2,
|
|
b"",
|
|
b"Bluetooth adapter failed status=6 diagnostic=peer vanished",
|
|
)
|
|
|
|
with mock.patch.object(worker.subprocess, "run", side_effect=execute):
|
|
result = worker.publish_ai_export_to_android(args, time.monotonic() + 30)
|
|
self.assertEqual("pending", result["result"])
|
|
self.assertIn("peer vanished", result["diagnostic"])
|
|
|
|
def test_missing_generation_reports_correlated_protocol_timeout(self):
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
reference = Path(directory) / "android-generation-v1.json"
|
|
reference.write_text('{"run_id":"sy_stale"}', encoding="utf-8")
|
|
now = [0.0]
|
|
with self.assertRaisesRegex(
|
|
RuntimeError, "timeout waiting for correlated android-generation-v1.json"
|
|
):
|
|
worker.wait_json(
|
|
reference, 0.2,
|
|
lambda candidate: candidate.get("run_id") == "sy_fresh",
|
|
clock=lambda: now[0],
|
|
sleeper=lambda duration: now.__setitem__(0, now[0] + duration),
|
|
)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|