Spaces:
Running
Running
File size: 4,675 Bytes
6918275 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 | """Real loopback HTTP JSON-RPC transport; no external service or model keys."""
import asyncio
import json
import threading
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
import pytest
from plugins.argus_supervisor.helpers.a2a_client import OpenTrainingA2AClient
from plugins.argus_supervisor.helpers.heartbeat import Supervisor
from plugins.argus_supervisor.helpers.models import SupervisorError
@pytest.mark.asyncio
@pytest.mark.parametrize("terminal", ["completed", "failed", "input-required"])
async def test_submit_poll_terminal_recovery_over_http(setup, terminal):
s, remote, argus = setup
observed_headers = []
class Handler(BaseHTTPRequestHandler):
def do_GET(self):
self.reply(asyncio.run(remote.transport("GET", self.path, None)))
def do_POST(self):
observed_headers.append(self.headers.get("Authorization"))
payload = json.loads(self.rfile.read(int(self.headers["Content-Length"])))
self.reply(asyncio.run(remote.transport("POST", "", payload)))
def reply(self, result):
body = json.dumps(result).encode()
self.send_response(200)
self.send_header("Content-Type", "application/json")
self.send_header("Content-Length", str(len(body)))
self.end_headers()
self.wfile.write(body)
def log_message(self, *args):
pass
server = ThreadingHTTPServer(("127.0.0.1", 0), Handler)
thread = threading.Thread(target=server.serve_forever, daemon=True)
thread.start()
try:
client = OpenTrainingA2AClient("https://test.example/a2a", token="test-key")
# Production constructor refuses HTTP. Test alone substitutes a loopback
# transport address to exercise actual aiohttp without external TLS/DNS.
client.url = f"http://127.0.0.1:{server.server_port}"
s.client = client
await s.discover()
c = s.store.load("campaign-1")
c.remote_task_id = None
await s.mutation(c, "claim_control", {}, None)
await s.mutation(
c, "submit", {"text": "Harmless fixture task"}, c.controller_epoch
)
await s.tick()
remote.task["status"]["state"] = terminal
remote.task["metadata"]["argus_supervisor"]["revision"] = "2"
await s.tick()
count = len(argus.calls)
restarted = Supervisor(s.store, client, argus)
await restarted.tick()
assert len(argus.calls) == count
assert s.store.load(c.campaign_id).remote_state == terminal
assert len([x for x in remote.calls if x[0] == "message/send"]) == 1
assert observed_headers and all(
h == "Bearer test-key" for h in observed_headers
)
# Remote takes over: persisted local lease must no longer permit mutation.
remote.lease["epoch"] += 1
await restarted.tick()
c = s.store.load(c.campaign_id)
assert c.mode == "observer"
with pytest.raises(SupervisorError):
await restarted.mutation(c, "cancel", {"id": "task-1"}, c.controller_epoch)
finally:
server.shutdown()
server.server_close()
thread.join(timeout=2)
@pytest.mark.asyncio
async def test_read_only_status_uses_canonical_rpc_mount_root():
paths = []
class Handler(BaseHTTPRequestHandler):
def do_POST(self):
paths.append(self.path)
payload = json.loads(self.rfile.read(int(self.headers['Content-Length'])))
if self.path != '/a2a/':
self.send_error(405)
return
body = json.dumps({'jsonrpc': '2.0', 'id': payload['id'],
'result': {'id': 'existing-task', 'status': {'state': 'completed'}}}).encode()
self.send_response(200)
self.send_header('Content-Type', 'application/json')
self.send_header('Content-Length', str(len(body)))
self.end_headers()
self.wfile.write(body)
def log_message(self, *args):
pass
server = ThreadingHTTPServer(('127.0.0.1', 0), Handler)
thread = threading.Thread(target=server.serve_forever, daemon=True)
thread.start()
try:
client = OpenTrainingA2AClient('https://test.example/a2a')
client.url = f'http://127.0.0.1:{server.server_port}/a2a'
client.capabilities = {'status': 'tasks/get'}
result = await client.status('existing-task')
assert result['state'] == 'completed'
assert paths == ['/a2a/']
finally:
server.shutdown()
server.server_close()
thread.join(timeout=2)
|