hkuds--openharness
tjb-tech
85 行
2.7 KiB
Python
85 行
2.7 KiB
Python
"""Tests for background task management."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from openharness.tasks.manager import BackgroundTaskManager
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_shell_task_and_read_output(tmp_path: Path, monkeypatch):
|
|
monkeypatch.setenv("OPENHARNESS_DATA_DIR", str(tmp_path / "data"))
|
|
manager = BackgroundTaskManager()
|
|
|
|
task = await manager.create_shell_task(
|
|
command="printf 'hello task'",
|
|
description="hello",
|
|
cwd=tmp_path,
|
|
)
|
|
|
|
await asyncio.wait_for(manager._waiters[task.id], timeout=5) # type: ignore[attr-defined]
|
|
updated = manager.get_task(task.id)
|
|
assert updated is not None
|
|
assert updated.status == "completed"
|
|
assert "hello task" in manager.read_task_output(task.id)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_agent_task_with_command_override_and_write(tmp_path: Path, monkeypatch):
|
|
monkeypatch.setenv("OPENHARNESS_DATA_DIR", str(tmp_path / "data"))
|
|
manager = BackgroundTaskManager()
|
|
|
|
task = await manager.create_agent_task(
|
|
prompt="first",
|
|
description="agent",
|
|
cwd=tmp_path,
|
|
command="while read line; do echo \"got:$line\"; break; done",
|
|
)
|
|
|
|
await asyncio.wait_for(manager._waiters[task.id], timeout=5) # type: ignore[attr-defined]
|
|
assert "got:first" in manager.read_task_output(task.id)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_write_to_stopped_agent_task_restarts_process(tmp_path: Path, monkeypatch):
|
|
monkeypatch.setenv("OPENHARNESS_DATA_DIR", str(tmp_path / "data"))
|
|
manager = BackgroundTaskManager()
|
|
|
|
task = await manager.create_agent_task(
|
|
prompt="ready",
|
|
description="agent",
|
|
cwd=tmp_path,
|
|
command="while read line; do echo \"got:$line\"; break; done",
|
|
)
|
|
await asyncio.wait_for(manager._waiters[task.id], timeout=5) # type: ignore[attr-defined]
|
|
|
|
await manager.write_to_task(task.id, "follow-up")
|
|
await asyncio.wait_for(manager._waiters[task.id], timeout=5) # type: ignore[attr-defined]
|
|
|
|
output = manager.read_task_output(task.id)
|
|
assert "got:ready" in output
|
|
assert "got:follow-up" in output
|
|
updated = manager.get_task(task.id)
|
|
assert updated is not None
|
|
assert updated.metadata["restart_count"] == "1"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_stop_task(tmp_path: Path, monkeypatch):
|
|
monkeypatch.setenv("OPENHARNESS_DATA_DIR", str(tmp_path / "data"))
|
|
manager = BackgroundTaskManager()
|
|
|
|
task = await manager.create_shell_task(
|
|
command="sleep 30",
|
|
description="sleeper",
|
|
cwd=tmp_path,
|
|
)
|
|
await manager.stop_task(task.id)
|
|
updated = manager.get_task(task.id)
|
|
assert updated is not None
|
|
assert updated.status == "killed"
|