92 lines
5.4 KiB
Python
92 lines
5.4 KiB
Python
import http.client
|
|
import json
|
|
import sys
|
|
import tempfile
|
|
import threading
|
|
import socket
|
|
from concurrent.futures import ThreadPoolExecutor
|
|
from unittest.mock import patch
|
|
import unittest
|
|
from pathlib import Path
|
|
sys.path.insert(0,str(Path(__file__).resolve().parents[1]/'coordinator'))
|
|
from robot_bt_coordinator.service import Coordinator
|
|
from robot_bt_coordinator.backends import ManualBackend
|
|
from robot_bt_coordinator.http_api import make_server
|
|
|
|
class HttpTest(unittest.TestCase):
|
|
def setUp(self):
|
|
self.tmp=tempfile.TemporaryDirectory();self.c=Coordinator(str(Path(self.tmp.name)/'db'),ManualBackend(),{'robot_01'})
|
|
self.server=make_server(self.c,'127.0.0.1',0,'test-secret','operator-secret')
|
|
self.thread=threading.Thread(target=self.server.serve_forever,daemon=True);self.thread.start()
|
|
def tearDown(self):
|
|
self.server.shutdown();self.server.server_close();self.thread.join();self.c.close();self.tmp.cleanup()
|
|
def request(self,method,path,body=None,token='test-secret'):
|
|
conn=http.client.HTTPConnection('127.0.0.1',self.server.server_port,timeout=3)
|
|
headers={'Authorization':'Bearer '+token,'Content-Type':'application/json'}
|
|
conn.request(method,path,body,headers);r=conn.getresponse();result=(r.status,json.loads(r.read()));conn.close();return result
|
|
def test_auth_required_and_health_allowed(self):
|
|
self.assertEqual(self.request('GET','/v1/capabilities',token='wrong')[0],401)
|
|
self.assertEqual(self.request('GET','/healthz',token='wrong')[0],200)
|
|
def test_submit_get_cancel_and_event_cursor(self):
|
|
req=dict(client_request_id='http1',robot_id='robot_01',instruction='move water',known_info={})
|
|
code,t=self.request('POST','/v1/tasks',json.dumps(req));self.assertEqual(code,202)
|
|
tid=t['task_id'];self.assertEqual(self.request('GET','/v1/tasks/'+tid)[1]['status'],'QUEUED')
|
|
self.assertEqual(self.request('POST',f'/v1/tasks/{tid}/cancel','{}')[1]['status'],'CANCELED')
|
|
self.assertEqual(self.request('GET',f'/v1/tasks/{tid}/events?after=0')[0],200)
|
|
def test_duplicate_json_keys_nonfinite_and_unknown_endpoint_rejected(self):
|
|
for payload in ['{"robot_id":"a","robot_id":"b"}','{"quantity":NaN}']:
|
|
self.assertEqual(self.request('POST','/v1/tasks',payload)[0],400)
|
|
self.assertEqual(self.request('GET','/not-found')[0],404)
|
|
|
|
def test_listener_can_queue_ten_fresh_connections_before_accept(self):
|
|
# Model a burst while the accept loop is busy. This is a capacity check,
|
|
# not a throughput or latency assertion on the host running the suite.
|
|
listener=make_server(self.c,'127.0.0.1',0,'test-secret')
|
|
clients=[]
|
|
try:
|
|
for _ in range(10):
|
|
try:
|
|
clients.append(socket.create_connection(('127.0.0.1',listener.server_port),timeout=2))
|
|
except OSError as error:
|
|
self.fail('listener could not queue ten clients: '+str(error))
|
|
finally:
|
|
for client in clients:client.close()
|
|
listener.server_close()
|
|
|
|
def test_ten_concurrent_submissions_are_independently_accepted_and_deduplicated(self):
|
|
gate=threading.Barrier(10)
|
|
def submit(index):
|
|
body=json.dumps(dict(client_request_id='burst-'+str(index),robot_id='robot_01',instruction='move water',known_info={}))
|
|
gate.wait(timeout=10)
|
|
return self.request('POST','/v1/tasks',body)
|
|
with ThreadPoolExecutor(max_workers=10) as pool:
|
|
results=list(pool.map(submit,range(10)))
|
|
self.assertEqual([code for code,_ in results],[202]*10)
|
|
self.assertEqual(len({task['task_id'] for _,task in results}),10)
|
|
self.assertTrue(all(task['status']=='QUEUED' for _,task in results))
|
|
with ThreadPoolExecutor(max_workers=10) as pool:
|
|
repeats=list(pool.map(submit,range(10)))
|
|
self.assertEqual([task['task_id'] for _,task in repeats],[task['task_id'] for _,task in results])
|
|
self.assertTrue(all(code==202 and task['deduplicated'] for code,task in repeats))
|
|
self.assertEqual(len(self.c.store.all()),10)
|
|
|
|
def test_recovery_backend_failure_is_sanitized_service_unavailable(self):
|
|
with patch.object(self.c,'intervene',side_effect=RuntimeError('private-token-123 from ROS')):
|
|
code,body=self.request('POST','/v1/tasks/task/interventions','{}',token='operator-secret')
|
|
self.assertEqual(code,503);self.assertEqual(body['error_code'],'RECONCILIATION_UNAVAILABLE')
|
|
self.assertNotIn('private-token',json.dumps(body))
|
|
|
|
def test_recovery_timeout_returns_json_instead_of_dropping_connection(self):
|
|
with patch.object(self.c,'intervene',side_effect=TimeoutError('private-token-123 outcome unknown')):
|
|
try:code,body=self.request('POST','/v1/tasks/task/interventions','{}',token='operator-secret')
|
|
except http.client.RemoteDisconnected:self.fail('recovery timeout dropped the HTTP connection without a JSON result')
|
|
self.assertEqual(code,504);self.assertEqual(body['error_code'],'RECONCILIATION_TIMEOUT')
|
|
self.assertNotIn('private-token',json.dumps(body))
|
|
|
|
def test_recovery_backend_is_not_called_with_ordinary_api_token(self):
|
|
with patch.object(self.c,'intervene') as reconcile:
|
|
self.assertEqual(self.request('POST','/v1/tasks/task/interventions','{}')[0],401)
|
|
reconcile.assert_not_called()
|
|
|
|
if __name__=='__main__':unittest.main()
|