Buckets:
| """Unit tests for the shared python backend helpers (python_utils.py). | |
| Run standalone (Python standard library only, no backend venv needed): | |
| python3 -m unittest python_utils_test | |
| These mirror the server-less helper tests in backend/python/mlx/test.py | |
| (TestSharedHelpers), but live here so they run on any platform: the mlx | |
| test module imports grpc/backend_pb2 at import time and needs the MLX venv, | |
| whereas python_utils has no third-party dependency. Proto Message objects | |
| are faked with types.SimpleNamespace (real proto fields default to ""). | |
| """ | |
| import json | |
| import types | |
| import unittest | |
| from python_utils import attach_media_parts, messages_to_dicts, parse_options | |
| def _msg(**fields): | |
| """Fake a proto Message: every unset field is the empty string, as protobuf.""" | |
| defaults = { | |
| "role": "", | |
| "content": "", | |
| "name": "", | |
| "tool_call_id": "", | |
| "reasoning_content": "", | |
| "tool_calls": "", | |
| } | |
| defaults.update(fields) | |
| return types.SimpleNamespace(**defaults) | |
| class TestParseOptions(unittest.TestCase): | |
| def test_type_inference(self): | |
| opts = parse_options( | |
| ["temperature:0.7", "max_tokens:128", "trust:true", "name:hello", "no_colon_skipped"] | |
| ) | |
| self.assertEqual(opts["temperature"], 0.7) | |
| self.assertEqual(opts["max_tokens"], 128) | |
| self.assertIs(opts["trust"], True) | |
| self.assertEqual(opts["name"], "hello") | |
| self.assertNotIn("no_colon_skipped", opts) | |
| class TestMessagesToDicts(unittest.TestCase): | |
| def test_basic_fields(self): | |
| out = messages_to_dicts( | |
| [ | |
| _msg(role="user", content="hi"), | |
| _msg(role="tool", content="42", tool_call_id="call_1", name="f"), | |
| ] | |
| ) | |
| self.assertEqual(out[0], {"role": "user", "content": "hi"}) | |
| self.assertEqual(out[1]["tool_call_id"], "call_1") | |
| self.assertEqual(out[1]["name"], "f") | |
| def test_tool_call_arguments_string_decoded_to_mapping(self): | |
| # OpenAI wire format ships function.arguments as a JSON *string*; chat | |
| # templates iterate it as a mapping, so it must come back as a dict. | |
| out = messages_to_dicts( | |
| [ | |
| _msg( | |
| role="assistant", | |
| tool_calls=json.dumps( | |
| [ | |
| { | |
| "id": "call_1", | |
| "type": "function", | |
| "function": { | |
| "name": "get_weather", | |
| "arguments": '{"location": "Rome"}', | |
| }, | |
| } | |
| ] | |
| ), | |
| ) | |
| ] | |
| ) | |
| args = out[0]["tool_calls"][0]["function"]["arguments"] | |
| self.assertEqual(args, {"location": "Rome"}) | |
| self.assertEqual(dict(args.items()), {"location": "Rome"}) | |
| def test_tool_call_arguments_already_mapping_is_idempotent(self): | |
| out = messages_to_dicts( | |
| [ | |
| _msg( | |
| role="assistant", | |
| tool_calls=json.dumps( | |
| [{"function": {"name": "f", "arguments": {"a": 1}}}] | |
| ), | |
| ) | |
| ] | |
| ) | |
| self.assertEqual(out[0]["tool_calls"][0]["function"]["arguments"], {"a": 1}) | |
| def test_tool_call_arguments_invalid_json_left_as_string(self): | |
| out = messages_to_dicts( | |
| [ | |
| _msg( | |
| role="assistant", | |
| tool_calls=json.dumps( | |
| [{"function": {"name": "f", "arguments": "not-json"}}] | |
| ), | |
| ) | |
| ] | |
| ) | |
| self.assertEqual(out[0]["tool_calls"][0]["function"]["arguments"], "not-json") | |
| def test_tool_call_without_function_key(self): | |
| out = messages_to_dicts( | |
| [_msg(role="assistant", tool_calls=json.dumps([{"id": "call_1"}]))] | |
| ) | |
| self.assertEqual(out[0]["tool_calls"], [{"id": "call_1"}]) | |
| def test_tool_calls_invalid_json_dropped(self): | |
| out = messages_to_dicts([_msg(role="assistant", tool_calls="{not json")]) | |
| self.assertNotIn("tool_calls", out[0]) | |
| class TestAttachMediaParts(unittest.TestCase): | |
| def test_image_marker_added_to_last_user_turn(self): | |
| messages = [ | |
| {"role": "system", "content": "be brief"}, | |
| {"role": "user", "content": "first"}, | |
| {"role": "assistant", "content": "ok"}, | |
| {"role": "user", "content": "how high is the water?"}, | |
| ] | |
| out = attach_media_parts(messages, n_images=1) | |
| self.assertEqual( | |
| out[3]["content"], | |
| [{"type": "image"}, {"type": "text", "text": "how high is the water?"}], | |
| ) | |
| # Earlier turns and the input list itself are untouched. | |
| self.assertEqual(out[:3], messages[:3]) | |
| self.assertEqual(messages[3]["content"], "how high is the water?") | |
| def test_counts_and_order_images_then_videos(self): | |
| out = attach_media_parts( | |
| [{"role": "user", "content": "describe"}], n_images=2, n_videos=1 | |
| ) | |
| self.assertEqual( | |
| out[0]["content"], | |
| [ | |
| {"type": "image"}, | |
| {"type": "image"}, | |
| {"type": "video"}, | |
| {"type": "text", "text": "describe"}, | |
| ], | |
| ) | |
| def test_empty_text_yields_media_only_parts(self): | |
| out = attach_media_parts([{"role": "user", "content": ""}], n_images=1) | |
| self.assertEqual(out[0]["content"], [{"type": "image"}]) | |
| def test_other_message_keys_are_preserved(self): | |
| out = attach_media_parts( | |
| [{"role": "user", "content": "hi", "name": "bob"}], n_images=1 | |
| ) | |
| self.assertEqual(out[0]["name"], "bob") | |
| def test_no_media_is_a_no_op(self): | |
| self.assertIsNone(attach_media_parts([{"role": "user", "content": "hi"}])) | |
| def test_no_user_turn_is_a_no_op(self): | |
| self.assertIsNone( | |
| attach_media_parts([{"role": "system", "content": "hi"}], n_images=1) | |
| ) | |
| def test_content_already_parts_is_a_no_op(self): | |
| self.assertIsNone( | |
| attach_media_parts( | |
| [{"role": "user", "content": [{"type": "text", "text": "hi"}]}], | |
| n_images=1, | |
| ) | |
| ) | |
| if __name__ == "__main__": | |
| unittest.main() | |
Xet Storage Details
- Size:
- 6.51 kB
- Xet hash:
- 3bd5f0c26fea2964b12795d98bdabf8cd85b516e91a4c4a8f466a9c7b202b3cd
·
Xet efficiently stores files, intelligently splitting them into unique chunks and accelerating uploads and downloads. More info.