|
| 1 | +from typing import Any |
| 2 | + |
| 3 | + |
| 4 | +def mock_json_from_schema(schema: dict[str, Any]) -> Any: |
| 5 | + """Generate a mock JSON value based on a given JSON schema. |
| 6 | +
|
| 7 | + - For object schemas: returns a dict of mocked properties. |
| 8 | + - For arrays: returns a list with one mocked item. |
| 9 | + - For primitives: returns a sensible example / default / enum[0]. |
| 10 | + - Handles oneOf/anyOf by choosing the first option. |
| 11 | + - Special handling for LangChain message types. |
| 12 | + """ |
| 13 | + |
| 14 | + def _is_langchain_messages_array(sub_schema: dict[str, Any]) -> bool: |
| 15 | + """Check if this is a LangChain messages array.""" |
| 16 | + if sub_schema.get("type") != "array": |
| 17 | + return False |
| 18 | + items = sub_schema.get("items", {}) |
| 19 | + if not isinstance(items, dict): |
| 20 | + return False |
| 21 | + # Check if it has oneOf with message types |
| 22 | + one_of = items.get("oneOf", []) |
| 23 | + if not one_of: |
| 24 | + return False |
| 25 | + # Look for HumanMessage or similar patterns |
| 26 | + for option in one_of: |
| 27 | + if isinstance(option, dict): |
| 28 | + title = option.get("title", "") |
| 29 | + if "Message" in title: |
| 30 | + return True |
| 31 | + return False |
| 32 | + |
| 33 | + def _mock_langchain_human_message() -> dict[str, Any]: |
| 34 | + """Generate a mock HumanMessage for LangChain.""" |
| 35 | + return {"type": "human", "content": "What's the weather like today?"} |
| 36 | + |
| 37 | + def _mock_value( |
| 38 | + sub_schema: dict[str, Any], required: bool = True, parent_key: str = "" |
| 39 | + ) -> Any: |
| 40 | + # 1) Default wins |
| 41 | + if "default" in sub_schema: |
| 42 | + return sub_schema["default"] |
| 43 | + |
| 44 | + # 2) Special handling for LangChain messages array |
| 45 | + if parent_key == "messages" and _is_langchain_messages_array(sub_schema): |
| 46 | + return [_mock_langchain_human_message()] |
| 47 | + |
| 48 | + # 3) Handle oneOf/anyOf - pick the first option (or HumanMessage if available) |
| 49 | + if "oneOf" in sub_schema and isinstance(sub_schema["oneOf"], list): |
| 50 | + if sub_schema["oneOf"]: |
| 51 | + # Try to find HumanMessage first |
| 52 | + for option in sub_schema["oneOf"]: |
| 53 | + if ( |
| 54 | + isinstance(option, dict) |
| 55 | + and option.get("title") == "HumanMessage" |
| 56 | + ): |
| 57 | + return _mock_value(option, required) |
| 58 | + # Otherwise use first option |
| 59 | + return _mock_value(sub_schema["oneOf"][0], required) |
| 60 | + return None |
| 61 | + |
| 62 | + if "anyOf" in sub_schema and isinstance(sub_schema["anyOf"], list): |
| 63 | + if sub_schema["anyOf"]: |
| 64 | + return _mock_value(sub_schema["anyOf"][0], required) |
| 65 | + return None |
| 66 | + |
| 67 | + t = sub_schema.get("type") |
| 68 | + |
| 69 | + # 4) Enums: pick the first option |
| 70 | + enum = sub_schema.get("enum") |
| 71 | + if enum and isinstance(enum, list): |
| 72 | + return enum[0] |
| 73 | + |
| 74 | + # 5) Handle const values |
| 75 | + if "const" in sub_schema: |
| 76 | + return sub_schema["const"] |
| 77 | + |
| 78 | + # 6) Objects: recurse into mock_json_from_schema |
| 79 | + if t == "object": |
| 80 | + if "properties" not in sub_schema: |
| 81 | + return {} |
| 82 | + return mock_json_from_schema(sub_schema) |
| 83 | + |
| 84 | + # 7) Arrays: mock a single item based on "items" schema |
| 85 | + if t == "array": |
| 86 | + item_schema = sub_schema.get("items", {}) |
| 87 | + # If items is not a dict, just return empty list |
| 88 | + if not isinstance(item_schema, dict): |
| 89 | + return [] |
| 90 | + return [_mock_value(item_schema, required=True)] |
| 91 | + |
| 92 | + # 8) Primitives |
| 93 | + if t == "string": |
| 94 | + # Check for specific titles that might indicate what example to use |
| 95 | + title = sub_schema.get("title", "").lower() |
| 96 | + if "content" in title: |
| 97 | + return "What's the weather like today?" |
| 98 | + # If there's a format, we could specialize later (email, date, etc.) |
| 99 | + return "example" if required else "" |
| 100 | + |
| 101 | + if t == "integer": |
| 102 | + return 0 |
| 103 | + |
| 104 | + if t == "number": |
| 105 | + return 0.0 |
| 106 | + |
| 107 | + if t == "boolean": |
| 108 | + return True if required else False |
| 109 | + |
| 110 | + # 9) Fallback |
| 111 | + return None |
| 112 | + |
| 113 | + # Top-level: if it's an object with properties, build a dict |
| 114 | + if schema.get("type") == "object": |
| 115 | + if "properties" not in schema: |
| 116 | + return {} |
| 117 | + |
| 118 | + props: dict[str, Any] = schema.get("properties", {}) |
| 119 | + required_keys = set(schema.get("required", [])) |
| 120 | + result: dict[str, Any] = {} |
| 121 | + |
| 122 | + for key, prop_schema in props.items(): |
| 123 | + if not isinstance(prop_schema, dict): |
| 124 | + continue |
| 125 | + is_required = key in required_keys |
| 126 | + result[key] = _mock_value(prop_schema, required=is_required, parent_key=key) |
| 127 | + |
| 128 | + return result |
| 129 | + |
| 130 | + # If it's not an object schema, just mock the value directly |
| 131 | + return _mock_value(schema, required=True) |
0 commit comments