53 lines
2.6 KiB
Python
53 lines
2.6 KiB
Python
"""Trusted transport protocol tests; HTTP mocked with no network."""
|
|
import json
|
|
import asyncio
|
|
import functools
|
|
|
|
def async_test(func):
|
|
@functools.wraps(func)
|
|
def wrapper(*args, **kwargs):
|
|
return asyncio.run(func(*args, **kwargs))
|
|
return wrapper
|
|
import pytest
|
|
import httpx
|
|
from control.accounts import Accounts
|
|
from control.chat_runs import HermesRuns
|
|
|
|
@async_test
|
|
async def test_native_runs_transport_exact_idempotency_and_session(tmp_path):
|
|
accounts = Accounts(str(tmp_path/'accounts.db'))
|
|
user = accounts.create('test-user','Test','admin','offline-password')
|
|
accounts.update(user['id'], status='ready', hermes_url='http://127.0.0.1:20000', hermes_api_key='private-key')
|
|
seen = []
|
|
def handler(request):
|
|
seen.append(request)
|
|
if request.url.path.endswith('/events'):
|
|
return httpx.Response(200, text='data: {"event":"message.delta","delta":"Hi"}\n\ndata: {"event":"run.completed","output":"Hi"}\n\n')
|
|
return httpx.Response(202, json={'run_id':'native-run','status':'started'})
|
|
async with httpx.AsyncClient(transport=httpx.MockTransport(handler)) as client:
|
|
adapter = HermesRuns(accounts, client)
|
|
payload = {'input':'Hi','session_id':'stable-session'}
|
|
assert (await adapter.start(user['id'],'hope-run',payload))['run_id'] == 'native-run'
|
|
assert seen[0].url.path == '/v1/runs'
|
|
assert seen[0].headers['authorization'] == 'Bearer private-key'
|
|
assert seen[0].headers['idempotency-key'] == 'hope-run'
|
|
assert json.loads(seen[0].content) == payload
|
|
events = [e async for e in adapter.events(user['id'],'native-run')]
|
|
assert [e['event'] for e in events] == ['message.delta','run.completed']
|
|
await adapter.stop(user['id'],'native-run')
|
|
assert seen[-1].url.path == '/v1/runs/native-run/stop'
|
|
await adapter.approve(user['id'],'native-run','approval-1','approve')
|
|
assert json.loads(seen[-1].content) == {'request_id':'approval-1','choice':'once'}
|
|
accounts.close()
|
|
|
|
@async_test
|
|
async def test_transport_rejects_deleted_account_and_bad_trusted_url(tmp_path):
|
|
accounts = Accounts(str(tmp_path/'accounts.db'))
|
|
user = accounts.create('test-user','Test','admin','offline-password')
|
|
accounts.update(user['id'], status='ready', hermes_url='http://evil:password@localhost:2', hermes_api_key='private-key')
|
|
async with httpx.AsyncClient(transport=httpx.MockTransport(lambda request: pytest.fail('Network requested'))) as client:
|
|
adapter = HermesRuns(accounts, client)
|
|
with pytest.raises(RuntimeError):
|
|
await adapter.start(user['id'],'rid',{'input':'Hi'})
|
|
accounts.close()
|