Something went wrong. Try again.
This repository has no description
Something went wrong. Try again.
9.3 kB · 194 lines
Python
at main
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195"""End to end over real sockets: game -> sidecar -> (fake) OpenRouter."""import jsonimport threadingimport unittestimport urllib.errorimport urllib.requestfrom dataclasses import replacefrom http.server import BaseHTTPRequestHandler, ThreadingHTTPServerfrom pathlib import Path
from sidecar.config import load_settingsfrom sidecar.jev_client import JevClient, JevError, parse_choicefrom sidecar.server import make_handler
class FakeOpenRouter(BaseHTTPRequestHandler): """Answers every question with a scripted choice, or 'none'/'other'/'both'.""" script = {} status = 200 seen = []
def log_message(self, *args): pass
def do_POST(self): body = json.loads(self.rfile.read(int(self.headers["Content-Length"]))) FakeOpenRouter.seen.append((self.headers, body)) answers = {} for name, question in body["questions"].items(): options = list(question["criteria"]) fallback = next(o for o in ("none", "other", "both", options[0]) if o in options) choice = FakeOpenRouter.script.get(name, fallback) answers[name] = { "choice": choice, "confidence": 0.93, "probabilities": {o: (1.0 if o == choice else 0.0) for o in options}, } data = json.dumps({"answers": answers}).encode() self.send_response(FakeOpenRouter.status) self.send_header("Content-Type", "application/json") self.send_header("Content-Length", str(len(data))) self.end_headers() self.wfile.write(data)
def serve(handler): server = ThreadingHTTPServer(("127.0.0.1", 0), handler) threading.Thread(target=server.serve_forever, daemon=True).start() return server
def post(url, body, content_type): request = urllib.request.Request(url, data=body.encode(), method="POST", headers={"Content-Type": content_type}) with urllib.request.urlopen(request, timeout=5) as response: return response.status, response.read().decode()
class WireTests(unittest.TestCase): def setUp(self): FakeOpenRouter.script, FakeOpenRouter.status, FakeOpenRouter.seen = {}, 200, [] self.upstream = serve(FakeOpenRouter) self.settings = replace( load_settings(env_path=Path("does-not-exist.env"), environ={}), endpoint="http://127.0.0.1:%d/api/alpha/decisions" % self.upstream.server_address[1], ) client = JevClient("test-key", self.settings.model, self.settings.endpoint, 5) self.sidecar = serve(make_handler(client, self.settings)) self.base = "http://127.0.0.1:%d" % self.sidecar.server_address[1]
def tearDown(self): for server in (self.sidecar, self.upstream): server.shutdown() server.server_close()
def test_game_endpoint_returns_five_ints_per_act(self): FakeOpenRouter.script = {"act": "explain", "addressee": "trip", "explAdv": "controlling", "reference": "rockyMarriage"} status, body = post(self.base + "/da", "text\tTrip you are so controlling\ncontext\tBigQuestion\nsaid\tGRACE: Well?\n", "text/plain") self.assertEqual(status, 200) self.assertEqual(body, "23 1 100 -1 -1\n10 -1 61 147 -1\n43 -1 -1 -1 -1\n")
def test_upstream_request_is_the_decisions_shape(self): post(self.base + "/da", "text\thello\ncontext\tGreeting\n", "text/plain") headers, body = FakeOpenRouter.seen[0] self.assertEqual(headers["Authorization"], "Bearer test-key") self.assertEqual(body["model"], "~typesafe/jev-latest") self.assertEqual(body["state"]["player_said"], "hello") self.assertEqual(body["state"]["what_is_happening"], ["Greeting"]) self.assertEqual(body["questions"]["act"]["type"], "choice")
def test_game_endpoint_reports_failure_so_the_game_can_fall_back(self): FakeOpenRouter.status = 500 with self.assertRaises(urllib.error.HTTPError) as raised: post(self.base + "/da", "text\thello\n", "text/plain") self.assertEqual(raised.exception.code, 502)
def test_json_endpoint(self): FakeOpenRouter.script = {"act": "greet", "addressee": "grace"} status, body = post(self.base + "/classify", json.dumps({"text": "hi grace"}), "application/json") payload = json.loads(body) self.assertEqual(status, 200) self.assertTrue(payload["understood"]) self.assertEqual((payload["acts"][0]["da_id"], payload["acts"][0]["char_id"]), (30, 0))
def test_monitor_page_and_history(self): FakeOpenRouter.script = {"act": "greet"} post(self.base + "/da", "text\thello\nsaid\tTRIP: Hey!\n", "text/plain") FakeOpenRouter.status = 500 with self.assertRaises(urllib.error.HTTPError): post(self.base + "/da", "text\tbroken\n", "text/plain") with urllib.request.urlopen(self.base + "/", timeout=5) as response: self.assertIn("jev-facade monitor", response.read().decode()) with urllib.request.urlopen(self.base + "/history", timeout=5) as response: history = json.loads(response.read()) self.assertEqual([e["text"] for e in history], ["hello", "broken"]) self.assertEqual((history[0]["source"], history[0]["said"]), ("game", ["TRIP: Hey!"])) self.assertEqual(history[0]["answers"]["act"]["top"][0], ["greet", 1.0]) self.assertIn("HTTP 500", history[1]["error"])
def test_event_stream_pushes_new_lines(self): FakeOpenRouter.script = {"act": "thank"} stream = urllib.request.urlopen(self.base + "/events", timeout=5) try: post(self.base + "/da", "text\tthanks\n", "text/plain") line = stream.readline().decode() finally: stream.close() self.assertTrue(line.startswith("data: ")) self.assertEqual(json.loads(line[6:])["acts"][0]["name"], "thank")
def test_posts_from_web_pages_are_refused(self): # A page the player has open could otherwise spend their API credit. for headers in ({"Origin": "https://evil.example"}, {"Host": "evil.example"}): request = urllib.request.Request(self.base + "/classify", data=b'{"text": "hi"}', method="POST", headers=headers) with self.assertRaises(urllib.error.HTTPError) as raised: urllib.request.urlopen(request, timeout=5) self.assertEqual(raised.exception.code, 403) self.assertEqual(FakeOpenRouter.seen, [])
def test_the_monitor_pages_own_origin_is_allowed(self): request = urllib.request.Request(self.base + "/classify", data=b'{"text": "hi"}', method="POST", headers={"Origin": self.base}) with urllib.request.urlopen(request, timeout=5) as response: self.assertEqual(response.status, 200)
def test_reads_under_a_foreign_host_name_are_refused(self): # DNS rebinding would otherwise let a web page read what the player typed. for path in ("/history", "/events", "/"): request = urllib.request.Request(self.base + path, headers={"Host": "evil.example"}) with self.assertRaises(urllib.error.HTTPError) as raised: urllib.request.urlopen(request, timeout=5) self.assertEqual(raised.exception.code, 403)
def test_bad_content_length_is_not_fatal(self): request = urllib.request.Request(self.base + "/da", data=b"", method="POST", headers={"Content-Length": "banana"}) # Read as an empty body: an empty line, answered without asking Jev. with urllib.request.urlopen(request, timeout=5) as response: self.assertEqual((response.status, response.read()), (200, b"43 -1 -1 -1 -1\n")) self.assertEqual(FakeOpenRouter.seen, [])
def test_malformed_json_requests_get_a_400_not_a_crash(self): for body in ('{"text": null}', '{"contexts": [1]}', "not json"): with self.assertRaises(urllib.error.HTTPError) as raised: post(self.base + "/classify", body, "application/json") self.assertEqual(raised.exception.code, 400) status, _ = post(self.base + "/classify", '{"text": "hi", "contexts": [1, null]}', "application/json") self.assertEqual(status, 200)
def test_empty_key_is_refused_up_front(self): with self.assertRaises(JevError): JevClient("", "m", "http://x", 1)
class ParseChoiceTests(unittest.TestCase): def test_confidence_is_optional(self): answer = parse_choice("act", {"choice": "a", "probabilities": {"a": 0.7, "b": 0.3}}, {"a", "b"}) self.assertIsNone(answer.confidence) self.assertEqual(answer.probability("a"), 0.7)
def test_choice_must_have_a_probability(self): with self.assertRaises(JevError): parse_choice("act", {"choice": "a", "probabilities": {"b": 1.0}}, {"a", "b"})
def test_choice_must_be_one_that_was_offered(self): with self.assertRaises(JevError): parse_choice("act", {"choice": "z", "probabilities": {"z": 1.0}}, {"a", "b"})
if __name__ == "__main__": unittest.main()