mirrors/hkuds--openharness
1
0
Fork 0
Code Issues Pull Requests Actions 2 Deployments Models Agent Notes Packages Projects Releases Wiki Activity

feat(vision): add image-to-text fallback for text-only models (#227)

Browse Source 
* feat(provider): add ModelScope inference API support

* fix(cli/auth): wire ModelScope into CLI and auth manager

* feat(vision): add image-to-text fallback for non-multimodal models

* fix: remove leftover merge conflict marker in query.py

* test: fix cross-platform test failures

* test: remove unused os import in test_environment.py
This commit is contained in:
Mcy0618
2026-05-03 16:47:56 +08:00
committed by GitHub
parent e536394812
commit d611d6ddf6
13 changed files with 742 additions and 11 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/openharness/api/provider.py
+61
View File
@@ -2,6 +2,7 @@
from __future__ import annotations
import re
from dataclasses import dataclass
from openharness.auth.external import describe_external_binding
@@ -123,3 +124,63 @@ def auth_status(settings: Settings) -> str:
if resolved.source.startswith("external:"):
return f"configured ({resolved.source.removeprefix('external:')})"
return "configured"
# ---------------------------------------------------------------------------
# Multimodal (vision) capability detection
# ---------------------------------------------------------------------------
# Known multimodal model patterns (lowercase, regex).
# These models can accept image input natively.
_MULTIMODAL_MODEL_PATTERNS: list[re.Pattern[str]] = [
# Anthropic Claude 3+ (all Claude 3 and later support images)
re.compile(r"^claude-3(?:\.\d+)?(?:-sonnet|-opus|-haiku)?"),
re.compile(r"^claude-(?:sonnet|opus|haiku)-\d"),
# OpenAI GPT-4o / o-series
re.compile(r"^gpt-4o"),
re.compile(r"^o[1349]-"),
# Google Gemini
re.compile(r"^gemini-(?:pro-)?vision"),
re.compile(r"^gemini-2\.\d+"),
# Qwen / DashScope VL series
re.compile(r"^qwen-vl"),
re.compile(r"^qwen2\.5?-vl"),
re.compile(r"^qvq-"),
# DeepSeek VL
re.compile(r"^deepseek-vl"),
re.compile(r"^deepseek-vision"),
# Open-source multimodal
re.compile(r"^llava"),
re.compile(r"^cogvlm"),
re.compile(r"^internvl"),
re.compile(r"^glm-4v"),
# Moonshot / Kimi (k2.5 supports images)
re.compile(r"^kimi-k2\.5"),
# StepFun (阶跃星辰) — Step-2 and Step-1v support images
re.compile(r"^step-2"),
re.compile(r"^step-1v"),
# MiniMax VL
re.compile(r"^minimax-vl"),
# Zhipu GLM-4V
re.compile(r"^glm-4v"),
# Mistral Pixtral
re.compile(r"^pixtral"),
# Groq vision models (llama-3.2-vision, etc.)
re.compile(r"vision"),
# Generic: model names containing "vl" or "vision" as a word boundary
re.compile(r"(?:^|[-\s/])vl(?:$|[-\s])"),
]
def is_model_multimodal(model: str) -> bool:
"""Return True when the model name indicates multimodal (vision) capability.
This is a heuristic based on known model naming conventions. It errs on
the side of returning False for unknown models so that the image-to-text
fallback tool is used rather than silently failing.
"""
normalized = model.strip().lower()
# Strip provider prefix like "anthropic/" or "openai/"
if "/" in normalized:
normalized = normalized.split("/", 1)[-1]
return any(pattern.search(normalized) is not None for pattern in _MULTIMODAL_MODEL_PATTERNS)
src/openharness/config/settings.py
+29
View File
@@ -456,6 +456,32 @@ def _profile_from_flat_settings(settings: "Settings") -> tuple[str, ProviderProf
return name, profile
class VisionModelConfig(BaseModel):
"""Configuration for the vision model used by the image_to_text tool.
When the active model does not support multimodal input, the agent loop
automatically falls back to this vision model to describe images.
"""
model: str = ""
api_key: str = ""
base_url: str = ""
@classmethod
def from_env(cls) -> "VisionModelConfig":
"""Load vision model config from environment variables."""
return cls(
model=os.environ.get("OPENHARNESS_VISION_MODEL", "").strip(),
api_key=os.environ.get("OPENHARNESS_VISION_API_KEY", "").strip(),
base_url=os.environ.get("OPENHARNESS_VISION_BASE_URL", "").strip(),
)
@property
def is_configured(self) -> bool:
"""Return True when both model and api_key are set."""
return bool(self.model and self.api_key)
class Settings(BaseModel):
"""Main settings model for OpenHarness."""
@@ -493,6 +519,9 @@ class Settings(BaseModel):
passes: int = 1
verbose: bool = False
# Vision model (image-to-text fallback)
vision: VisionModelConfig = Field(default_factory=VisionModelConfig)
def merged_profiles(self) -> dict[str, ProviderProfile]:
"""Return the saved profiles merged over the built-in catalog."""
merged = default_provider_profiles()
src/openharness/engine/query.py
+89 -1
View File
@@ -18,9 +18,15 @@ from openharness.api.client import (
ApiTextDeltaEvent,
SupportsStreamingMessages,
)
from openharness.api.provider import is_model_multimodal
from openharness.api.usage import UsageSnapshot
from openharness.config.paths import get_data_dir
from openharness.engine.messages import ConversationMessage, ToolResultBlock
from openharness.engine.messages import (
ConversationMessage,
ImageBlock,
TextBlock,
ToolResultBlock,
)
from openharness.engine.stream_events import (
AssistantTextDelta,
AssistantTurnComplete,
@@ -501,6 +507,83 @@ def _offload_tool_output_if_needed(
return inline, artifact_path
# ---------------------------------------------------------------------------
# Image preprocessing — convert ImageBlocks to text for non-multimodal models
# ---------------------------------------------------------------------------
_IMAGE_PREPROCESS_STATUS = "Converting image to text description via vision model…"
async def _preprocess_images_in_messages(
messages: list[ConversationMessage],
context: QueryContext,
) -> AsyncIterator[StreamEvent]:
"""Scan messages for ImageBlocks and convert them to text if the active
model does not support multimodal input.
Yields status events during conversion so the UI stays responsive.
"""
if is_model_multimodal(context.model):
return
vision_config = context.tool_metadata.get("vision_model_config")
if not vision_config:
# No vision model configured — skip preprocessing.
return
# Collect all ImageBlocks with their parent message index and block index
pending: list[tuple[int, int, ImageBlock]] = []
for msg_idx, msg in enumerate(messages):
if msg.role != "user":
continue
for blk_idx, block in enumerate(msg.content):
if isinstance(block, ImageBlock):
pending.append((msg_idx, blk_idx, block))
if not pending:
return
yield StatusEvent(message=_IMAGE_PREPROCESS_STATUS)
# Process images in parallel
async def _describe(msg_idx: int, blk_idx: int, block: ImageBlock) -> tuple[int, int, str]:
tool = context.tool_registry.get("image_to_text")
if tool is None:
return msg_idx, blk_idx, "[Image: could not describe — image_to_text tool not available]"
# Build tool input
tool_input_data: dict[str, object] = {
"image_data": block.data,
"media_type": block.media_type,
"prompt": "Describe this image in detail, including any text, "
"UI elements, code, diagrams, or visual information present.",
}
try:
parsed = tool.input_model.model_validate(tool_input_data)
except Exception:
return msg_idx, blk_idx, "[Image: could not parse image data]"
exec_context = ToolExecutionContext(
cwd=context.cwd,
metadata={
"vision_model_config": vision_config,
**(context.tool_metadata or {}),
},
)
result = await tool.execute(parsed, exec_context)
if result.is_error:
return msg_idx, blk_idx, f"[Image description failed: {result.output}]"
return msg_idx, blk_idx, result.output
results = await asyncio.gather(*[_describe(mi, bi, blk) for mi, bi, blk in pending])
# Replace ImageBlocks with TextBlocks in-place
for msg_idx, blk_idx, description in results:
msg = messages[msg_idx]
msg.content[blk_idx] = TextBlock(text=description)
async def run_query(
context: QueryContext,
messages: list[ConversationMessage],
@@ -571,6 +654,11 @@ async def run_query(
messages, was_compacted = last_compaction_result
# ---------------------------------------------------------------
# --- image preprocessing: convert ImageBlocks to text for non-vision models ---
async for event in _preprocess_images_in_messages(messages, context):
yield event, None
# -----------------------------------------------------------------------------
final_message: ConversationMessage | None = None
usage = UsageSnapshot()
src/openharness/tools/__init__.py
+2
View File
@@ -19,6 +19,7 @@ from openharness.tools.file_read_tool import FileReadTool
from openharness.tools.file_write_tool import FileWriteTool
from openharness.tools.glob_tool import GlobTool
from openharness.tools.grep_tool import GrepTool
from openharness.tools.image_to_text_tool import ImageToTextTool
from openharness.tools.list_mcp_resources_tool import ListMcpResourcesTool
from openharness.tools.lsp_tool import LspTool
from openharness.tools.mcp_auth_tool import McpAuthTool
@@ -57,6 +58,7 @@ def create_default_tool_registry(mcp_manager=None) -> ToolRegistry:
McpAuthTool(),
GlobTool(),
GrepTool(),
ImageToTextTool(),
SkillTool(),
ToolSearchTool(),
WebFetchTool(),
src/openharness/tools/image_to_text_tool.py
+237
View File
@@ -0,0 +1,237 @@
"""Convert images to text descriptions using a multimodal model.
This tool acts as a bridge for pure-text models: when the user attaches an
image but the active model cannot process images natively, the agent loop
(or the model itself) can invoke this tool to obtain a text/JSON description
of the image via a separately configured vision-capable model.
"""
from __future__ import annotations
import base64
import logging
from pathlib import Path
from typing import Any
from pydantic import BaseModel, Field
from openharness.api.openai_client import OpenAICompatibleClient
from openharness.tools.base import BaseTool, ToolExecutionContext, ToolResult
log = logging.getLogger(__name__)
# Default system prompt for image description.
_DEFAULT_VISION_PROMPT = (
"You are an image description assistant. "
"Describe the image in detail, including any text, objects, people, "
"colors, layout, and context. If the image contains code, UI screenshots, "
"diagrams, or data visualizations, describe them precisely so that a "
"text-only AI model can understand the content."
)
class ImageToTextToolInput(BaseModel):
"""Arguments for converting an image to text."""
image_data: str | None = Field(
default=None,
description="Base64-encoded image data. Provide either image_data or image_path.",
)
image_path: str | None = Field(
default=None,
description="Local file path to the image. Provide either image_data or image_path.",
)
prompt: str = Field(
default=_DEFAULT_VISION_PROMPT,
description="Custom instruction for describing the image. "
"Defaults to a general-purpose description prompt.",
)
media_type: str = Field(
default="image/png",
description="MIME type of the image (e.g. image/png, image/jpeg, image/webp). "
"Only used when image_data is provided.",
)
max_tokens: int = Field(
default=2048,
ge=256,
le=16384,
description="Maximum tokens for the vision model response.",
)
class ImageToTextTool(BaseTool):
"""Use a multimodal model to describe an image and return text."""
name = "image_to_text"
description = (
"Convert an image to a detailed text description using a vision-capable model. "
"Use this when you need to understand the content of an image but your current "
"model does not support image input."
)
input_model = ImageToTextToolInput
async def execute(
self, arguments: ImageToTextToolInput, context: ToolExecutionContext
) -> ToolResult:
# 1. Resolve image data
image_data, media_type = await self._resolve_image(arguments, context)
if image_data is None:
return ToolResult(
output="image_to_text failed: provide either image_data (base64) or image_path",
is_error=True,
)
# 2. Get vision model config from context metadata
vision_config = context.metadata.get("vision_model_config", {})
if not isinstance(vision_config, dict):
vision_config = {}
model = vision_config.get("model", "")
api_key = vision_config.get("api_key", "")
base_url = vision_config.get("base_url", "")
if not model or not api_key:
log.warning(
"image_to_text: vision model not configured. "
"Set vision.model and vision.api_key in settings."
)
return ToolResult(
output=(
"image_to_text failed: vision model is not configured. "
"Please set vision.model and vision.api_key in your settings, "
"or configure the OPENHARNESS_VISION_MODEL and "
"OPENHARNESS_VISION_API_KEY environment variables."
),
is_error=True,
)
# 3. Call the vision model
try:
description = await self._call_vision_model(
image_data=image_data,
media_type=media_type or arguments.media_type,
prompt=arguments.prompt,
model=model,
api_key=api_key,
base_url=base_url,
max_tokens=arguments.max_tokens,
)
except Exception as exc:
log.exception("image_to_text: vision model call failed")
return ToolResult(
output=f"image_to_text failed: vision model error: {exc}",
is_error=True,
)
return ToolResult(
output=(
f"[Image description via {model}]\n\n{description}"
)
)
def is_read_only(self, arguments: BaseModel) -> bool:
del arguments
return True
# ------------------------------------------------------------------
# Internal helpers
# ------------------------------------------------------------------
@staticmethod
async def _resolve_image(
arguments: ImageToTextToolInput,
context: ToolExecutionContext,
) -> tuple[str | None, str | None]:
"""Resolve image data from either base64 string or file path."""
if arguments.image_data:
return arguments.image_data, arguments.media_type
if arguments.image_path:
path = Path(arguments.image_path)
if not path.is_absolute():
path = context.cwd / path
path = path.expanduser().resolve()
if not path.exists():
log.warning("image_to_text: image not found at %s", path)
return None, None
try:
raw = path.read_bytes()
data = base64.b64encode(raw).decode("ascii")
except OSError as exc:
log.warning("image_to_text: failed to read %s: %s", path, exc)
return None, None
# Guess media type from extension
ext = path.suffix.lower()
media_type = {
".png": "image/png",
".jpg": "image/jpeg",
".jpeg": "image/jpeg",
".gif": "image/gif",
".webp": "image/webp",
".bmp": "image/bmp",
".svg": "image/svg+xml",
}.get(ext, "image/png")
return data, media_type
return None, None
@staticmethod
async def _call_vision_model(
*,
image_data: str,
media_type: str,
prompt: str,
model: str,
api_key: str,
base_url: str,
max_tokens: int,
) -> str:
"""Call the vision model via OpenAI-compatible API."""
client = OpenAICompatibleClient(
api_key=api_key,
base_url=base_url or None,
)
from openharness.api.client import ApiMessageRequest
from openharness.engine.messages import (
ConversationMessage,
ImageBlock,
TextBlock,
)
# Build a user message with the image
user_content: list[Any] = [TextBlock(text=prompt)]
user_content.append(
ImageBlock(
media_type=media_type,
data=image_data,
)
)
user_message = ConversationMessage(role="user", content=user_content)
# Stream the response and collect text
collected_text = ""
async for event in client.stream_message(
ApiMessageRequest(
model=model,
messages=[user_message],
system_prompt="",
max_tokens=max_tokens,
tools=[],
)
):
from openharness.api.client import ApiTextDeltaEvent, ApiMessageCompleteEvent
if isinstance(event, ApiTextDeltaEvent):
collected_text += event.text
elif isinstance(event, ApiMessageCompleteEvent):
# Also grab any text from the final message
text = event.message.text
if text and text not in collected_text:
collected_text = text
return collected_text.strip() or "(no description returned)"
src/openharness/ui/runtime.py
+28
View File
@@ -44,6 +44,33 @@ StreamRenderer = Callable[[StreamEvent], Awaitable[None]]
ClearHandler = Callable[[], Awaitable[None]]
def _resolve_vision_config(settings) -> dict[str, str]:
"""Resolve the vision model configuration from settings or environment.
Priority: settings.vision fields > environment variables > empty.
"""
from openharness.config.settings import VisionModelConfig
cfg = settings.vision
if cfg.is_configured:
return {
"model": cfg.model,
"api_key": cfg.api_key,
"base_url": cfg.base_url,
}
# Fall back to environment variables
env_cfg = VisionModelConfig.from_env()
if env_cfg.is_configured:
return {
"model": env_cfg.model,
"api_key": env_cfg.api_key,
"base_url": env_cfg.base_url,
}
return {}
@dataclass
class RuntimeBundle:
"""Shared runtime objects for one interactive session."""
@@ -314,6 +341,7 @@ async def build_runtime(
"extra_skill_dirs": normalized_skill_dirs,
"extra_plugin_roots": normalized_plugin_roots,
"session_id": session_id,
"vision_model_config": _resolve_vision_config(settings),
**restored_metadata,
},
)
tests/test_auth/test_external.py
+3
View File
@@ -256,6 +256,9 @@ def test_cli_codex_login_binds_without_switching(monkeypatch, tmp_path: Path):
)
monkeypatch.setenv("OPENHARNESS_CONFIG_DIR", str(config_dir))
monkeypatch.setenv("CODEX_HOME", str(codex_home))
# Prevent env var leakage from overriding the configured api_key
monkeypatch.delenv("ANTHROPIC_API_KEY", raising=False)
monkeypatch.delenv("OPENAI_API_KEY", raising=False)
(config_dir / "settings.json").write_text(
json.dumps(
tests/test_commands/test_cli.py
+3
View File
@@ -95,6 +95,9 @@ def test_select_from_menu_uses_questionary_when_tty(monkeypatch):
def test_setup_flow_creates_kimi_profile_with_profile_scoped_key(tmp_path: Path, monkeypatch):
runner = CliRunner()
monkeypatch.setenv("OPENHARNESS_CONFIG_DIR", str(tmp_path))
# Prevent env var leakage from overriding the configured api_key
monkeypatch.delenv("ANTHROPIC_API_KEY", raising=False)
monkeypatch.delenv("OPENAI_API_KEY", raising=False)
selections = iter(["claude-api", "kimi-anthropic"])
prompts = iter(
tests/test_commands/test_registry.py
+4 -1
View File
@@ -732,6 +732,9 @@ async def test_version_context_and_share_commands(tmp_path: Path, monkeypatch):
async def test_auth_feedback_and_project_context_commands(tmp_path: Path, monkeypatch):
monkeypatch.setenv("OPENHARNESS_CONFIG_DIR", str(tmp_path / "config"))
monkeypatch.setenv("OPENHARNESS_DATA_DIR", str(tmp_path / "data"))
# Prevent env var leakage from overriding the configured api_key
monkeypatch.delenv("ANTHROPIC_API_KEY", raising=False)
monkeypatch.delenv("OPENAI_API_KEY", raising=False)
registry = create_default_command_registry()
context = _make_context(tmp_path)
@@ -790,7 +793,7 @@ async def test_agents_session_files_and_reload_plugins_commands(tmp_path: Path,
files_command, files_args = registry.lookup("/files app.py")
files_result = await files_command.handler(files_args, context)
assert "src/app.py" in files_result.message
assert "src/app.py" in files_result.message.replace("\\", "/")
files_dirs_command, files_dirs_args = registry.lookup("/files dirs")
files_dirs_result = await files_dirs_command.handler(files_dirs_args, context)
tests/test_prompts/test_environment.py
+2 -3
View File
@@ -2,7 +2,6 @@
from __future__ import annotations
import os
import subprocess
from pathlib import Path
@@ -40,8 +39,8 @@ def test_detect_shell_fallback(monkeypatch):
def test_detect_git_info_in_repo(tmp_path: Path):
# Create a git repo
os.system(f"git init {tmp_path} > /dev/null 2>&1")
# Create a git repo (cross-platform: use subprocess, not os.system with /dev/null)
subprocess.run(["git", "init", str(tmp_path)], capture_output=True)
is_git, branch = detect_git_info(str(tmp_path))
assert is_git is True
# branch may be None for empty repo or "main"/"master"
tests/test_tools/test_core_tools.py
+3 -3
View File
@@ -97,7 +97,7 @@ async def test_glob_tool_accepts_absolute_patterns(tmp_path: Path, monkeypatch):
)
assert result.is_error is False
assert result.output.splitlines() == ["pkg/a.py"]
assert result.output.replace("\\", "/").splitlines() == ["pkg/a.py"]
@pytest.mark.asyncio
@@ -215,13 +215,13 @@ async def test_lsp_tool(tmp_path: Path):
LspToolInput(operation="go_to_definition", file_path="pkg/app.py", symbol="greet"),
context,
)
assert "pkg/utils.py:1:1" in definition.output
assert "pkg/utils.py:1:1" in definition.output.replace("\\", "/")
references = await LspTool().execute(
LspToolInput(operation="find_references", file_path="pkg/app.py", symbol="greet"),
context,
)
assert "pkg/app.py:1:from pkg.utils import greet" in references.output
assert "pkg/app.py:1:from pkg.utils import greet" in references.output.replace("\\", "/")
hover = await LspTool().execute(
LspToolInput(operation="hover", file_path="pkg/app.py", symbol="greet"),
tests/test_tools/test_image_to_text_tool.py
+278
View File
@@ -0,0 +1,278 @@
"""Tests for image_to_text tool and multimodal detection."""
from __future__ import annotations
from pathlib import Path
import pytest
from openharness.api.provider import is_model_multimodal
from openharness.config.settings import VisionModelConfig
from openharness.tools.base import ToolExecutionContext
from openharness.tools.image_to_text_tool import ImageToTextTool, ImageToTextToolInput
# ---------------------------------------------------------------------------
# is_model_multimodal tests
# ---------------------------------------------------------------------------
@pytest.mark.parametrize(
("model", "expected"),
[
# Anthropic Claude 3+ (multimodal)
("claude-sonnet-4-6", True),
("claude-opus-4-6", True),
("claude-haiku-4-5", True),
("claude-3-5-sonnet-20241022", True),
("claude-3-opus-20240229", True),
("claude-3-haiku-20240307", True),
# OpenAI multimodal
("gpt-4o", True),
("gpt-4o-mini", True),
("o1-mini", True),
("o3-mini", True),
("o4-mini", True),
# Google Gemini
("gemini-2.5-flash", True),
("gemini-2.0-flash", True),
("gemini-pro-vision", True),
# Qwen VL
("qwen-vl-max", True),
("qwen2.5-vl-72b", True),
("qvq-72b-preview", True),
# DeepSeek VL
("deepseek-vl2", True),
# Other multimodal
("llava-v1.6-34b", True),
("pixtral-12b", True),
("step-2-16k", True),
("step-1v-32k", True),
("kimi-k2.5", True),
# Non-multimodal models
("claude-2.1", False),
("gpt-4", False),
("gpt-3.5-turbo", False),
("deepseek-chat", False),
("deepseek-reasoner", False),
("qwen-turbo", False),
("qwen-plus", False),
("kimi-k2", False),
("step-1-8k", False),
("glm-4", False),
("gemini-1.0-pro", False),
("unknown-model-123", False),
("", False),
# With provider prefix
("anthropic/claude-sonnet-4-6", True),
("openai/gpt-4o", True),
("openai/gpt-4", False),
],
)
def test_is_model_multimodal(model: str, expected: bool) -> None:
assert is_model_multimodal(model) == expected
# ---------------------------------------------------------------------------
# ImageToTextTool input validation tests
# ---------------------------------------------------------------------------
class TestImageToTextToolInput:
"""Validate the tool's input model."""
def test_valid_image_data(self) -> None:
inp = ImageToTextToolInput(
image_data="iVBORw0KGgo=",
media_type="image/png",
)
assert inp.image_data == "iVBORw0KGgo="
assert inp.media_type == "image/png"
assert inp.prompt # default prompt
def test_valid_image_path(self) -> None:
inp = ImageToTextToolInput(
image_path="/tmp/test.png",
)
assert inp.image_path == "/tmp/test.png"
assert inp.image_data is None
def test_default_prompt(self) -> None:
inp = ImageToTextToolInput(image_data="data")
assert "image" in inp.prompt.lower()
def test_custom_prompt(self) -> None:
inp = ImageToTextToolInput(
image_data="data",
prompt="Extract all text from this image",
)
assert inp.prompt == "Extract all text from this image"
def test_max_tokens_range(self) -> None:
# Default
inp = ImageToTextToolInput(image_data="data")
assert inp.max_tokens == 2048
# Custom valid
inp = ImageToTextToolInput(image_data="data", max_tokens=4096)
assert inp.max_tokens == 4096
def test_neither_image_data_nor_path(self) -> None:
"""Both fields are optional in the model, but the tool will error."""
inp = ImageToTextToolInput()
assert inp.image_data is None
assert inp.image_path is None
# ---------------------------------------------------------------------------
# ImageToTextTool execution tests (no real API calls)
# ---------------------------------------------------------------------------
@pytest.mark.asyncio
async def test_execute_no_input(tmp_path: Path) -> None:
"""Tool returns error when neither image_data nor image_path is provided."""
tool = ImageToTextTool()
context = ToolExecutionContext(cwd=tmp_path)
result = await tool.execute(
ImageToTextToolInput(),
context,
)
assert result.is_error
assert "provide either" in result.output
@pytest.mark.asyncio
async def test_execute_nonexistent_path(tmp_path: Path) -> None:
"""Tool returns error when image_path does not exist."""
tool = ImageToTextTool()
context = ToolExecutionContext(cwd=tmp_path)
result = await tool.execute(
ImageToTextToolInput(image_path="/nonexistent/path/image.png"),
context,
)
assert result.is_error
assert "provide either" in result.output
@pytest.mark.asyncio
async def test_execute_no_vision_config(tmp_path: Path) -> None:
"""Tool returns error when vision model is not configured."""
tool = ImageToTextTool()
context = ToolExecutionContext(
cwd=tmp_path,
metadata={"vision_model_config": {}},
)
result = await tool.execute(
ImageToTextToolInput(image_data="iVBORw0KGgo="),
context,
)
assert result.is_error
assert "vision model is not configured" in result.output
@pytest.mark.asyncio
async def test_execute_with_image_path(tmp_path: Path) -> None:
"""Tool reads a real image file and attempts to describe it."""
# Create a minimal valid PNG file
png_path = tmp_path / "test_image.png"
# Minimal valid PNG (1x1 pixel, white)
minimal_png = bytes([
0x89, 0x50, 0x4E, 0x47, 0x0D, 0x0A, 0x1A, 0x0A, # PNG signature
0x00, 0x00, 0x00, 0x0D, 0x49, 0x48, 0x44, 0x52, # IHDR chunk
0x00, 0x00, 0x00, 0x01, 0x00, 0x00, 0x00, 0x01,
0x08, 0x02, 0x00, 0x00, 0x00, 0x90, 0x77, 0x53,
0xDE, 0x00, 0x00, 0x00, 0x0C, 0x49, 0x44, 0x41, # IDAT chunk
0x54, 0x08, 0xD7, 0x63, 0x60, 0x60, 0x00, 0x00,
0x00, 0x04, 0x00, 0x01, 0x27, 0x34, 0x27, 0x0A,
0x00, 0x00, 0x00, 0x00, 0x49, 0x45, 0x4E, 0x44, # IEND chunk
0xAE, 0x42, 0x60, 0x82,
])
png_path.write_bytes(minimal_png)
tool = ImageToTextTool()
context = ToolExecutionContext(
cwd=tmp_path,
metadata={
"vision_model_config": {
"model": "gpt-4o",
"api_key": "test-key",
"base_url": "",
}
},
)
result = await tool.execute(
ImageToTextToolInput(image_path=str(png_path)),
context,
)
# Should fail at API call (not at file reading), since the API key is fake
assert result.is_error
assert "vision model error" in result.output
@pytest.mark.asyncio
async def test_is_read_only() -> None:
"""image_to_text is a read-only tool."""
tool = ImageToTextTool()
assert tool.is_read_only(ImageToTextToolInput(image_data="data"))
# ---------------------------------------------------------------------------
# VisionModelConfig tests
# ---------------------------------------------------------------------------
class TestVisionModelConfig:
"""Validate the VisionModelConfig model."""
def test_default_empty(self) -> None:
cfg = VisionModelConfig()
assert cfg.model == ""
assert cfg.api_key == ""
assert cfg.base_url == ""
assert not cfg.is_configured
def test_configured(self) -> None:
cfg = VisionModelConfig(
model="gpt-4o",
api_key="sk-test",
base_url="https://api.openai.com/v1",
)
assert cfg.is_configured
assert cfg.model == "gpt-4o"
assert cfg.api_key == "sk-test"
def test_partial_not_configured(self) -> None:
cfg = VisionModelConfig(model="gpt-4o")
assert not cfg.is_configured
def test_from_env(self, monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.setenv("OPENHARNESS_VISION_MODEL", "gpt-4o")
monkeypatch.setenv("OPENHARNESS_VISION_API_KEY", "sk-env-key")
monkeypatch.setenv("OPENHARNESS_VISION_BASE_URL", "https://api.example.com/v1")
cfg = VisionModelConfig.from_env()
assert cfg.model == "gpt-4o"
assert cfg.api_key == "sk-env-key"
assert cfg.base_url == "https://api.example.com/v1"
assert cfg.is_configured
def test_from_env_partial(self, monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.setenv("OPENHARNESS_VISION_MODEL", "gpt-4o")
monkeypatch.delenv("OPENHARNESS_VISION_API_KEY", raising=False)
cfg = VisionModelConfig.from_env()
assert not cfg.is_configured
# ---------------------------------------------------------------------------
# Tool registry integration test
# ---------------------------------------------------------------------------
def test_tool_registered() -> None:
"""image_to_text tool is registered in the default registry."""
from openharness.tools import create_default_tool_registry
registry = create_default_tool_registry()
tool = registry.get("image_to_text")
assert tool is not None
assert tool.name == "image_to_text"
assert "vision" in tool.description.lower()
assert tool.input_model.__name__ == "ImageToTextToolInput"
assert tool.input_model.__module__ == "openharness.tools.image_to_text_tool"
tests/test_tools/test_integration_flows.py
+3 -3
View File
@@ -40,13 +40,13 @@ async def test_search_edit_flow_across_registry(tmp_path: Path):
context,
)
glob_result = await glob.execute(glob.input_model(pattern="**/*.py"), context)
assert "src/demo.py" in glob_result.output
assert "src/demo.py" in glob_result.output.replace("\\", "/")
grep_result = await grep.execute(
grep.input_model(pattern="beta", file_glob="**/*.py"),
context,
)
assert "src/demo.py:2:beta" in grep_result.output
assert "src/demo.py:2:beta" in grep_result.output.replace("\\", "/")
await edit.execute(
edit.input_model(path="src/demo.py", old_str="beta", new_str="gamma"),
@@ -290,7 +290,7 @@ async def test_lsp_flow_across_registry(tmp_path: Path):
lsp.input_model(operation="go_to_definition", file_path="pkg/app.py", symbol="greet"),
context,
)
assert "pkg/utils.py:1:1" in definition_result.output
assert "pkg/utils.py:1:1" in definition_result.output.replace("\\", "/")
hover_result = await lsp.execute(
lsp.input_model(operation="hover", file_path="pkg/app.py", symbol="greet"),