hkuds--openharness
tjb-tech
5dd8b952ec
A lightweight open-source Python implementation of the Agent Harness architecture. 44x lighter than Claude Code (11K vs 512K lines), 98% core tool coverage. - 43 tools with Pydantic validation and parallel execution - Skills system compatible with anthropics/skills (17+ tested) - Plugin system compatible with claude-code/plugins (12+ tested) - API retry with exponential backoff - Multi-level permissions with path rules - React/Ink TUI with "Oh my Harness!" branding - 114 unit tests + 6 E2E test suites - MIT License
111 行
3.6 KiB
Python
111 行
3.6 KiB
Python
"""Tests for task and team tools."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from openharness.tasks import get_task_manager
|
|
from openharness.tools.agent_tool import AgentTool, AgentToolInput
|
|
from openharness.tools.base import ToolExecutionContext
|
|
from openharness.tools.task_create_tool import TaskCreateTool, TaskCreateToolInput
|
|
from openharness.tools.task_output_tool import TaskOutputTool, TaskOutputToolInput
|
|
from openharness.tools.task_update_tool import TaskUpdateTool, TaskUpdateToolInput
|
|
from openharness.tools.team_create_tool import TeamCreateTool, TeamCreateToolInput
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_task_create_and_output_tool(tmp_path: Path, monkeypatch):
|
|
monkeypatch.setenv("OPENHARNESS_DATA_DIR", str(tmp_path / "data"))
|
|
context = ToolExecutionContext(cwd=tmp_path)
|
|
|
|
create_result = await TaskCreateTool().execute(
|
|
TaskCreateToolInput(
|
|
type="local_bash",
|
|
description="echo",
|
|
command="printf 'tool task'",
|
|
),
|
|
context,
|
|
)
|
|
assert create_result.is_error is False
|
|
task_id = create_result.output.split()[2]
|
|
|
|
manager = get_task_manager()
|
|
for _ in range(20):
|
|
if "tool task" in manager.read_task_output(task_id):
|
|
break
|
|
await asyncio.sleep(0.1)
|
|
output_result = await TaskOutputTool().execute(
|
|
TaskOutputToolInput(task_id=task_id),
|
|
context,
|
|
)
|
|
assert "tool task" in output_result.output
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_team_create_tool(tmp_path: Path):
|
|
result = await TeamCreateTool().execute(
|
|
TeamCreateToolInput(name="demo", description="test"),
|
|
ToolExecutionContext(cwd=tmp_path),
|
|
)
|
|
assert result.is_error is False
|
|
assert "Created team demo" == result.output
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_task_update_tool_updates_metadata(tmp_path: Path, monkeypatch):
|
|
monkeypatch.setenv("OPENHARNESS_DATA_DIR", str(tmp_path / "data"))
|
|
context = ToolExecutionContext(cwd=tmp_path)
|
|
|
|
create_result = await TaskCreateTool().execute(
|
|
TaskCreateToolInput(
|
|
type="local_bash",
|
|
description="updatable",
|
|
command="printf 'tool task'",
|
|
),
|
|
context,
|
|
)
|
|
task_id = create_result.output.split()[2]
|
|
|
|
update_result = await TaskUpdateTool().execute(
|
|
TaskUpdateToolInput(
|
|
task_id=task_id,
|
|
progress=60,
|
|
status_note="waiting on verification",
|
|
description="renamed task",
|
|
),
|
|
context,
|
|
)
|
|
assert update_result.is_error is False
|
|
|
|
task = get_task_manager().get_task(task_id)
|
|
assert task is not None
|
|
assert task.description == "renamed task"
|
|
assert task.metadata["progress"] == "60"
|
|
assert task.metadata["status_note"] == "waiting on verification"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_agent_tool_supports_remote_and_teammate_modes(tmp_path: Path, monkeypatch):
|
|
monkeypatch.setenv("OPENHARNESS_DATA_DIR", str(tmp_path / "data"))
|
|
context = ToolExecutionContext(cwd=tmp_path)
|
|
|
|
for mode in ("remote_agent", "in_process_teammate"):
|
|
result = await AgentTool().execute(
|
|
AgentToolInput(
|
|
description=f"{mode} smoke",
|
|
prompt="ready",
|
|
mode=mode,
|
|
command="python -u -c \"import sys; print(sys.stdin.readline().strip())\"",
|
|
),
|
|
context,
|
|
)
|
|
assert result.is_error is False
|
|
task_id = result.output.split()[-1]
|
|
task = get_task_manager().get_task(task_id)
|
|
assert task is not None
|
|
assert task.type == mode
|
|
assert task.metadata["agent_mode"] == mode
|