1import json2import threading3import unittest4from http.client import HTTPConnection5from pathlib import Path6import importlib.util789MODULE_PATH = Path(__file__).with_name("model.py")10SPEC = importlib.util.spec_from_file_location("tiny_agent_lab_model", MODULE_PATH)11model = importlib.util.module_from_spec(SPEC)12assert SPEC.loader is not None13SPEC.loader.exec_module(model)141516class ModelFixtureTests(unittest.TestCase):17 def test_hello_uses_latest_user_marker(self):18 result = model.complete({"model": "tiny-lab", "messages": [19 {"role": "user", "content": "old LAB_HELLO_old"},20 {"role": "assistant", "content": "done"},21 {"role": "user", "content": "LAB_HELLO_new"},22 ]})23 self.assertEqual(result["choices"][0]["message"]["content"], "LAB_DONE LAB_HELLO_new")2425 def test_question_requires_advertised_tool_and_returns_canonical_batch(self):26 body = {"messages": [{"role": "user", "content": "LAB_QUESTION_123"}], "tools": [{"type": "function", "function": {"name": "clarify"}}]}27 call = model.complete(body)["choices"][0]["message"]["tool_calls"][0]28 self.assertEqual(call["function"]["name"], "clarify")29 args = json.loads(call["function"]["arguments"])30 self.assertEqual(args["questions"][0]["choices"], ["Continue", "Stop"])31 self.assertFalse(args["questions"][0]["multi_select"])32 error = model.complete({"messages": body["messages"]})["choices"][0]["message"]["content"]33 self.assertIn("clarify tool is unavailable", error)3435 def test_reply_quote_does_not_replace_current_scenario(self):36 body = {"messages": [{"role": "user", "content":37 '[Replying to your previous message: "LAB_DONE LAB_HELLO_old"]\n\nLAB_QUESTION_new'}],38 "tools": [{"type": "function", "function": {"name": "clarify"}}]}39 message = model.complete(body)["choices"][0]["message"]40 arguments = json.loads(message["tool_calls"][0]["function"]["arguments"])41 self.assertIn("LAB_QUESTION_new", arguments["questions"][0]["question"])4243 def test_multi_and_approval_tool_results_complete_turn(self):44 multi = model.complete({"messages": [{"role": "user", "content": "LAB_MULTI_123"}], "tools": [{"type": "function", "function": {"name": "clarify"}}]})45 args = json.loads(multi["choices"][0]["message"]["tool_calls"][0]["function"]["arguments"])46 self.assertTrue(args["questions"][0]["multi_select"])47 approval = model.complete({"messages": [{"role": "user", "content": "LAB_APPROVAL_123"}], "tools": [{"type": "function", "function": {"name": "execute_code"}}]})48 approval_call = approval["choices"][0]["message"]["tool_calls"][0]49 self.assertEqual(json.loads(approval_call["function"]["arguments"])["code"], "print('tinyagent-lab')")50 done = model.complete({"messages": [51 {"role": "user", "content": "LAB_APPROVAL_123"},52 {"role": "assistant", "content": None, "tool_calls": [approval_call]},53 {"role": "tool", "tool_call_id": approval_call["id"], "content": '{"stdout":"tinyagent-lab\\n"}'},54 ]})55 self.assertIn("tinyagent-lab", done["choices"][0]["message"]["content"])5657 def test_mention_requires_explicit_operator_hex_and_media_is_visible(self):58 operator = "a" * 6459 mentioned = model.complete({"messages": [{"role": "user", "content": f"LAB_MENTION_123 @{operator}"}]})60 self.assertEqual(mentioned["choices"][0]["message"]["content"], f"LAB_DONE LAB_MENTION_123 @{operator}")61 absent = model.complete({"messages": [{"role": "user", "content": "LAB_MENTION_124 @operator"}]})62 self.assertEqual(absent["choices"][0]["message"]["content"], "LAB_DONE LAB_MENTION_124")63 media = model.complete({"messages": [{"role": "user", "content": "LAB_MEDIA_123 [file: tinyagent-attachment-notes.txt]"}]})64 self.assertEqual(media["choices"][0]["message"]["content"], "LAB_DONE LAB_MEDIA_123 MEDIA_RECEIVED tinyagent-attachment-notes.txt")6566 def test_http_health_models_and_sse(self):67 server = model.ThreadingHTTPServer(("127.0.0.1", 0), model.Handler)68 threading.Thread(target=server.serve_forever, daemon=True).start()69 try:70 conn = HTTPConnection("127.0.0.1", server.server_port)71 conn.request("GET", "/healthz")72 self.assertEqual(conn.getresponse().status, 200)73 conn.close()74 conn = HTTPConnection("127.0.0.1", server.server_port)75 body = json.dumps({"model": "tiny-lab", "stream": True, "messages": [{"role": "user", "content": "LAB_HELLO_sse"}]}).encode()76 conn.request("POST", "/v1/chat/completions", body, {"Content-Type": "application/json"})77 response = conn.getresponse()78 payload = response.read().decode()79 self.assertEqual(response.status, 200)80 self.assertIn('LAB_DONE LAB_HELLO_sse', payload)81 self.assertIn("data: [DONE]", payload)82 conn.close()83 finally:84 server.shutdown()85 server.server_close()868788if __name__ == "__main__":89 unittest.main()