57 lines
3.3 KiB
Python
57 lines
3.3 KiB
Python
"""Local synthetic mock-server tests, no model calls or secrets."""
|
|
import http.client
|
|
import json
|
|
import threading
|
|
import time
|
|
import unittest
|
|
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
|
|
import bridge
|
|
|
|
KEY = b'synthetic-only-unit-test-secret'
|
|
class Upstream(BaseHTTPRequestHandler):
|
|
calls = 0
|
|
slow = False
|
|
entered = threading.Event()
|
|
def log_message(self, *args): pass
|
|
def do_GET(self):
|
|
body = b'{}'; self.send_response(200); self.send_header('Content-Length','2'); self.end_headers(); self.wfile.write(body)
|
|
def do_POST(self):
|
|
type(self).calls += 1
|
|
type(self).entered.set()
|
|
self.rfile.read(int(self.headers['Content-Length']))
|
|
if type(self).slow: time.sleep(0.25)
|
|
assert self.headers['Authorization'] == 'Bearer ' + KEY.decode()
|
|
body = b'{"text":"synthetic-only-canary"}'
|
|
self.send_response(200);self.send_header('Content-Length',str(len(body)));self.end_headers();self.wfile.write(body)
|
|
|
|
class BridgeTests(unittest.TestCase):
|
|
def setUp(self):
|
|
Upstream.calls=0;Upstream.slow=False;Upstream.entered.clear()
|
|
self.u=ThreadingHTTPServer(('127.0.0.1',0),Upstream)
|
|
self.b=bridge.Server(('127.0.0.1',0),bridge.Handler,key=KEY,upstream=self.u.server_address)
|
|
for s in [self.u,self.b]: threading.Thread(target=s.serve_forever,daemon=True).start()
|
|
def tearDown(self):
|
|
for s in [self.b,self.u]:s.shutdown();s.server_close()
|
|
def call(self,body=b'x',headers=None,path='/v1/audio/transcriptions',method='POST'):
|
|
h={'Authorization':'Bearer '+KEY.decode(),'Content-Type':'multipart/form-data; boundary=test'}
|
|
if headers: h.update(headers)
|
|
c=http.client.HTTPConnection(*self.b.server_address,timeout=2);c.request(method,path,body,h);r=c.getresponse();result=(r.status,r.read());c.close();return result
|
|
def test_valid_fixed_upstream(self):
|
|
self.assertEqual(self.call(),(200,b'{"text":"synthetic-only-canary"}'));self.assertEqual(Upstream.calls,1)
|
|
def test_auth_denied_without_upstream(self):
|
|
self.assertEqual(self.call(headers={'Authorization':'Bearer wrong'})[0],401);self.assertEqual(Upstream.calls,0)
|
|
def test_body_bound_before_read(self):
|
|
self.assertEqual(self.call(headers={'Content-Length':str(bridge.MAX_BODY+1)})[0],413);self.assertEqual(Upstream.calls,0)
|
|
def test_fixed_path_only(self):
|
|
self.assertEqual(self.call(path='/v1/chat/completions')[0],404);self.assertEqual(Upstream.calls,0)
|
|
def test_concurrency_exactly_one(self):
|
|
Upstream.slow=True;out=[];t=threading.Thread(target=lambda:out.append(self.call()[0]));t.start();self.assertTrue(Upstream.entered.wait(1));self.assertEqual(self.call()[0],429);t.join();self.assertEqual(out,[200]);self.assertEqual(Upstream.calls,1)
|
|
def test_rate_429_no_retry(self):
|
|
self.b.gate=bridge.Gate(rate=0,burst=1);self.assertEqual(self.call()[0],200);self.assertEqual(self.call()[0],429);self.assertEqual(Upstream.calls,1)
|
|
def test_upstream_down_no_retry(self):
|
|
self.b.upstream=('127.0.0.1',1);self.assertEqual(self.call()[0],502);self.assertEqual(Upstream.calls,0)
|
|
def test_wrong_content_type(self):
|
|
self.assertEqual(self.call(headers={'Content-Type':'application/json'})[0],415);self.assertEqual(Upstream.calls,0)
|
|
|
|
if __name__=='__main__':unittest.main(verbosity=2)
|