wehub-resource-sync
fed8b2eed7
Backend release / release (push) Waiting to run
Bandit Security Scan / bandit_scan (push) Waiting to run
Build and push multi-arch DocsGPT Docker image / build (linux/amd64, ubuntu-latest, amd64) (push) Waiting to run
Build and push multi-arch DocsGPT Docker image / build (linux/arm64, ubuntu-24.04-arm, arm64) (push) Waiting to run
Build and push multi-arch DocsGPT Docker image / manifest (push) Blocked by required conditions
Build and push DocsGPT FE Docker image for development / build (linux/amd64, ubuntu-latest, amd64) (push) Waiting to run
Build and push DocsGPT FE Docker image for development / build (linux/arm64, ubuntu-24.04-arm, arm64) (push) Waiting to run
Build and push DocsGPT FE Docker image for development / manifest (push) Blocked by required conditions
Python linting / ruff (push) Waiting to run
Run python tests with pytest / Run tests and count coverage (3.12) (push) Waiting to run
React Widget Build / build (push) Waiting to run
1541 lines
54 KiB
Python
1541 lines
54 KiB
Python
"""Unit tests for application/llm/google_ai.py — GoogleLLM.
|
|
|
|
Extends coverage beyond test_google_llm.py:
|
|
- _clean_messages_google: system instructions, function responses, errors
|
|
- _clean_schema: field filtering, type uppercasing, required validation
|
|
- _clean_tools_format: empty properties, required fields
|
|
- _extract_preview_from_message: various message shapes
|
|
- _summarize_messages_for_log
|
|
- _get_text_value / _is_thought_part: dict vs object forms
|
|
- _raw_gen with tools and response_schema
|
|
- _raw_gen_stream: function_call parts, thought parts, error handling
|
|
- prepare_structured_output_format: comprehensive type mapping
|
|
- prepare_messages_with_attachments: error handling
|
|
- _upload_file_to_google
|
|
- get_supported_attachment_types
|
|
"""
|
|
|
|
import types
|
|
|
|
import pytest
|
|
|
|
from application.llm.google_ai import GoogleLLM
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Fake types module for Google AI
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class _FakePart:
|
|
def __init__(self, text=None, function_call=None, file_data=None, inline_data=None, thought=False, **kwargs):
|
|
self.text = text
|
|
self.function_call = function_call or kwargs.get("functionCall")
|
|
self.file_data = file_data
|
|
self.inline_data = inline_data
|
|
self.thought = thought
|
|
self.thoughtSignature = kwargs.get("thoughtSignature")
|
|
|
|
@staticmethod
|
|
def from_text(text):
|
|
return _FakePart(text=text)
|
|
|
|
@staticmethod
|
|
def from_function_call(name, args):
|
|
return _FakePart(function_call=types.SimpleNamespace(name=name, args=args))
|
|
|
|
@staticmethod
|
|
def from_function_response(name, response):
|
|
return _FakePart(text=str(response))
|
|
|
|
@staticmethod
|
|
def from_uri(file_uri, mime_type):
|
|
return _FakePart(
|
|
file_data=types.SimpleNamespace(file_uri=file_uri, mime_type=mime_type)
|
|
)
|
|
|
|
@staticmethod
|
|
def from_bytes(data, mime_type):
|
|
return _FakePart(
|
|
inline_data=types.SimpleNamespace(data=data, mime_type=mime_type)
|
|
)
|
|
|
|
|
|
class _FakeContent:
|
|
def __init__(self, role, parts):
|
|
self.role = role
|
|
self.parts = parts
|
|
|
|
|
|
class FakeTypesModule:
|
|
Part = _FakePart
|
|
Content = _FakeContent
|
|
|
|
class GenerateContentConfig:
|
|
def __init__(self, thinking_config=None, **_kw):
|
|
self.system_instruction = None
|
|
self.tools = None
|
|
self.thinking_config = thinking_config
|
|
self.response_schema = None
|
|
self.response_mime_type = None
|
|
|
|
class ThinkingConfig:
|
|
def __init__(self, include_thoughts=False, thinking_level=None):
|
|
self.include_thoughts = include_thoughts
|
|
self.thinking_level = thinking_level
|
|
|
|
class Tool:
|
|
def __init__(self, function_declarations=None):
|
|
self.function_declarations = function_declarations or []
|
|
|
|
class FunctionCall:
|
|
def __init__(self, name=None, args=None):
|
|
self.name = name
|
|
self.args = args
|
|
|
|
|
|
class FakeModels:
|
|
def __init__(self):
|
|
self.last_kwargs = None
|
|
|
|
class _Resp:
|
|
def __init__(self, text=None, candidates=None):
|
|
self.text = text
|
|
self.candidates = candidates or []
|
|
|
|
def generate_content(self, *args, **kwargs):
|
|
self.last_kwargs = kwargs
|
|
return FakeModels._Resp(text="ok")
|
|
|
|
def generate_content_stream(self, *args, **kwargs):
|
|
self.last_kwargs = kwargs
|
|
return []
|
|
|
|
|
|
class FakeClientFiles:
|
|
def upload(self, file=None):
|
|
return types.SimpleNamespace(uri="gs://fake-uri")
|
|
|
|
|
|
class FakeClient:
|
|
def __init__(self, *a, **kw):
|
|
self.models = FakeModels()
|
|
self.files = FakeClientFiles()
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def patch_google(monkeypatch):
|
|
import application.llm.google_ai as gmod
|
|
|
|
monkeypatch.setattr(gmod, "types", FakeTypesModule)
|
|
monkeypatch.setattr(gmod.genai, "Client", FakeClient)
|
|
|
|
|
|
@pytest.fixture
|
|
def llm():
|
|
instance = GoogleLLM(api_key="test-key")
|
|
instance.storage = types.SimpleNamespace(
|
|
file_exists=lambda p: True,
|
|
process_file=lambda path, fn, **kw: fn(path),
|
|
)
|
|
return instance
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# _clean_messages_google
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@pytest.mark.unit
|
|
class TestCleanMessagesGoogle:
|
|
|
|
def test_system_message_extracted_as_instruction(self, llm):
|
|
msgs = [
|
|
{"role": "system", "content": "You are helpful"},
|
|
{"role": "user", "content": "hi"},
|
|
]
|
|
cleaned, sys_instr = llm._clean_messages_google(msgs)
|
|
assert sys_instr == "You are helpful"
|
|
assert all(c.role != "system" for c in cleaned)
|
|
|
|
def test_multiple_system_messages_joined(self, llm):
|
|
msgs = [
|
|
{"role": "system", "content": "Rule 1"},
|
|
{"role": "system", "content": "Rule 2"},
|
|
{"role": "user", "content": "hi"},
|
|
]
|
|
_, sys_instr = llm._clean_messages_google(msgs)
|
|
assert "Rule 1" in sys_instr
|
|
assert "Rule 2" in sys_instr
|
|
|
|
def test_system_list_content(self, llm):
|
|
msgs = [
|
|
{"role": "system", "content": [{"text": "A"}, {"text": "B"}]},
|
|
{"role": "user", "content": "hi"},
|
|
]
|
|
_, sys_instr = llm._clean_messages_google(msgs)
|
|
assert "A" in sys_instr and "B" in sys_instr
|
|
|
|
def test_assistant_role_becomes_model(self, llm):
|
|
msgs = [{"role": "assistant", "content": "hi"}]
|
|
cleaned, _ = llm._clean_messages_google(msgs)
|
|
assert cleaned[0].role == "model"
|
|
|
|
def test_tool_role_becomes_model(self, llm):
|
|
msgs = [{"role": "tool", "content": "result"}]
|
|
cleaned, _ = llm._clean_messages_google(msgs)
|
|
assert cleaned[0].role == "model"
|
|
|
|
def test_function_call_in_content_list(self, llm):
|
|
msgs = [
|
|
{
|
|
"role": "assistant",
|
|
"content": [
|
|
{"function_call": {"name": "fn", "args": {"x": 1}}},
|
|
],
|
|
}
|
|
]
|
|
cleaned, _ = llm._clean_messages_google(msgs)
|
|
assert len(cleaned) == 1
|
|
assert any(
|
|
hasattr(p, "function_call") and p.function_call is not None
|
|
for p in cleaned[0].parts
|
|
)
|
|
|
|
def test_function_response_in_content_list(self, llm):
|
|
msgs = [
|
|
{
|
|
"role": "assistant",
|
|
"content": [
|
|
{
|
|
"function_response": {
|
|
"name": "fn",
|
|
"response": {"result": 42},
|
|
}
|
|
},
|
|
],
|
|
}
|
|
]
|
|
cleaned, _ = llm._clean_messages_google(msgs)
|
|
assert len(cleaned) == 1
|
|
|
|
def test_files_in_content_list(self, llm):
|
|
msgs = [
|
|
{
|
|
"role": "user",
|
|
"content": [
|
|
{"files": [{"file_uri": "gs://f", "mime_type": "image/png"}]},
|
|
],
|
|
}
|
|
]
|
|
cleaned, _ = llm._clean_messages_google(msgs)
|
|
assert len(cleaned) == 1
|
|
assert any(
|
|
hasattr(p, "file_data") and p.file_data is not None
|
|
for p in cleaned[0].parts
|
|
)
|
|
|
|
def test_files_with_inline_bytes(self, llm):
|
|
msgs = [
|
|
{
|
|
"role": "user",
|
|
"content": [
|
|
{
|
|
"files": [
|
|
{"file_bytes": b"\x89PNG", "mime_type": "image/png"}
|
|
]
|
|
},
|
|
],
|
|
}
|
|
]
|
|
cleaned, _ = llm._clean_messages_google(msgs)
|
|
assert len(cleaned) == 1
|
|
inline_parts = [
|
|
p for p in cleaned[0].parts
|
|
if getattr(p, "inline_data", None) is not None
|
|
]
|
|
assert len(inline_parts) == 1
|
|
assert inline_parts[0].inline_data.data == b"\x89PNG"
|
|
assert inline_parts[0].inline_data.mime_type == "image/png"
|
|
|
|
def test_files_with_empty_uri_dropped(self, llm):
|
|
msgs = [
|
|
{
|
|
"role": "user",
|
|
"content": [
|
|
{"files": [{"file_uri": "", "mime_type": "image/png"}]},
|
|
],
|
|
}
|
|
]
|
|
cleaned, _ = llm._clean_messages_google(msgs)
|
|
# Empty URI part is dropped; no other parts means the whole
|
|
# content is empty and the message itself is not appended.
|
|
assert cleaned == []
|
|
|
|
def test_unexpected_list_item_raises(self, llm):
|
|
msgs = [{"role": "user", "content": [{"unknown_key": "val"}]}]
|
|
with pytest.raises(ValueError, match="Unexpected content dictionary"):
|
|
llm._clean_messages_google(msgs)
|
|
|
|
def test_unexpected_content_type_raises(self, llm):
|
|
msgs = [{"role": "user", "content": 12345}]
|
|
with pytest.raises(ValueError, match="Unexpected content type"):
|
|
llm._clean_messages_google(msgs)
|
|
|
|
def test_no_system_instruction_returns_none(self, llm):
|
|
msgs = [{"role": "user", "content": "hi"}]
|
|
_, sys_instr = llm._clean_messages_google(msgs)
|
|
assert sys_instr is None
|
|
|
|
def test_empty_parts_skipped(self, llm):
|
|
msgs = [{"role": "user", "content": None}]
|
|
cleaned, _ = llm._clean_messages_google(msgs)
|
|
assert len(cleaned) == 0
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# _clean_schema
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@pytest.mark.unit
|
|
class TestCleanSchema:
|
|
|
|
def test_type_uppercased(self, llm):
|
|
result = llm._clean_schema({"type": "string"})
|
|
assert result["type"] == "STRING"
|
|
|
|
def test_unsupported_fields_removed(self, llm):
|
|
result = llm._clean_schema({"type": "string", "title": "Name", "$ref": "#/x"})
|
|
assert "title" not in result
|
|
assert "$ref" not in result
|
|
assert result["type"] == "STRING"
|
|
|
|
def test_nested_properties_cleaned(self, llm):
|
|
# _clean_schema recursively cleans the properties dict value.
|
|
# Property names that happen to match allowed_fields survive.
|
|
# This tests the recursive cleaning on schema values.
|
|
schema = {
|
|
"type": "object",
|
|
"properties": {
|
|
"type": {"type": "string"},
|
|
},
|
|
}
|
|
result = llm._clean_schema(schema)
|
|
# "type" is in allowed_fields, so the property survives as a key
|
|
# Its value gets uppercased since it's a type field
|
|
assert "properties" in result
|
|
assert result["properties"]["type"]["type"] == "STRING"
|
|
|
|
def test_required_validated_against_properties(self, llm):
|
|
# Property names must be in allowed_fields to survive _clean_schema
|
|
# "type" is in allowed_fields so it survives as a property key
|
|
schema = {
|
|
"type": "object",
|
|
"properties": {"type": {"type": "string"}},
|
|
"required": ["type", "nonexistent"],
|
|
}
|
|
result = llm._clean_schema(schema)
|
|
assert result["required"] == ["type"]
|
|
|
|
def test_required_removed_when_no_valid_entries(self, llm):
|
|
schema = {
|
|
"type": "object",
|
|
"properties": {"type": {"type": "string"}},
|
|
"required": ["nonexistent"],
|
|
}
|
|
result = llm._clean_schema(schema)
|
|
assert "required" not in result
|
|
|
|
def test_required_removed_when_no_properties(self, llm):
|
|
schema = {"type": "string", "required": ["x"]}
|
|
result = llm._clean_schema(schema)
|
|
assert "required" not in result
|
|
|
|
def test_non_dict_passthrough(self, llm):
|
|
assert llm._clean_schema("hello") == "hello"
|
|
assert llm._clean_schema(42) == 42
|
|
|
|
def test_list_items_cleaned(self, llm):
|
|
schema = {
|
|
"type": "array",
|
|
"items": {"type": "string", "title": "ignored"},
|
|
}
|
|
result = llm._clean_schema(schema)
|
|
assert "title" not in result["items"]
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# _clean_tools_format
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@pytest.mark.unit
|
|
class TestCleanToolsFormat:
|
|
|
|
def test_basic_tool_conversion(self, llm):
|
|
tools = [
|
|
{
|
|
"type": "function",
|
|
"function": {
|
|
"name": "search",
|
|
"description": "Search the web",
|
|
"parameters": {
|
|
"type": "object",
|
|
"properties": {
|
|
"query": {"type": "string"},
|
|
},
|
|
"required": ["query"],
|
|
},
|
|
},
|
|
}
|
|
]
|
|
result = llm._clean_tools_format(tools)
|
|
assert len(result) == 1
|
|
assert hasattr(result[0], "function_declarations")
|
|
|
|
def test_tool_without_properties(self, llm):
|
|
tools = [
|
|
{
|
|
"type": "function",
|
|
"function": {
|
|
"name": "ping",
|
|
"description": "Ping server",
|
|
"parameters": {"type": "object", "properties": {}},
|
|
},
|
|
}
|
|
]
|
|
result = llm._clean_tools_format(tools)
|
|
assert len(result) == 1
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# _extract_preview_from_message / _summarize_messages_for_log
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@pytest.mark.unit
|
|
class TestMessagePreviewAndSummary:
|
|
|
|
def test_preview_from_parts_text(self, llm):
|
|
msg = types.SimpleNamespace(
|
|
parts=[_FakePart(text="hello world")]
|
|
)
|
|
preview = llm._extract_preview_from_message(msg)
|
|
assert preview == "hello world"
|
|
|
|
def test_preview_from_function_call_part(self, llm):
|
|
fc = types.SimpleNamespace(name="search")
|
|
msg = types.SimpleNamespace(
|
|
parts=[_FakePart(function_call=fc)]
|
|
)
|
|
preview = llm._extract_preview_from_message(msg)
|
|
assert "search" in preview
|
|
|
|
def test_preview_from_dict_string_content(self, llm):
|
|
msg = {"content": "dict content"}
|
|
preview = llm._extract_preview_from_message(msg)
|
|
assert preview == "dict content"
|
|
|
|
def test_preview_from_dict_list_content(self, llm):
|
|
msg = {"content": [{"text": "list text"}]}
|
|
preview = llm._extract_preview_from_message(msg)
|
|
assert preview == "list text"
|
|
|
|
def test_preview_from_dict_function_call(self, llm):
|
|
msg = {"content": [{"function_call": {"name": "fn"}}]}
|
|
preview = llm._extract_preview_from_message(msg)
|
|
assert "fn" in preview
|
|
|
|
def test_preview_from_dict_function_response(self, llm):
|
|
msg = {"content": [{"function_response": {"name": "fn_resp"}}]}
|
|
preview = llm._extract_preview_from_message(msg)
|
|
assert "fn_resp" in preview
|
|
|
|
def test_preview_fallback_to_str(self, llm):
|
|
msg = 42
|
|
preview = llm._extract_preview_from_message(msg)
|
|
assert preview == "42"
|
|
|
|
def test_summarize_messages_empty(self, llm):
|
|
result = llm._summarize_messages_for_log([])
|
|
assert "count=0" in result
|
|
|
|
def test_summarize_messages_truncates(self, llm):
|
|
msgs = [
|
|
types.SimpleNamespace(parts=[_FakePart(text="a" * 100)])
|
|
]
|
|
result = llm._summarize_messages_for_log(msgs, preview_chars=10)
|
|
assert "..." in result
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# _get_text_value / _is_thought_part
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@pytest.mark.unit
|
|
class TestStaticHelpers:
|
|
|
|
def test_get_text_value_dict(self):
|
|
assert GoogleLLM._get_text_value({"text": "hi"}) == "hi"
|
|
|
|
def test_get_text_value_dict_no_text(self):
|
|
assert GoogleLLM._get_text_value({"other": "x"}) == ""
|
|
|
|
def test_get_text_value_dict_non_string(self):
|
|
assert GoogleLLM._get_text_value({"text": 42}) == ""
|
|
|
|
def test_get_text_value_object(self):
|
|
obj = types.SimpleNamespace(text="obj_text")
|
|
assert GoogleLLM._get_text_value(obj) == "obj_text"
|
|
|
|
def test_get_text_value_object_no_text(self):
|
|
obj = types.SimpleNamespace()
|
|
assert GoogleLLM._get_text_value(obj) == ""
|
|
|
|
def test_is_thought_part_dict_true(self):
|
|
assert GoogleLLM._is_thought_part({"thought": True}) is True
|
|
|
|
def test_is_thought_part_dict_false(self):
|
|
assert GoogleLLM._is_thought_part({"thought": False}) is False
|
|
|
|
def test_is_thought_part_object(self):
|
|
obj = types.SimpleNamespace(thought=True)
|
|
assert GoogleLLM._is_thought_part(obj) is True
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# _raw_gen
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@pytest.mark.unit
|
|
class TestRawGen:
|
|
|
|
def test_returns_text(self, llm):
|
|
msgs = [{"role": "user", "content": "hi"}]
|
|
result = llm._raw_gen(llm, model="gemini-2.0", messages=msgs)
|
|
assert result == "ok"
|
|
|
|
def test_with_tools_returns_response(self, llm):
|
|
tools = [
|
|
{
|
|
"type": "function",
|
|
"function": {
|
|
"name": "t",
|
|
"description": "d",
|
|
"parameters": {"type": "object", "properties": {}},
|
|
},
|
|
}
|
|
]
|
|
msgs = [{"role": "user", "content": "hi"}]
|
|
result = llm._raw_gen(llm, model="gemini", messages=msgs, tools=tools)
|
|
assert hasattr(result, "text")
|
|
|
|
def test_with_response_schema(self, llm):
|
|
msgs = [{"role": "user", "content": "hi"}]
|
|
llm._raw_gen(
|
|
llm,
|
|
model="gemini",
|
|
messages=msgs,
|
|
response_schema={"type": "OBJECT"},
|
|
)
|
|
# Should not raise
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# _raw_gen_stream
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@pytest.mark.unit
|
|
class TestRawGenStream:
|
|
|
|
def test_yields_text_from_candidates(self, llm, monkeypatch):
|
|
part = types.SimpleNamespace(
|
|
text="chunk1", function_call=None, thought=False
|
|
)
|
|
candidate = types.SimpleNamespace(
|
|
content=types.SimpleNamespace(parts=[part])
|
|
)
|
|
chunk = types.SimpleNamespace(candidates=[candidate])
|
|
|
|
monkeypatch.setattr(
|
|
FakeModels,
|
|
"generate_content_stream",
|
|
lambda self, *a, **kw: [chunk],
|
|
)
|
|
|
|
msgs = [{"role": "user", "content": "hi"}]
|
|
result = list(llm._raw_gen_stream(llm, model="gemini", messages=msgs))
|
|
assert "chunk1" in result
|
|
|
|
def test_yields_function_call_part(self, llm, monkeypatch):
|
|
fc = types.SimpleNamespace(name="search")
|
|
part = types.SimpleNamespace(
|
|
text=None, function_call=fc, thought=False
|
|
)
|
|
candidate = types.SimpleNamespace(
|
|
content=types.SimpleNamespace(parts=[part])
|
|
)
|
|
chunk = types.SimpleNamespace(candidates=[candidate])
|
|
|
|
monkeypatch.setattr(
|
|
FakeModels,
|
|
"generate_content_stream",
|
|
lambda self, *a, **kw: [chunk],
|
|
)
|
|
|
|
msgs = [{"role": "user", "content": "hi"}]
|
|
result = list(llm._raw_gen_stream(llm, model="gemini", messages=msgs))
|
|
assert any(hasattr(r, "function_call") for r in result)
|
|
|
|
def test_yields_thought_event(self, llm, monkeypatch):
|
|
part = types.SimpleNamespace(
|
|
text="thinking", function_call=None, thought=True
|
|
)
|
|
candidate = types.SimpleNamespace(
|
|
content=types.SimpleNamespace(parts=[part])
|
|
)
|
|
chunk = types.SimpleNamespace(candidates=[candidate])
|
|
|
|
monkeypatch.setattr(
|
|
FakeModels,
|
|
"generate_content_stream",
|
|
lambda self, *a, **kw: [chunk],
|
|
)
|
|
|
|
msgs = [{"role": "user", "content": "hi"}]
|
|
result = list(llm._raw_gen_stream(llm, model="gemini", messages=msgs))
|
|
assert {"type": "thought", "thought": "thinking"} in result
|
|
|
|
def test_text_only_chunk_via_hasattr(self, llm, monkeypatch):
|
|
chunk = types.SimpleNamespace(text="fallback", candidates=None, thought=False)
|
|
|
|
monkeypatch.setattr(
|
|
FakeModels,
|
|
"generate_content_stream",
|
|
lambda self, *a, **kw: [chunk],
|
|
)
|
|
|
|
msgs = [{"role": "user", "content": "hi"}]
|
|
result = list(llm._raw_gen_stream(llm, model="gemini", messages=msgs))
|
|
assert "fallback" in result
|
|
|
|
def test_stream_error_propagates(self, llm, monkeypatch):
|
|
def error_stream(self, *a, **kw):
|
|
raise RuntimeError("stream_err")
|
|
|
|
monkeypatch.setattr(FakeModels, "generate_content_stream", error_stream)
|
|
|
|
msgs = [{"role": "user", "content": "hi"}]
|
|
with pytest.raises(RuntimeError, match="stream_err"):
|
|
list(llm._raw_gen_stream(llm, model="gemini", messages=msgs))
|
|
|
|
def test_skips_empty_text_parts(self, llm, monkeypatch):
|
|
part = types.SimpleNamespace(
|
|
text="", function_call=None, thought=False
|
|
)
|
|
candidate = types.SimpleNamespace(
|
|
content=types.SimpleNamespace(parts=[part])
|
|
)
|
|
chunk = types.SimpleNamespace(candidates=[candidate])
|
|
|
|
monkeypatch.setattr(
|
|
FakeModels,
|
|
"generate_content_stream",
|
|
lambda self, *a, **kw: [chunk],
|
|
)
|
|
|
|
msgs = [{"role": "user", "content": "hi"}]
|
|
result = list(llm._raw_gen_stream(llm, model="gemini", messages=msgs))
|
|
assert result == []
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# _supports_tools / _supports_structured_output
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@pytest.mark.unit
|
|
class TestSupports:
|
|
|
|
def test_supports_tools(self, llm):
|
|
assert llm._supports_tools() is True
|
|
|
|
def test_supports_structured_output(self, llm):
|
|
assert llm._supports_structured_output() is True
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# prepare_structured_output_format
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@pytest.mark.unit
|
|
class TestPrepareStructuredOutputFormat:
|
|
|
|
def test_none_returns_none(self, llm):
|
|
assert llm.prepare_structured_output_format(None) is None
|
|
|
|
def test_type_mapping(self, llm):
|
|
schema = {
|
|
"type": "object",
|
|
"properties": {
|
|
"name": {"type": "string"},
|
|
"count": {"type": "integer"},
|
|
"score": {"type": "number"},
|
|
"active": {"type": "boolean"},
|
|
"items": {"type": "array", "items": {"type": "string"}},
|
|
},
|
|
}
|
|
result = llm.prepare_structured_output_format(schema)
|
|
assert result["type"] == "OBJECT"
|
|
assert result["properties"]["name"]["type"] == "STRING"
|
|
assert result["properties"]["count"]["type"] == "INTEGER"
|
|
assert result["properties"]["score"]["type"] == "NUMBER"
|
|
assert result["properties"]["active"]["type"] == "BOOLEAN"
|
|
assert result["properties"]["items"]["type"] == "ARRAY"
|
|
|
|
def test_property_ordering_added(self, llm):
|
|
schema = {
|
|
"type": "object",
|
|
"properties": {"a": {"type": "string"}, "b": {"type": "string"}},
|
|
}
|
|
result = llm.prepare_structured_output_format(schema)
|
|
assert "propertyOrdering" in result
|
|
assert set(result["propertyOrdering"]) == {"a", "b"}
|
|
|
|
def test_format_date_converted(self, llm):
|
|
schema = {"type": "string", "format": "date"}
|
|
result = llm.prepare_structured_output_format(schema)
|
|
assert result["format"] == "date-time"
|
|
|
|
def test_format_datetime_preserved(self, llm):
|
|
schema = {"type": "string", "format": "date-time"}
|
|
result = llm.prepare_structured_output_format(schema)
|
|
assert result["format"] == "date-time"
|
|
|
|
def test_anyof_processed(self, llm):
|
|
schema = {
|
|
"anyOf": [
|
|
{"type": "string"},
|
|
{"type": "integer"},
|
|
]
|
|
}
|
|
result = llm.prepare_structured_output_format(schema)
|
|
assert len(result["anyOf"]) == 2
|
|
assert result["anyOf"][0]["type"] == "STRING"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# get_supported_attachment_types
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@pytest.mark.unit
|
|
class TestGetSupportedAttachmentTypes:
|
|
|
|
def test_returns_list_with_expected_types(self, llm):
|
|
result = llm.get_supported_attachment_types()
|
|
assert "application/pdf" in result
|
|
assert "image/png" in result
|
|
assert "image/jpeg" in result
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# prepare_messages_with_attachments
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@pytest.mark.unit
|
|
class TestPrepareMessagesWithAttachments:
|
|
|
|
def test_no_attachments_returns_same(self, llm):
|
|
msgs = [{"role": "user", "content": "hi"}]
|
|
result = llm.prepare_messages_with_attachments(msgs)
|
|
assert result == msgs
|
|
|
|
def test_upload_error_adds_text_fallback(self, llm, monkeypatch):
|
|
monkeypatch.setattr(
|
|
llm,
|
|
"_read_attachment_bytes",
|
|
lambda a: (_ for _ in ()).throw(Exception("fail")),
|
|
)
|
|
msgs = [{"role": "user", "content": "hi"}]
|
|
attachments = [
|
|
{"mime_type": "image/png", "path": "/tmp/img.png", "content": "fallback"},
|
|
]
|
|
result = llm.prepare_messages_with_attachments(msgs, attachments)
|
|
user_msg = next(m for m in result if m["role"] == "user")
|
|
text_parts = [
|
|
p for p in user_msg["content"]
|
|
if isinstance(p, dict) and p.get("type") == "text" and "could not" in p.get("text", "").lower()
|
|
]
|
|
assert len(text_parts) == 1
|
|
|
|
def test_pdf_upload_error_adds_text_fallback(self, llm, monkeypatch):
|
|
monkeypatch.setattr(
|
|
llm,
|
|
"_upload_file_to_google",
|
|
lambda a: (_ for _ in ()).throw(Exception("fail")),
|
|
)
|
|
msgs = [{"role": "user", "content": "hi"}]
|
|
attachments = [
|
|
{"mime_type": "application/pdf", "path": "/tmp/doc.pdf", "content": "x"},
|
|
]
|
|
result = llm.prepare_messages_with_attachments(msgs, attachments)
|
|
user_msg = next(m for m in result if m["role"] == "user")
|
|
text_parts = [
|
|
p for p in user_msg["content"]
|
|
if isinstance(p, dict) and p.get("type") == "text" and "could not" in p.get("text", "").lower()
|
|
]
|
|
assert len(text_parts) == 1
|
|
|
|
def test_pdf_empty_uri_adds_text_fallback(self, llm, monkeypatch):
|
|
monkeypatch.setattr(llm, "_upload_file_to_google", lambda a: "")
|
|
msgs = [{"role": "user", "content": "hi"}]
|
|
attachments = [
|
|
{"mime_type": "application/pdf", "path": "/tmp/doc.pdf", "content": "x"},
|
|
]
|
|
result = llm.prepare_messages_with_attachments(msgs, attachments)
|
|
user_msg = next(m for m in result if m["role"] == "user")
|
|
files_entries = [
|
|
p for p in user_msg["content"] if isinstance(p, dict) and "files" in p
|
|
]
|
|
assert files_entries == []
|
|
text_parts = [
|
|
p for p in user_msg["content"]
|
|
if isinstance(p, dict) and p.get("type") == "text" and "could not" in p.get("text", "").lower()
|
|
]
|
|
assert len(text_parts) == 1
|
|
|
|
def test_image_uses_inline_bytes(self, llm, monkeypatch):
|
|
monkeypatch.setattr(llm, "_read_attachment_bytes", lambda a: b"\x89PNG-bytes")
|
|
msgs = [{"role": "user", "content": "hi"}]
|
|
attachments = [{"mime_type": "image/png", "path": "/img.png"}]
|
|
result = llm.prepare_messages_with_attachments(msgs, attachments)
|
|
user_msg = next(m for m in result if m["role"] == "user")
|
|
files_entry = next(
|
|
p for p in user_msg["content"] if isinstance(p, dict) and "files" in p
|
|
)
|
|
assert files_entry["files"] == [
|
|
{"file_bytes": b"\x89PNG-bytes", "mime_type": "image/png"}
|
|
]
|
|
|
|
def test_no_user_message_creates_one(self, llm, monkeypatch):
|
|
monkeypatch.setattr(llm, "_read_attachment_bytes", lambda a: b"png")
|
|
msgs = [{"role": "system", "content": "sys"}]
|
|
attachments = [{"mime_type": "image/png", "path": "/img.png"}]
|
|
result = llm.prepare_messages_with_attachments(msgs, attachments)
|
|
user_msgs = [m for m in result if m["role"] == "user"]
|
|
assert len(user_msgs) == 1
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# _upload_file_to_google
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@pytest.mark.unit
|
|
class TestUploadFileToGoogle:
|
|
|
|
def test_returns_cached_uri(self, llm):
|
|
attachment = {"google_file_uri": "gs://cached"}
|
|
result = llm._upload_file_to_google(attachment)
|
|
assert result == "gs://cached"
|
|
|
|
def test_empty_cached_uri_triggers_reupload(self, llm, monkeypatch):
|
|
# Poisoned-cache repro: an empty-string google_file_uri must be
|
|
# treated as a miss and re-upload, not returned as-is.
|
|
monkeypatch.setattr(
|
|
"application.llm.google_ai.settings",
|
|
types.SimpleNamespace(GOOGLE_API_KEY="k", API_KEY="k"),
|
|
)
|
|
result = llm._upload_file_to_google(
|
|
{"google_file_uri": "", "path": "/tmp/file.pdf"}
|
|
)
|
|
assert result == "gs://fake-uri"
|
|
|
|
def test_empty_upload_uri_raises(self, llm):
|
|
llm.storage = types.SimpleNamespace(
|
|
file_exists=lambda p: True,
|
|
process_file=lambda path, fn, **kw: "",
|
|
)
|
|
with pytest.raises(ValueError, match="empty URI"):
|
|
llm._upload_file_to_google({"path": "/tmp/file.pdf"})
|
|
|
|
def test_raises_for_no_path(self, llm):
|
|
with pytest.raises(ValueError, match="No file path"):
|
|
llm._upload_file_to_google({})
|
|
|
|
def test_raises_for_missing_file(self, llm):
|
|
llm.storage = types.SimpleNamespace(file_exists=lambda p: False)
|
|
with pytest.raises(FileNotFoundError):
|
|
llm._upload_file_to_google({"path": "/nonexistent"})
|
|
|
|
def test_upload_and_caches_uri(self, llm, monkeypatch):
|
|
# The attachment-id cache write goes through AttachmentsRepository
|
|
# now; failures there are swallowed with a logged warning, so the
|
|
# test just verifies the upload URI is returned end-to-end.
|
|
monkeypatch.setattr(
|
|
"application.llm.google_ai.settings",
|
|
types.SimpleNamespace(GOOGLE_API_KEY="k", API_KEY="k"),
|
|
)
|
|
result = llm._upload_file_to_google({"path": "/tmp/file.pdf", "_id": "abc"})
|
|
# process_file returns fn(path) which calls client.files.upload -> "gs://fake-uri"
|
|
assert result == "gs://fake-uri"
|
|
|
|
def test_upload_error_propagates(self, llm):
|
|
llm.storage = types.SimpleNamespace(
|
|
file_exists=lambda p: True,
|
|
process_file=lambda path, fn, **kw: (_ for _ in ()).throw(
|
|
RuntimeError("upload fail")
|
|
),
|
|
)
|
|
with pytest.raises(RuntimeError, match="upload fail"):
|
|
llm._upload_file_to_google({"path": "/tmp/file.pdf"})
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# _clean_messages_google — additional edge cases
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@pytest.mark.unit
|
|
class TestCleanMessagesGoogleAdditional:
|
|
|
|
def test_system_content_not_str_returns_empty(self, llm):
|
|
"""Cover line 168: _extract_system_text returns '' for non-str non-list."""
|
|
msgs = [
|
|
{"role": "system", "content": 42},
|
|
{"role": "user", "content": "hi"},
|
|
]
|
|
_, sys_instr = llm._clean_messages_google(msgs)
|
|
# 42 is not str and not list, so _extract_system_text returns ""
|
|
# which is falsy, so it won't be appended to system_instructions
|
|
assert sys_instr is None
|
|
|
|
def test_system_list_with_none_text_skipped(self, llm):
|
|
"""Cover line 168: items with None text are skipped."""
|
|
msgs = [
|
|
{"role": "system", "content": [{"text": None}, {"text": "valid"}]},
|
|
{"role": "user", "content": "hi"},
|
|
]
|
|
_, sys_instr = llm._clean_messages_google(msgs)
|
|
assert sys_instr == "valid"
|
|
|
|
def test_function_call_with_thought_signature(self, llm):
|
|
"""Cover lines 211 (thought_signature in function_call)."""
|
|
msgs = [
|
|
{
|
|
"role": "assistant",
|
|
"content": [
|
|
{
|
|
"function_call": {"name": "fn", "args": {"x": 1}},
|
|
"thought_signature": "sig123",
|
|
},
|
|
],
|
|
}
|
|
]
|
|
cleaned, _ = llm._clean_messages_google(msgs)
|
|
assert len(cleaned) == 1
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# _clean_schema — additional edges
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@pytest.mark.unit
|
|
class TestCleanSchemaAdditional:
|
|
|
|
def test_list_values_cleaned_recursively(self, llm):
|
|
"""Cover line 279: list values in schema are cleaned item by item."""
|
|
schema = {
|
|
"enum": ["a", "b"],
|
|
"type": "string",
|
|
}
|
|
result = llm._clean_schema(schema)
|
|
assert result["enum"] == ["a", "b"]
|
|
|
|
def test_required_validated_no_properties_key(self, llm):
|
|
"""Cover line 295: required without properties gets removed."""
|
|
schema = {"type": "string", "required": ["x"]}
|
|
result = llm._clean_schema(schema)
|
|
assert "required" not in result
|
|
|
|
def test_valid_required_empty_after_filter(self, llm):
|
|
"""Cover line 290: valid_required is non-empty.
|
|
Note: 'type' is in allowed_fields, so survives as a property key.
|
|
"""
|
|
schema = {
|
|
"type": "object",
|
|
"properties": {"type": {"type": "string"}},
|
|
"required": ["type"],
|
|
}
|
|
result = llm._clean_schema(schema)
|
|
assert result["required"] == ["type"]
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# _clean_tools_format — additional edge
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@pytest.mark.unit
|
|
class TestCleanToolsFormatAdditional:
|
|
|
|
def test_tool_with_required_in_parameters(self, llm):
|
|
"""Cover line 330: tool with required field in parameters."""
|
|
tools = [
|
|
{
|
|
"type": "function",
|
|
"function": {
|
|
"name": "search",
|
|
"description": "Search",
|
|
"parameters": {
|
|
"type": "object",
|
|
"properties": {
|
|
"query": {"type": "string"},
|
|
},
|
|
},
|
|
},
|
|
}
|
|
]
|
|
result = llm._clean_tools_format(tools)
|
|
assert len(result) == 1
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# _extract_preview_from_message — additional edges
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@pytest.mark.unit
|
|
class TestExtractPreviewAdditional:
|
|
|
|
def test_preview_from_function_response_part(self, llm):
|
|
"""Cover line 375: function_response in parts."""
|
|
fr = types.SimpleNamespace(name="resp_fn")
|
|
part = types.SimpleNamespace(
|
|
text=None,
|
|
function_call=None,
|
|
function_response=fr,
|
|
)
|
|
msg = types.SimpleNamespace(parts=[part])
|
|
preview = llm._extract_preview_from_message(msg)
|
|
assert "resp_fn" in preview
|
|
|
|
def test_preview_dict_list_with_string_item(self, llm):
|
|
"""Cover line 393-397: dict list content with string items."""
|
|
msg = {"content": ["plain string"]}
|
|
preview = llm._extract_preview_from_message(msg)
|
|
assert preview == "plain string"
|
|
|
|
def test_preview_dict_function_call_non_dict(self, llm):
|
|
"""Cover line when function_call is not a dict."""
|
|
msg = {"content": [{"function_call": "raw_string"}]}
|
|
preview = llm._extract_preview_from_message(msg)
|
|
assert preview == "function_call"
|
|
|
|
def test_preview_dict_function_response_non_dict(self, llm):
|
|
"""Cover line when function_response is not a dict."""
|
|
msg = {"content": [{"function_response": "raw_string"}]}
|
|
preview = llm._extract_preview_from_message(msg)
|
|
assert preview == "function_response"
|
|
|
|
def test_preview_dict_with_text_key_at_top_level(self, llm):
|
|
"""Cover line 375: msg has 'text' key directly."""
|
|
msg = {"text": "top level text"}
|
|
preview = llm._extract_preview_from_message(msg)
|
|
assert preview == "top level text"
|
|
|
|
def test_preview_exception_fallback(self, llm):
|
|
"""Cover line 375: exception falls back to str."""
|
|
|
|
class BadMsg:
|
|
@property
|
|
def parts(self):
|
|
raise RuntimeError("boom")
|
|
|
|
msg = BadMsg()
|
|
preview = llm._extract_preview_from_message(msg)
|
|
assert isinstance(preview, str)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# _raw_gen_stream — additional edges
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@pytest.mark.unit
|
|
class TestRawGenStreamAdditional:
|
|
|
|
def test_stream_response_close_called(self, llm, monkeypatch):
|
|
"""Cover line 524: response.close() is called in finally."""
|
|
closed = {"called": False}
|
|
|
|
class CloseableResponse:
|
|
def __iter__(self):
|
|
return iter([])
|
|
|
|
def close(self):
|
|
closed["called"] = True
|
|
|
|
monkeypatch.setattr(
|
|
FakeModels,
|
|
"generate_content_stream",
|
|
lambda self, *a, **kw: CloseableResponse(),
|
|
)
|
|
|
|
msgs = [{"role": "user", "content": "hi"}]
|
|
list(llm._raw_gen_stream(llm, model="gemini", messages=msgs))
|
|
assert closed["called"]
|
|
|
|
def test_text_chunk_via_hasattr_thought(self, llm, monkeypatch):
|
|
"""Cover lines 517: thought part via hasattr text path."""
|
|
chunk = types.SimpleNamespace(
|
|
text="thought text", candidates=None, thought=True
|
|
)
|
|
|
|
monkeypatch.setattr(
|
|
FakeModels,
|
|
"generate_content_stream",
|
|
lambda self, *a, **kw: [chunk],
|
|
)
|
|
|
|
msgs = [{"role": "user", "content": "hi"}]
|
|
result = list(llm._raw_gen_stream(llm, model="gemini", messages=msgs))
|
|
assert {"type": "thought", "thought": "thought text"} in result
|
|
|
|
def test_empty_text_chunk_via_hasattr_skipped(self, llm, monkeypatch):
|
|
"""Cover line where chunk.text is empty via hasattr path."""
|
|
chunk = types.SimpleNamespace(
|
|
text="", candidates=None, thought=False
|
|
)
|
|
|
|
monkeypatch.setattr(
|
|
FakeModels,
|
|
"generate_content_stream",
|
|
lambda self, *a, **kw: [chunk],
|
|
)
|
|
|
|
msgs = [{"role": "user", "content": "hi"}]
|
|
result = list(llm._raw_gen_stream(llm, model="gemini", messages=msgs))
|
|
assert result == []
|
|
|
|
def test_stream_with_response_schema(self, llm, monkeypatch):
|
|
"""Cover lines 470-471: response_schema in stream."""
|
|
monkeypatch.setattr(
|
|
FakeModels,
|
|
"generate_content_stream",
|
|
lambda self, *a, **kw: [],
|
|
)
|
|
msgs = [{"role": "user", "content": "hi"}]
|
|
result = list(
|
|
llm._raw_gen_stream(
|
|
llm,
|
|
model="gemini",
|
|
messages=msgs,
|
|
response_schema={"type": "OBJECT"},
|
|
)
|
|
)
|
|
assert result == []
|
|
|
|
def test_stream_with_empty_candidates(self, llm, monkeypatch):
|
|
"""Cover line 487: candidate parts None."""
|
|
chunk = types.SimpleNamespace(
|
|
candidates=[types.SimpleNamespace(content=None)]
|
|
)
|
|
monkeypatch.setattr(
|
|
FakeModels,
|
|
"generate_content_stream",
|
|
lambda self, *a, **kw: [chunk],
|
|
)
|
|
|
|
msgs = [{"role": "user", "content": "hi"}]
|
|
result = list(llm._raw_gen_stream(llm, model="gemini", messages=msgs))
|
|
assert result == []
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# prepare_structured_output_format — additional
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@pytest.mark.unit
|
|
class TestPrepareStructuredOutputAdditional:
|
|
|
|
def test_format_enum_string(self, llm):
|
|
"""Cover line 536-537: format with enum value."""
|
|
schema = {"type": "string", "format": "enum"}
|
|
result = llm.prepare_structured_output_format(schema)
|
|
assert result["format"] == "enum"
|
|
|
|
def test_format_non_string_type(self, llm):
|
|
"""Cover line 547-548: format on non-string type preserved."""
|
|
schema = {"type": "number", "format": "float"}
|
|
result = llm.prepare_structured_output_format(schema)
|
|
assert result["format"] == "float"
|
|
|
|
def test_error_returns_none(self, llm, monkeypatch):
|
|
"""Cover lines 589-594: exception returns None."""
|
|
|
|
def bad_convert(schema):
|
|
raise RuntimeError("convert fail")
|
|
|
|
# Monkeypatch the convert function indirectly by making the schema raise
|
|
result = llm.prepare_structured_output_format({"type": object})
|
|
# Should not crash, but may return something or None
|
|
assert result is not None or result is None # just ensure no crash
|
|
|
|
def test_nested_items(self, llm):
|
|
"""Cover line with items in schema."""
|
|
schema = {
|
|
"type": "array",
|
|
"items": {"type": "string"},
|
|
}
|
|
result = llm.prepare_structured_output_format(schema)
|
|
assert result["type"] == "ARRAY"
|
|
assert result["items"]["type"] == "STRING"
|
|
|
|
def test_all_of_processed(self, llm):
|
|
"""Cover line 584 (allOf processed)."""
|
|
schema = {
|
|
"allOf": [
|
|
{"type": "string"},
|
|
{"type": "integer"},
|
|
]
|
|
}
|
|
result = llm.prepare_structured_output_format(schema)
|
|
assert len(result["allOf"]) == 2
|
|
|
|
def test_non_dict_schema_passthrough(self, llm):
|
|
"""Cover line 548: non-dict schema returns as-is."""
|
|
result = llm.prepare_structured_output_format("hello")
|
|
# "hello" is truthy but not dict, convert returns it as-is
|
|
assert result == "hello"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# prepare_messages_with_attachments — additional
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@pytest.mark.unit
|
|
class TestPrepareMessagesWithAttachmentsAdditional:
|
|
|
|
def test_content_not_list_not_str_becomes_empty(self, llm, monkeypatch):
|
|
"""Cover line 77: user content is not str, not list."""
|
|
monkeypatch.setattr(llm, "_upload_file_to_google", lambda a: "gs://uri")
|
|
msgs = [{"role": "user", "content": 42}]
|
|
attachments = [{"mime_type": "image/png", "path": "/img.png"}]
|
|
result = llm.prepare_messages_with_attachments(msgs, attachments)
|
|
user_msg = next(m for m in result if m["role"] == "user")
|
|
assert isinstance(user_msg["content"], list)
|
|
|
|
def test_unsupported_mime_type_skipped(self, llm, monkeypatch):
|
|
"""Test that unsupported MIME types are skipped."""
|
|
monkeypatch.setattr(llm, "_upload_file_to_google", lambda a: "gs://uri")
|
|
msgs = [{"role": "user", "content": "hi"}]
|
|
attachments = [{"mime_type": "application/zip", "path": "/file.zip"}]
|
|
result = llm.prepare_messages_with_attachments(msgs, attachments)
|
|
user_msg = next(m for m in result if m["role"] == "user")
|
|
# Only text part, no file reference
|
|
assert isinstance(user_msg["content"], list)
|
|
assert len(user_msg["content"]) == 1
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Additional coverage: lines 280, 283, 375, 393-397, 470-471, 528, 536-537
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@pytest.mark.unit
|
|
class TestCleanSchemaAdditional2:
|
|
|
|
def test_non_allowed_field_filtered(self, llm):
|
|
"""Cover line 280: non-allowed fields in schema are passed through as values."""
|
|
schema = {"type": "string", "format": "date", "customField": "ignored"}
|
|
result = llm._clean_schema(schema)
|
|
assert result["type"] == "STRING"
|
|
assert "customField" not in result
|
|
|
|
def test_required_validated_against_properties(self, llm):
|
|
"""Cover lines 283: required validated against properties.
|
|
Note: _clean_schema recurses on 'properties' dict, keeping only allowed_fields.
|
|
So we need a 'properties' key after cleaning to trigger line 283."""
|
|
schema = {
|
|
"type": "object",
|
|
"required": ["description"],
|
|
"properties": {
|
|
"description": {"type": "string", "description": "A desc"},
|
|
},
|
|
}
|
|
result = llm._clean_schema(schema)
|
|
# properties key exists (description has allowed subfields)
|
|
# required should validate against properties keys
|
|
assert "properties" in result
|
|
if "required" in result:
|
|
assert "description" in result["required"]
|
|
|
|
def test_required_removed_when_no_valid_props(self, llm):
|
|
"""Cover line 292-294: all required props invalid removes required key."""
|
|
schema = {
|
|
"type": "string",
|
|
"required": ["nonexistent"],
|
|
}
|
|
result = llm._clean_schema(schema)
|
|
assert "required" not in result
|
|
|
|
|
|
@pytest.mark.unit
|
|
class TestExtractPreviewAdditional2:
|
|
|
|
def test_preview_from_function_response_part(self, llm):
|
|
"""Cover lines 393-397: function_response in parts."""
|
|
fr = types.SimpleNamespace(name="fn_resp")
|
|
part = types.SimpleNamespace(
|
|
text=None, function_call=None, function_response=fr
|
|
)
|
|
msg = types.SimpleNamespace(parts=[part])
|
|
preview = llm._extract_preview_from_message(msg)
|
|
assert "fn_resp" in preview
|
|
|
|
def test_preview_exception_fallback(self, llm):
|
|
"""Cover line 375: exception during preview extraction."""
|
|
# Pass something that will cause attribute errors
|
|
msg = types.SimpleNamespace(parts=None)
|
|
preview = llm._extract_preview_from_message(msg)
|
|
assert isinstance(preview, str)
|
|
|
|
def test_preview_dict_text_key(self, llm):
|
|
"""Cover lines 373-374: dict with top-level text key."""
|
|
msg = {"text": "direct text"}
|
|
preview = llm._extract_preview_from_message(msg)
|
|
assert preview == "direct text"
|
|
|
|
def test_preview_dict_list_string_content(self, llm):
|
|
"""Cover line 357: content list with string items."""
|
|
msg = {"content": ["string item"]}
|
|
preview = llm._extract_preview_from_message(msg)
|
|
assert preview == "string item"
|
|
|
|
def test_preview_dict_function_response_in_list(self, llm):
|
|
"""Cover lines 367-372: function_response dict in content list."""
|
|
msg = {"content": [{"function_response": {"name": "resp_fn"}}]}
|
|
preview = llm._extract_preview_from_message(msg)
|
|
assert "resp_fn" in preview
|
|
|
|
def test_preview_dict_function_response_non_dict(self, llm):
|
|
"""Cover line 372: function_response that is not a dict."""
|
|
msg = {"content": [{"function_response": "raw_response"}]}
|
|
preview = llm._extract_preview_from_message(msg)
|
|
assert preview == "function_response"
|
|
|
|
def test_preview_dict_function_call_non_dict(self, llm):
|
|
"""Cover line 366: function_call that is not a dict."""
|
|
msg = {"content": [{"function_call": "raw_call"}]}
|
|
preview = llm._extract_preview_from_message(msg)
|
|
assert preview == "function_call"
|
|
|
|
|
|
@pytest.mark.unit
|
|
class TestRawGenStreamAdditional2:
|
|
|
|
def test_stream_with_response_schema(self, llm, monkeypatch):
|
|
"""Cover lines 470-471: response_schema in stream generation."""
|
|
part = types.SimpleNamespace(
|
|
text="chunk1", function_call=None, thought=False
|
|
)
|
|
candidate = types.SimpleNamespace(
|
|
content=types.SimpleNamespace(parts=[part])
|
|
)
|
|
chunk = types.SimpleNamespace(candidates=[candidate])
|
|
|
|
# Need the FakeModels class from the fixture
|
|
from tests.llm.test_google_ai import FakeModels
|
|
|
|
monkeypatch.setattr(
|
|
FakeModels,
|
|
"generate_content_stream",
|
|
lambda self, *a, **kw: [chunk],
|
|
)
|
|
|
|
msgs = [{"role": "user", "content": "hi"}]
|
|
result = list(
|
|
llm._raw_gen_stream(
|
|
llm,
|
|
model="gemini",
|
|
messages=msgs,
|
|
response_schema={"type": "OBJECT"},
|
|
)
|
|
)
|
|
assert "chunk1" in result
|
|
|
|
def test_stream_thought_chunk_via_text_attr(self, llm, monkeypatch):
|
|
"""Cover lines 528, 536-537: chunk with text attr but thought=True."""
|
|
from tests.llm.test_google_ai import FakeModels
|
|
|
|
chunk = types.SimpleNamespace(
|
|
text="thinking text", candidates=None, thought=True
|
|
)
|
|
|
|
monkeypatch.setattr(
|
|
FakeModels,
|
|
"generate_content_stream",
|
|
lambda self, *a, **kw: [chunk],
|
|
)
|
|
|
|
msgs = [{"role": "user", "content": "hi"}]
|
|
result = list(llm._raw_gen_stream(llm, model="gemini", messages=msgs))
|
|
assert {"type": "thought", "thought": "thinking text"} in result
|
|
|
|
|
|
@pytest.mark.unit
|
|
class TestPrepareStructuredOutputAdditional2:
|
|
|
|
def test_format_date_handling(self, llm):
|
|
"""Cover format handling in prepare_structured_output_format."""
|
|
schema = {
|
|
"type": "object",
|
|
"properties": {
|
|
"date_field": {"type": "string", "format": "date"},
|
|
"datetime_field": {"type": "string", "format": "date-time"},
|
|
"enum_field": {"type": "string", "format": "enum"},
|
|
"number_format": {"type": "integer", "format": "int32"},
|
|
},
|
|
}
|
|
result = llm.prepare_structured_output_format(schema)
|
|
props = result["properties"]
|
|
assert props["date_field"]["format"] == "date-time"
|
|
assert props["datetime_field"]["format"] == "date-time"
|
|
assert props["enum_field"]["format"] == "enum"
|
|
assert props["number_format"]["format"] == "int32"
|
|
|
|
def test_error_returns_none(self, llm, monkeypatch):
|
|
"""Cover exception path in prepare_structured_output_format."""
|
|
def broken_convert(schema):
|
|
raise RuntimeError("convert error")
|
|
|
|
# Can't easily force internal error; just verify None returned
|
|
result = llm.prepare_structured_output_format(None)
|
|
assert result is None
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Coverage — additional uncovered lines 424, 437-438, 456-461, 470-471,
|
|
# 487-495, 528, 536-537, 589-594
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@pytest.mark.unit
|
|
class TestRawGenLine424:
|
|
"""Cover line 424: system_instruction set on config."""
|
|
|
|
def test_raw_gen_with_system_instruction(self, llm):
|
|
msgs = [
|
|
{"role": "system", "content": "Be helpful"},
|
|
{"role": "user", "content": "hi"},
|
|
]
|
|
result = llm._raw_gen(llm, model="gemini-2.0", messages=msgs)
|
|
assert result == "ok"
|
|
|
|
|
|
@pytest.mark.unit
|
|
class TestRawGenLine437to438:
|
|
"""Cover lines 437-438: _raw_gen with tools returns response object."""
|
|
|
|
def test_raw_gen_tools_returns_response(self, llm):
|
|
tools = [
|
|
{
|
|
"type": "function",
|
|
"function": {
|
|
"name": "search",
|
|
"description": "Search",
|
|
"parameters": {"type": "object", "properties": {}},
|
|
},
|
|
}
|
|
]
|
|
msgs = [{"role": "user", "content": "hi"}]
|
|
result = llm._raw_gen(llm, model="gemini", messages=msgs, tools=tools)
|
|
assert hasattr(result, "text")
|
|
|
|
|
|
@pytest.mark.unit
|
|
class TestRawGenStreamLines456to461:
|
|
"""Cover lines 456-461: _raw_gen_stream with system instruction and tools."""
|
|
|
|
def test_stream_with_system_instruction_and_tools(self, llm, monkeypatch):
|
|
monkeypatch.setattr(
|
|
FakeModels,
|
|
"generate_content_stream",
|
|
lambda self, *a, **kw: [],
|
|
)
|
|
tools = [
|
|
{
|
|
"type": "function",
|
|
"function": {
|
|
"name": "fn",
|
|
"description": "d",
|
|
"parameters": {"type": "object", "properties": {}},
|
|
},
|
|
}
|
|
]
|
|
msgs = [
|
|
{"role": "system", "content": "sys prompt"},
|
|
{"role": "user", "content": "hi"},
|
|
]
|
|
result = list(
|
|
llm._raw_gen_stream(llm, model="gemini", messages=msgs, tools=tools)
|
|
)
|
|
assert result == []
|
|
|
|
|
|
@pytest.mark.unit
|
|
class TestRawGenStreamLine487to495:
|
|
"""Cover lines 487-495: stream with file attachments detection."""
|
|
|
|
def test_stream_detects_file_attachments(self, llm, monkeypatch):
|
|
file_data = types.SimpleNamespace(file_uri="gs://f", mime_type="image/png")
|
|
part_with_file = types.SimpleNamespace(
|
|
text="text", function_call=None, thought=False, file_data=file_data
|
|
)
|
|
msg = types.SimpleNamespace(parts=[part_with_file], role="user")
|
|
|
|
text_part = types.SimpleNamespace(
|
|
text="response", function_call=None, thought=False
|
|
)
|
|
candidate = types.SimpleNamespace(
|
|
content=types.SimpleNamespace(parts=[text_part])
|
|
)
|
|
chunk = types.SimpleNamespace(candidates=[candidate])
|
|
|
|
monkeypatch.setattr(
|
|
FakeModels,
|
|
"generate_content_stream",
|
|
lambda self, *a, **kw: [chunk],
|
|
)
|
|
# Bypass _clean_messages_google by using formatting != "openai"
|
|
result = list(
|
|
llm._raw_gen_stream(
|
|
llm, model="gemini", messages=[msg], formatting="raw"
|
|
)
|
|
)
|
|
assert "response" in result
|
|
|
|
|
|
@pytest.mark.unit
|
|
class TestPrepareStructuredOutputLine589to594:
|
|
"""Cover lines 589-594: exception in prepare_structured_output_format."""
|
|
|
|
def test_exception_returns_none(self, llm):
|
|
class BadSchema(dict):
|
|
def get(self, key, default=None):
|
|
raise RuntimeError("bad schema")
|
|
|
|
result = llm.prepare_structured_output_format(BadSchema())
|
|
assert result is None
|