rogee
d9e57c4900
- delete GatewayHTTPServer/GatewayHandler, create_server, /healthz and /v1/channel; event_channel module removed - load_config now requires CREATOR_PLATFORM_URL + CREATOR_GATEWAY_KEY - run() wires Gateway into OutboundAgent; SIGINT/SIGTERM stop the agent - deploy env example, install script, systemd unit docs and deployment docs updated to the outbound model; NATIVE_GATEWAY_ENDPOINT dead dev config removed
385 lines
15 KiB
Python
385 lines
15 KiB
Python
"""Outbound agent contract tests: connection, heartbeat, task dispatch, event push."""
|
||
from __future__ import annotations
|
||
|
||
import json
|
||
import threading
|
||
import time
|
||
import unittest
|
||
from importlib import import_module
|
||
from unittest.mock import Mock
|
||
|
||
import websocket
|
||
from websockets.sync.server import serve
|
||
|
||
agent_module = import_module(f"{__package__}.agent")
|
||
gateway_module = import_module(f"{__package__}.server.http")
|
||
|
||
AgentConfig = agent_module.AgentConfig
|
||
OutboundAgent = agent_module.OutboundAgent
|
||
RequestError = gateway_module.RequestError
|
||
|
||
|
||
class FakePlatform:
|
||
"""Minimal platform-side WS peer capturing frames and replaying scripted ones."""
|
||
|
||
def __init__(self, *, reject: bool = False) -> None:
|
||
self.reject = reject
|
||
self.server = serve(self._handler, "127.0.0.1", 0, process_request=self._process_request)
|
||
self.port = self.server.socket.getsockname()[1]
|
||
self.thread = threading.Thread(target=self.server.serve_forever, daemon=True)
|
||
self.thread.start()
|
||
self.received: list[dict] = []
|
||
self.lock = threading.Lock()
|
||
self.connections: list = []
|
||
self.on_task: dict[int, dict] = {}
|
||
self.reject_task_ids: set[int] = set()
|
||
self.keep_open = threading.Event()
|
||
|
||
def url(self) -> str:
|
||
return f"ws://127.0.0.1:{self.port}/v1/agent"
|
||
|
||
def _process_request(self, connection, request):
|
||
if self.reject:
|
||
from websockets.http11 import Response
|
||
return Response(403, "Forbidden", b"invalid access key\r\n")
|
||
return None
|
||
|
||
def _handler(self, ws) -> None:
|
||
with self.lock:
|
||
self.connections.append(ws)
|
||
try:
|
||
while True:
|
||
raw = ws.recv(timeout=5)
|
||
if not isinstance(raw, str):
|
||
continue
|
||
frame = json.loads(raw)
|
||
with self.lock:
|
||
self.received.append(frame)
|
||
if frame.get("type") == "task":
|
||
task_id = frame["id"]
|
||
if task_id in self.reject_task_ids:
|
||
ws.send(json.dumps({"type": "result", "id": task_id, "status": 409, "error": "duplicate task id"}))
|
||
continue
|
||
scripted = self.on_task.get(task_id)
|
||
if scripted is not None:
|
||
ws.send(json.dumps({"type": "result", "id": task_id, **scripted}))
|
||
except Exception:
|
||
pass # 连接关闭/超时即结束 handler；websockets 连接线程非 daemon，必须自然退出
|
||
|
||
def wait_for(self, predicate, timeout: float = 3.0) -> dict | None:
|
||
deadline = time.monotonic() + timeout
|
||
while time.monotonic() < deadline:
|
||
with self.lock:
|
||
for frame in self.received:
|
||
if predicate(frame):
|
||
return frame
|
||
time.sleep(0.01)
|
||
return None
|
||
|
||
def latest_connection(self):
|
||
with self.lock:
|
||
return self.connections[-1] if self.connections else None
|
||
|
||
def close(self) -> None:
|
||
self.keep_open.set()
|
||
self.server.shutdown()
|
||
for ws in list(self.connections):
|
||
try:
|
||
ws.close()
|
||
except Exception:
|
||
pass
|
||
|
||
|
||
def make_gateway() -> Mock:
|
||
gateway = Mock()
|
||
gateway.token = "unused"
|
||
manager = Mock()
|
||
# 事件桥依赖真实 Condition/版本语义；替身必须与 SubscriptionManager 同构。
|
||
manager.changed = threading.Condition()
|
||
manager.version = 0
|
||
manager._items = {}
|
||
gateway.subscriptions = manager
|
||
return gateway
|
||
|
||
|
||
def make_agent(gateway: Mock, platform: FakePlatform) -> OutboundAgent:
|
||
return OutboundAgent(gateway, AgentConfig(
|
||
platform_url=platform.url(),
|
||
access_key="agent-test-key-0001",
|
||
heartbeat_interval=0.1,
|
||
reconnect_base=0.05,
|
||
reconnect_max=0.2,
|
||
))
|
||
|
||
|
||
class OutboundAgentConnectionTests(unittest.TestCase):
|
||
def test_heartbeats_flow_with_version_until_stopped(self) -> None:
|
||
platform = FakePlatform()
|
||
agent = make_agent(make_gateway(), platform)
|
||
runner = threading.Thread(target=agent.run, daemon=True)
|
||
runner.start()
|
||
try:
|
||
beat = platform.wait_for(lambda f: f.get("type") == "heartbeat")
|
||
self.assertIsNotNone(beat, "first heartbeat missing")
|
||
self.assertEqual(beat.get("version"), agent.config.version)
|
||
second = platform.wait_for(
|
||
lambda f: f.get("type") == "heartbeat" and f is not beat, timeout=3.0
|
||
)
|
||
self.assertIsNotNone(second, "periodic heartbeat missing")
|
||
finally:
|
||
agent.stop()
|
||
runner.join(timeout=5)
|
||
platform.close()
|
||
self.assertFalse(runner.is_alive())
|
||
|
||
def test_auth_failure_is_reported_and_retried_with_backoff(self) -> None:
|
||
platform = FakePlatform(reject=True)
|
||
agent = make_agent(make_gateway(), platform)
|
||
runner = threading.Thread(target=agent.run, daemon=True)
|
||
runner.start()
|
||
try:
|
||
deadline = time.monotonic() + 3.0
|
||
while time.monotonic() < deadline and agent.reconnect_attempts < 2:
|
||
time.sleep(0.01)
|
||
self.assertGreaterEqual(agent.reconnect_attempts, 2, "auth failure must trigger retry")
|
||
self.assertTrue(agent.last_error, "auth failure must be reported")
|
||
finally:
|
||
agent.stop()
|
||
runner.join(timeout=5)
|
||
platform.close()
|
||
|
||
|
||
class OutboundAgentTaskTests(unittest.TestCase):
|
||
def test_task_dispatches_route_and_returns_result(self) -> None:
|
||
platform = FakePlatform()
|
||
gateway = make_gateway()
|
||
gateway.info.return_value = {"service": "browser-gateway", "node_id": "node-a"}
|
||
agent = make_agent(gateway, platform)
|
||
runner = threading.Thread(target=agent.run, daemon=True)
|
||
runner.start()
|
||
try:
|
||
self.wait_until_connected(agent)
|
||
platform.latest_connection().send(json.dumps({
|
||
"type": "task", "id": 7, "method": "GET", "path": "/v1/info", "payload": None,
|
||
}))
|
||
result = platform.wait_for(lambda f: f.get("type") == "result" and f.get("id") == 7)
|
||
self.assertIsNotNone(result, "task result missing")
|
||
self.assertEqual(result.get("status"), 200)
|
||
import base64
|
||
self.assertEqual(
|
||
json.loads(base64.b64decode(result.get("body"))),
|
||
{"service": "browser-gateway", "node_id": "node-a"},
|
||
)
|
||
finally:
|
||
agent.stop()
|
||
runner.join(timeout=5)
|
||
platform.close()
|
||
|
||
def test_unknown_route_returns_explicit_404_result(self) -> None:
|
||
platform = FakePlatform()
|
||
gateway = make_gateway()
|
||
agent = make_agent(gateway, platform)
|
||
runner = threading.Thread(target=agent.run, daemon=True)
|
||
runner.start()
|
||
try:
|
||
self.wait_until_connected(agent)
|
||
platform.latest_connection().send(json.dumps({
|
||
"type": "task", "id": 8, "method": "GET", "path": "/v1/nothing", "payload": None,
|
||
}))
|
||
result = platform.wait_for(lambda f: f.get("type") == "result" and f.get("id") == 8)
|
||
self.assertIsNotNone(result)
|
||
self.assertEqual(result.get("status"), 404)
|
||
finally:
|
||
agent.stop()
|
||
runner.join(timeout=5)
|
||
platform.close()
|
||
|
||
def test_duplicate_task_id_is_rejected_idempotently(self) -> None:
|
||
platform = FakePlatform()
|
||
gateway = make_gateway()
|
||
gateway.list_browsers.return_value = []
|
||
agent = make_agent(gateway, platform)
|
||
runner = threading.Thread(target=agent.run, daemon=True)
|
||
runner.start()
|
||
try:
|
||
self.wait_until_connected(agent)
|
||
frame = {"type": "task", "id": 9, "method": "GET", "path": "/v1/browsers", "payload": None}
|
||
platform.latest_connection().send(json.dumps(frame))
|
||
first = platform.wait_for(lambda f: f.get("type") == "result" and f.get("id") == 9)
|
||
self.assertIsNotNone(first)
|
||
platform.latest_connection().send(json.dumps(frame))
|
||
second = platform.wait_for(
|
||
lambda f: f.get("type") == "result" and f.get("id") == 9 and f is not first,
|
||
timeout=3.0,
|
||
)
|
||
self.assertIsNotNone(second)
|
||
self.assertEqual(second.get("status"), 409)
|
||
self.assertIn("duplicate", second.get("error", ""))
|
||
finally:
|
||
agent.stop()
|
||
runner.join(timeout=5)
|
||
platform.close()
|
||
|
||
def test_route_exception_maps_to_error_result(self) -> None:
|
||
platform = FakePlatform()
|
||
gateway = make_gateway()
|
||
gateway.list_browsers.side_effect = ValueError("invalid alias")
|
||
agent = make_agent(gateway, platform)
|
||
runner = threading.Thread(target=agent.run, daemon=True)
|
||
runner.start()
|
||
try:
|
||
self.wait_until_connected(agent)
|
||
platform.latest_connection().send(json.dumps({
|
||
"type": "task", "id": 10, "method": "GET", "path": "/v1/browsers", "payload": None,
|
||
}))
|
||
result = platform.wait_for(lambda f: f.get("type") == "result" and f.get("id") == 10)
|
||
self.assertIsNotNone(result)
|
||
self.assertEqual(result.get("status"), 400)
|
||
self.assertIn("invalid alias", result.get("error", ""))
|
||
finally:
|
||
agent.stop()
|
||
runner.join(timeout=5)
|
||
platform.close()
|
||
|
||
@staticmethod
|
||
def wait_until_connected(agent: OutboundAgent, timeout: float = 5.0) -> None:
|
||
deadline = time.monotonic() + timeout
|
||
while time.monotonic() < deadline and not agent.connected.is_set():
|
||
time.sleep(0.01)
|
||
if not agent.connected.is_set():
|
||
raise AssertionError("agent did not connect")
|
||
|
||
|
||
class OutboundAgentEventTests(unittest.TestCase):
|
||
def test_pending_events_are_pushed_and_ack_removes_them(self) -> None:
|
||
platform = FakePlatform()
|
||
gateway = make_gateway()
|
||
manager = gateway.subscriptions
|
||
manager.version = 1
|
||
item = Mock()
|
||
item.error = ""
|
||
item.stopped = False
|
||
item.pending.return_value = [
|
||
{"delivery_id": "d-1", "kind": "notice"},
|
||
{"delivery_id": "d-2", "kind": "checkpoint"},
|
||
]
|
||
manager._items["alias-a"] = item
|
||
manager._get.return_value = item
|
||
agent = make_agent(gateway, platform)
|
||
runner = threading.Thread(target=agent.run, daemon=True)
|
||
runner.start()
|
||
try:
|
||
OutboundAgentTaskTests.wait_until_connected(agent)
|
||
batch = platform.wait_for(lambda f: f.get("type") == "event")
|
||
self.assertIsNotNone(batch, "event batch missing")
|
||
self.assertEqual(batch["event"]["alias"], "alias-a")
|
||
self.assertEqual(len(batch["event"]["deliveries"]), 2)
|
||
platform.latest_connection().send(json.dumps({
|
||
"type": "event_ack", "alias": "alias-a", "delivery_ids": ["d-1", "d-2"],
|
||
}))
|
||
deadline = time.monotonic() + 3.0
|
||
while time.monotonic() < deadline and manager.ack.call_count == 0:
|
||
time.sleep(0.01)
|
||
manager.ack.assert_called_once()
|
||
args = manager.ack.call_args
|
||
self.assertEqual(args.args[0], "alias-a")
|
||
self.assertEqual(sorted(args.args[1]), ["d-1", "d-2"])
|
||
finally:
|
||
agent.stop()
|
||
runner.join(timeout=5)
|
||
platform.close()
|
||
|
||
def test_nack_resets_unconfirmed_batch_for_repush(self) -> None:
|
||
platform = FakePlatform()
|
||
gateway = make_gateway()
|
||
manager = gateway.subscriptions
|
||
manager.changed = threading.Condition()
|
||
item = Mock()
|
||
item.error = ""
|
||
item.stopped = False
|
||
item.pending.return_value = [{"delivery_id": "d-9", "kind": "notice"}]
|
||
manager._items["alias-a"] = item
|
||
manager._get.return_value = item
|
||
agent = make_agent(gateway, platform)
|
||
runner = threading.Thread(target=agent.run, daemon=True)
|
||
runner.start()
|
||
try:
|
||
OutboundAgentTaskTests.wait_until_connected(agent)
|
||
first = platform.wait_for(lambda f: f.get("type") == "event")
|
||
self.assertIsNotNone(first)
|
||
platform.latest_connection().send(json.dumps({
|
||
"type": "event_nack", "subscription": first["event"]["subscription"], "alias": "alias-a",
|
||
}))
|
||
repush = platform.wait_for(
|
||
lambda f: f.get("type") == "event" and f is not first, timeout=3.0
|
||
)
|
||
self.assertIsNotNone(repush, "nacked batch was not re-pushed")
|
||
self.assertEqual([d["delivery_id"] for d in repush["event"]["deliveries"]], ["d-9"])
|
||
finally:
|
||
agent.stop()
|
||
runner.join(timeout=5)
|
||
platform.close()
|
||
|
||
def test_reconnect_clears_unconfirmed_state_and_repushes(self) -> None:
|
||
platform = FakePlatform()
|
||
gateway = make_gateway()
|
||
manager = gateway.subscriptions
|
||
manager.changed = threading.Condition()
|
||
item = Mock()
|
||
item.error = ""
|
||
item.stopped = False
|
||
item.pending.return_value = [{"delivery_id": "d-7", "kind": "notice"}]
|
||
manager._items["alias-a"] = item
|
||
manager._get.return_value = item
|
||
agent = make_agent(gateway, platform)
|
||
runner = threading.Thread(target=agent.run, daemon=True)
|
||
runner.start()
|
||
try:
|
||
OutboundAgentTaskTests.wait_until_connected(agent)
|
||
first = platform.wait_for(lambda f: f.get("type") == "event")
|
||
self.assertIsNotNone(first)
|
||
# 断开连接：agent 应退避重连，并重推未确认批次
|
||
first_connection = platform.latest_connection()
|
||
first_connection.close()
|
||
deadline = time.monotonic() + 5.0
|
||
repush = None
|
||
while time.monotonic() < deadline:
|
||
repush = platform.wait_for(
|
||
lambda f: f.get("type") == "event" and f is not first, timeout=0.2
|
||
)
|
||
if repush is not None:
|
||
break
|
||
self.assertIsNotNone(repush, "unconfirmed batch not re-pushed after reconnect")
|
||
finally:
|
||
agent.stop()
|
||
runner.join(timeout=5)
|
||
platform.close()
|
||
|
||
def test_subscription_error_is_surfaced_as_error_frame(self) -> None:
|
||
platform = FakePlatform()
|
||
gateway = make_gateway()
|
||
manager = gateway.subscriptions
|
||
manager.changed = threading.Condition()
|
||
item = Mock()
|
||
item.error = "browser identity changed"
|
||
item.stopped = False
|
||
item.pending.return_value = []
|
||
manager._items["alias-a"] = item
|
||
manager._get.return_value = item
|
||
agent = make_agent(gateway, platform)
|
||
runner = threading.Thread(target=agent.run, daemon=True)
|
||
runner.start()
|
||
try:
|
||
OutboundAgentTaskTests.wait_until_connected(agent)
|
||
error = platform.wait_for(lambda f: f.get("type") == "error" and f.get("alias"))
|
||
self.assertIsNotNone(error, "subscription error frame missing")
|
||
self.assertIn("identity changed", error.get("error", ""))
|
||
finally:
|
||
agent.stop()
|
||
runner.join(timeout=5)
|
||
platform.close()
|
||
|
||
|
||
if __name__ == "__main__":
|
||
unittest.main()
|