mirror of
https://github.com/coleam00/Archon.git
synced 2025-12-31 06:08:03 -05:00
Revert "chore: remove example workflow directory"
This reverts commit c2a568e08c.
This commit is contained in:
Rasmus Widing
@@ -28,7 +28,7 @@ class SandboxType(str, Enum):
|
||||
"""Sandbox environment types"""
|
||||
|
||||
GIT_BRANCH = "git_branch"
|
||||
GIT_WORKTREE = "git_worktree" # Fully implemented - recommended for concurrent execution
|
||||
GIT_WORKTREE = "git_worktree" # Placeholder for Phase 2+
|
||||
E2B = "e2b" # Placeholder for Phase 2+
|
||||
DAGGER = "dagger" # Placeholder for Phase 2+
|
||||
|
||||
@@ -102,10 +102,7 @@ class CreateAgentWorkOrderRequest(BaseModel):
|
||||
"""
|
||||
|
||||
repository_url: str = Field(..., description="Git repository URL")
|
||||
sandbox_type: SandboxType = Field(
|
||||
default=SandboxType.GIT_WORKTREE,
|
||||
description="Sandbox environment type (defaults to git_worktree for efficient concurrent execution)"
|
||||
)
|
||||
sandbox_type: SandboxType = Field(..., description="Sandbox environment type")
|
||||
user_request: str = Field(..., description="User's description of the work to be done")
|
||||
selected_commands: list[str] = Field(
|
||||
default=["create-branch", "planning", "execute", "commit", "create-pr"],
|
||||
|
||||
@@ -164,7 +164,7 @@ class WorkflowOrchestrator:
|
||||
branch_name = context.get("create-branch")
|
||||
git_stats = await self._calculate_git_stats(
|
||||
branch_name,
|
||||
sandbox.working_dir
|
||||
sandbox.get_working_directory()
|
||||
)
|
||||
|
||||
await self.state_repository.update_status(
|
||||
@@ -188,7 +188,7 @@ class WorkflowOrchestrator:
|
||||
branch_name = context.get("create-branch")
|
||||
if branch_name:
|
||||
git_stats = await self._calculate_git_stats(
|
||||
branch_name, sandbox.working_dir
|
||||
branch_name, sandbox.get_working_directory()
|
||||
)
|
||||
await self.state_repository.update_status(
|
||||
agent_work_order_id,
|
||||
|
||||
@@ -1,178 +0,0 @@
|
||||
"""Tests for Port Allocation"""
|
||||
|
||||
import pytest
|
||||
from unittest.mock import patch
|
||||
|
||||
from src.agent_work_orders.utils.port_allocation import (
|
||||
get_ports_for_work_order,
|
||||
is_port_available,
|
||||
find_next_available_ports,
|
||||
create_ports_env_file,
|
||||
)
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_get_ports_for_work_order_deterministic():
|
||||
"""Test that same work order ID always gets same ports"""
|
||||
work_order_id = "wo-abc123"
|
||||
|
||||
backend1, frontend1 = get_ports_for_work_order(work_order_id)
|
||||
backend2, frontend2 = get_ports_for_work_order(work_order_id)
|
||||
|
||||
assert backend1 == backend2
|
||||
assert frontend1 == frontend2
|
||||
assert 9100 <= backend1 <= 9114
|
||||
assert 9200 <= frontend1 <= 9214
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_get_ports_for_work_order_range():
|
||||
"""Test that ports are within expected ranges"""
|
||||
work_order_id = "wo-test123"
|
||||
|
||||
backend, frontend = get_ports_for_work_order(work_order_id)
|
||||
|
||||
assert 9100 <= backend <= 9114
|
||||
assert 9200 <= frontend <= 9214
|
||||
assert frontend == backend + 100
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_get_ports_for_work_order_different_ids():
|
||||
"""Test that different work order IDs can get different ports"""
|
||||
ids = [f"wo-test{i}" for i in range(20)]
|
||||
port_pairs = [get_ports_for_work_order(wid) for wid in ids]
|
||||
|
||||
# With 15 slots, we should see some variation
|
||||
unique_backends = len(set(p[0] for p in port_pairs))
|
||||
assert unique_backends > 1 # At least some variation
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_get_ports_for_work_order_fallback_hash():
|
||||
"""Test fallback to hash when base36 conversion fails"""
|
||||
# Non-alphanumeric work order ID
|
||||
work_order_id = "--------"
|
||||
|
||||
backend, frontend = get_ports_for_work_order(work_order_id)
|
||||
|
||||
# Should still work via hash fallback
|
||||
assert 9100 <= backend <= 9114
|
||||
assert 9200 <= frontend <= 9214
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_is_port_available_mock_available():
|
||||
"""Test port availability check when port is available"""
|
||||
with patch("socket.socket") as mock_socket:
|
||||
mock_socket_instance = mock_socket.return_value.__enter__.return_value
|
||||
mock_socket_instance.bind.return_value = None # Successful bind
|
||||
|
||||
result = is_port_available(9100)
|
||||
|
||||
assert result is True
|
||||
mock_socket_instance.bind.assert_called_once_with(('localhost', 9100))
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_is_port_available_mock_unavailable():
|
||||
"""Test port availability check when port is unavailable"""
|
||||
with patch("socket.socket") as mock_socket:
|
||||
mock_socket_instance = mock_socket.return_value.__enter__.return_value
|
||||
mock_socket_instance.bind.side_effect = OSError("Port in use")
|
||||
|
||||
result = is_port_available(9100)
|
||||
|
||||
assert result is False
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_find_next_available_ports_first_available():
|
||||
"""Test finding ports when first choice is available"""
|
||||
work_order_id = "wo-test123"
|
||||
|
||||
# Mock all ports as available
|
||||
with patch(
|
||||
"src.agent_work_orders.utils.port_allocation.is_port_available",
|
||||
return_value=True,
|
||||
):
|
||||
backend, frontend = find_next_available_ports(work_order_id)
|
||||
|
||||
# Should get the deterministic ports
|
||||
expected_backend, expected_frontend = get_ports_for_work_order(work_order_id)
|
||||
assert backend == expected_backend
|
||||
assert frontend == expected_frontend
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_find_next_available_ports_fallback():
|
||||
"""Test finding ports when first choice is unavailable"""
|
||||
work_order_id = "wo-test123"
|
||||
|
||||
# Mock first port as unavailable, second as available
|
||||
def mock_availability(port):
|
||||
base_backend, _ = get_ports_for_work_order(work_order_id)
|
||||
return port != base_backend and port != base_backend + 100
|
||||
|
||||
with patch(
|
||||
"src.agent_work_orders.utils.port_allocation.is_port_available",
|
||||
side_effect=mock_availability,
|
||||
):
|
||||
backend, frontend = find_next_available_ports(work_order_id)
|
||||
|
||||
# Should get next available ports
|
||||
base_backend, _ = get_ports_for_work_order(work_order_id)
|
||||
assert backend != base_backend # Should be different from base
|
||||
assert 9100 <= backend <= 9114
|
||||
assert frontend == backend + 100
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_find_next_available_ports_exhausted():
|
||||
"""Test that RuntimeError is raised when all ports are unavailable"""
|
||||
work_order_id = "wo-test123"
|
||||
|
||||
# Mock all ports as unavailable
|
||||
with patch(
|
||||
"src.agent_work_orders.utils.port_allocation.is_port_available",
|
||||
return_value=False,
|
||||
):
|
||||
with pytest.raises(RuntimeError) as exc_info:
|
||||
find_next_available_ports(work_order_id)
|
||||
|
||||
assert "No available ports" in str(exc_info.value)
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_create_ports_env_file(tmp_path):
|
||||
"""Test creating .ports.env file"""
|
||||
worktree_path = str(tmp_path)
|
||||
backend_port = 9107
|
||||
frontend_port = 9207
|
||||
|
||||
create_ports_env_file(worktree_path, backend_port, frontend_port)
|
||||
|
||||
ports_env_path = tmp_path / ".ports.env"
|
||||
assert ports_env_path.exists()
|
||||
|
||||
content = ports_env_path.read_text()
|
||||
assert "BACKEND_PORT=9107" in content
|
||||
assert "FRONTEND_PORT=9207" in content
|
||||
assert "VITE_BACKEND_URL=http://localhost:9107" in content
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_create_ports_env_file_overwrites(tmp_path):
|
||||
"""Test that creating .ports.env file overwrites existing file"""
|
||||
worktree_path = str(tmp_path)
|
||||
ports_env_path = tmp_path / ".ports.env"
|
||||
|
||||
# Create existing file with old content
|
||||
ports_env_path.write_text("OLD_CONTENT=true\n")
|
||||
|
||||
# Create new file
|
||||
create_ports_env_file(worktree_path, 9100, 9200)
|
||||
|
||||
content = ports_env_path.read_text()
|
||||
assert "OLD_CONTENT" not in content
|
||||
assert "BACKEND_PORT=9100" in content
|
||||
@@ -7,7 +7,6 @@ from tempfile import TemporaryDirectory
|
||||
|
||||
from src.agent_work_orders.models import SandboxSetupError, SandboxType
|
||||
from src.agent_work_orders.sandbox_manager.git_branch_sandbox import GitBranchSandbox
|
||||
from src.agent_work_orders.sandbox_manager.git_worktree_sandbox import GitWorktreeSandbox
|
||||
from src.agent_work_orders.sandbox_manager.sandbox_factory import SandboxFactory
|
||||
|
||||
|
||||
@@ -197,157 +196,3 @@ def test_sandbox_factory_not_implemented():
|
||||
repository_url="https://github.com/owner/repo",
|
||||
sandbox_identifier="sandbox-test",
|
||||
)
|
||||
|
||||
|
||||
# GitWorktreeSandbox Tests
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_git_worktree_sandbox_setup_success():
|
||||
"""Test successful worktree sandbox setup"""
|
||||
sandbox = GitWorktreeSandbox(
|
||||
repository_url="https://github.com/owner/repo",
|
||||
sandbox_identifier="wo-test123",
|
||||
)
|
||||
|
||||
# Mock port allocation
|
||||
with patch(
|
||||
"src.agent_work_orders.sandbox_manager.git_worktree_sandbox.find_next_available_ports",
|
||||
return_value=(9107, 9207),
|
||||
), patch(
|
||||
"src.agent_work_orders.sandbox_manager.git_worktree_sandbox.create_worktree",
|
||||
return_value=("/tmp/worktree/path", None),
|
||||
), patch(
|
||||
"src.agent_work_orders.sandbox_manager.git_worktree_sandbox.setup_worktree_environment",
|
||||
):
|
||||
await sandbox.setup()
|
||||
|
||||
assert sandbox.backend_port == 9107
|
||||
assert sandbox.frontend_port == 9207
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_git_worktree_sandbox_setup_failure():
|
||||
"""Test failed worktree sandbox setup"""
|
||||
sandbox = GitWorktreeSandbox(
|
||||
repository_url="https://github.com/owner/repo",
|
||||
sandbox_identifier="wo-test123",
|
||||
)
|
||||
|
||||
# Mock port allocation success but worktree creation failure
|
||||
with patch(
|
||||
"src.agent_work_orders.sandbox_manager.git_worktree_sandbox.find_next_available_ports",
|
||||
return_value=(9107, 9207),
|
||||
), patch(
|
||||
"src.agent_work_orders.sandbox_manager.git_worktree_sandbox.create_worktree",
|
||||
return_value=(None, "Failed to create worktree"),
|
||||
):
|
||||
with pytest.raises(SandboxSetupError) as exc_info:
|
||||
await sandbox.setup()
|
||||
|
||||
assert "Failed to create worktree" in str(exc_info.value)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_git_worktree_sandbox_execute_command_success():
|
||||
"""Test successful command execution in worktree sandbox"""
|
||||
with TemporaryDirectory() as tmpdir:
|
||||
sandbox = GitWorktreeSandbox(
|
||||
repository_url="https://github.com/owner/repo",
|
||||
sandbox_identifier="wo-test123",
|
||||
)
|
||||
sandbox.working_dir = tmpdir
|
||||
|
||||
# Mock subprocess
|
||||
mock_process = MagicMock()
|
||||
mock_process.returncode = 0
|
||||
mock_process.communicate = AsyncMock(return_value=(b"Command output", b""))
|
||||
|
||||
with patch("asyncio.create_subprocess_shell", return_value=mock_process):
|
||||
result = await sandbox.execute_command("echo 'test'", timeout=10)
|
||||
|
||||
assert result.success is True
|
||||
assert result.exit_code == 0
|
||||
assert result.stdout == "Command output"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_git_worktree_sandbox_execute_command_timeout():
|
||||
"""Test command execution timeout in worktree sandbox"""
|
||||
import asyncio
|
||||
|
||||
with TemporaryDirectory() as tmpdir:
|
||||
sandbox = GitWorktreeSandbox(
|
||||
repository_url="https://github.com/owner/repo",
|
||||
sandbox_identifier="wo-test123",
|
||||
)
|
||||
sandbox.working_dir = tmpdir
|
||||
|
||||
# Mock subprocess that times out
|
||||
mock_process = MagicMock()
|
||||
mock_process.kill = MagicMock()
|
||||
mock_process.wait = AsyncMock()
|
||||
|
||||
async def mock_communicate():
|
||||
await asyncio.sleep(10)
|
||||
return (b"", b"")
|
||||
|
||||
mock_process.communicate = mock_communicate
|
||||
|
||||
with patch("asyncio.create_subprocess_shell", return_value=mock_process):
|
||||
result = await sandbox.execute_command("sleep 100", timeout=0.1)
|
||||
|
||||
assert result.success is False
|
||||
assert result.exit_code == -1
|
||||
assert "timed out" in result.error_message.lower()
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_git_worktree_sandbox_get_git_branch_name():
|
||||
"""Test getting current git branch name in worktree"""
|
||||
with TemporaryDirectory() as tmpdir:
|
||||
sandbox = GitWorktreeSandbox(
|
||||
repository_url="https://github.com/owner/repo",
|
||||
sandbox_identifier="wo-test123",
|
||||
)
|
||||
sandbox.working_dir = tmpdir
|
||||
|
||||
with patch(
|
||||
"src.agent_work_orders.sandbox_manager.git_worktree_sandbox.get_current_branch",
|
||||
new=AsyncMock(return_value="feat-wo-test123"),
|
||||
):
|
||||
branch = await sandbox.get_git_branch_name()
|
||||
|
||||
assert branch == "feat-wo-test123"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_git_worktree_sandbox_cleanup():
|
||||
"""Test worktree sandbox cleanup"""
|
||||
sandbox = GitWorktreeSandbox(
|
||||
repository_url="https://github.com/owner/repo",
|
||||
sandbox_identifier="wo-test123",
|
||||
)
|
||||
|
||||
with patch(
|
||||
"src.agent_work_orders.sandbox_manager.git_worktree_sandbox.remove_worktree",
|
||||
return_value=(True, None),
|
||||
):
|
||||
await sandbox.cleanup()
|
||||
|
||||
# No exception should be raised
|
||||
|
||||
|
||||
def test_sandbox_factory_git_worktree():
|
||||
"""Test creating git worktree sandbox via factory"""
|
||||
factory = SandboxFactory()
|
||||
|
||||
sandbox = factory.create_sandbox(
|
||||
sandbox_type=SandboxType.GIT_WORKTREE,
|
||||
repository_url="https://github.com/owner/repo",
|
||||
sandbox_identifier="wo-test123",
|
||||
)
|
||||
|
||||
assert isinstance(sandbox, GitWorktreeSandbox)
|
||||
assert sandbox.repository_url == "https://github.com/owner/repo"
|
||||
assert sandbox.sandbox_identifier == "wo-test123"
|
||||
|
||||
@@ -1,372 +0,0 @@
|
||||
"""Tests for Worktree Operations"""
|
||||
|
||||
import os
|
||||
import pytest
|
||||
from pathlib import Path
|
||||
from unittest.mock import MagicMock, patch
|
||||
from tempfile import TemporaryDirectory
|
||||
|
||||
from src.agent_work_orders.utils.worktree_operations import (
|
||||
_get_repo_hash,
|
||||
get_base_repo_path,
|
||||
get_worktree_path,
|
||||
ensure_base_repository,
|
||||
create_worktree,
|
||||
validate_worktree,
|
||||
remove_worktree,
|
||||
setup_worktree_environment,
|
||||
)
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_get_repo_hash_consistent():
|
||||
"""Test that same URL always produces same hash"""
|
||||
url = "https://github.com/owner/repo"
|
||||
|
||||
hash1 = _get_repo_hash(url)
|
||||
hash2 = _get_repo_hash(url)
|
||||
|
||||
assert hash1 == hash2
|
||||
assert len(hash1) == 8 # 8-character hash
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_get_repo_hash_different_urls():
|
||||
"""Test that different URLs produce different hashes"""
|
||||
url1 = "https://github.com/owner/repo1"
|
||||
url2 = "https://github.com/owner/repo2"
|
||||
|
||||
hash1 = _get_repo_hash(url1)
|
||||
hash2 = _get_repo_hash(url2)
|
||||
|
||||
assert hash1 != hash2
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_get_base_repo_path():
|
||||
"""Test getting base repository path"""
|
||||
url = "https://github.com/owner/repo"
|
||||
|
||||
path = get_base_repo_path(url)
|
||||
|
||||
assert "repos" in path
|
||||
assert "main" in path
|
||||
assert Path(path).is_absolute()
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_get_worktree_path():
|
||||
"""Test getting worktree path"""
|
||||
url = "https://github.com/owner/repo"
|
||||
work_order_id = "wo-test123"
|
||||
|
||||
path = get_worktree_path(url, work_order_id)
|
||||
|
||||
assert "repos" in path
|
||||
assert "trees" in path
|
||||
assert work_order_id in path
|
||||
assert Path(path).is_absolute()
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_ensure_base_repository_new_clone():
|
||||
"""Test ensuring base repository when it doesn't exist"""
|
||||
url = "https://github.com/owner/repo"
|
||||
mock_logger = MagicMock()
|
||||
|
||||
mock_result = MagicMock()
|
||||
mock_result.returncode = 0
|
||||
|
||||
with patch("subprocess.run", return_value=mock_result), patch(
|
||||
"os.path.exists", return_value=False
|
||||
), patch("pathlib.Path.mkdir"):
|
||||
base_path, error = ensure_base_repository(url, mock_logger)
|
||||
|
||||
assert base_path is not None
|
||||
assert error is None
|
||||
assert "main" in base_path
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_ensure_base_repository_already_exists():
|
||||
"""Test ensuring base repository when it already exists"""
|
||||
url = "https://github.com/owner/repo"
|
||||
mock_logger = MagicMock()
|
||||
|
||||
mock_result = MagicMock()
|
||||
mock_result.returncode = 0
|
||||
|
||||
with patch("subprocess.run", return_value=mock_result), patch(
|
||||
"os.path.exists", return_value=True
|
||||
):
|
||||
base_path, error = ensure_base_repository(url, mock_logger)
|
||||
|
||||
assert base_path is not None
|
||||
assert error is None
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_ensure_base_repository_clone_failure():
|
||||
"""Test ensuring base repository when clone fails"""
|
||||
url = "https://github.com/owner/repo"
|
||||
mock_logger = MagicMock()
|
||||
|
||||
mock_result = MagicMock()
|
||||
mock_result.returncode = 1
|
||||
mock_result.stderr = "Clone failed"
|
||||
|
||||
with patch("subprocess.run", return_value=mock_result), patch(
|
||||
"os.path.exists", return_value=False
|
||||
), patch("pathlib.Path.mkdir"):
|
||||
base_path, error = ensure_base_repository(url, mock_logger)
|
||||
|
||||
assert base_path is None
|
||||
assert error is not None
|
||||
assert "Clone failed" in error
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_create_worktree_success():
|
||||
"""Test creating worktree successfully"""
|
||||
url = "https://github.com/owner/repo"
|
||||
work_order_id = "wo-test123"
|
||||
branch_name = "feat-test"
|
||||
mock_logger = MagicMock()
|
||||
|
||||
mock_result = MagicMock()
|
||||
mock_result.returncode = 0
|
||||
|
||||
with patch(
|
||||
"src.agent_work_orders.utils.worktree_operations.ensure_base_repository",
|
||||
return_value=("/tmp/base", None),
|
||||
), patch("subprocess.run", return_value=mock_result), patch(
|
||||
"os.path.exists", return_value=False
|
||||
), patch("pathlib.Path.mkdir"):
|
||||
worktree_path, error = create_worktree(
|
||||
url, work_order_id, branch_name, mock_logger
|
||||
)
|
||||
|
||||
assert worktree_path is not None
|
||||
assert error is None
|
||||
assert work_order_id in worktree_path
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_create_worktree_already_exists():
|
||||
"""Test creating worktree when it already exists"""
|
||||
url = "https://github.com/owner/repo"
|
||||
work_order_id = "wo-test123"
|
||||
branch_name = "feat-test"
|
||||
mock_logger = MagicMock()
|
||||
|
||||
expected_path = get_worktree_path(url, work_order_id)
|
||||
|
||||
with patch(
|
||||
"src.agent_work_orders.utils.worktree_operations.ensure_base_repository",
|
||||
return_value=("/tmp/base", None),
|
||||
), patch("os.path.exists", return_value=True):
|
||||
worktree_path, error = create_worktree(
|
||||
url, work_order_id, branch_name, mock_logger
|
||||
)
|
||||
|
||||
assert worktree_path == expected_path
|
||||
assert error is None
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_create_worktree_branch_exists():
|
||||
"""Test creating worktree when branch already exists"""
|
||||
url = "https://github.com/owner/repo"
|
||||
work_order_id = "wo-test123"
|
||||
branch_name = "feat-test"
|
||||
mock_logger = MagicMock()
|
||||
|
||||
# First call fails with "already exists", second succeeds
|
||||
mock_result_fail = MagicMock()
|
||||
mock_result_fail.returncode = 1
|
||||
mock_result_fail.stderr = "already exists"
|
||||
|
||||
mock_result_success = MagicMock()
|
||||
mock_result_success.returncode = 0
|
||||
|
||||
with patch(
|
||||
"src.agent_work_orders.utils.worktree_operations.ensure_base_repository",
|
||||
return_value=("/tmp/base", None),
|
||||
), patch(
|
||||
"subprocess.run", side_effect=[mock_result_success, mock_result_fail, mock_result_success]
|
||||
), patch("os.path.exists", return_value=False), patch("pathlib.Path.mkdir"):
|
||||
worktree_path, error = create_worktree(
|
||||
url, work_order_id, branch_name, mock_logger
|
||||
)
|
||||
|
||||
assert worktree_path is not None
|
||||
assert error is None
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_create_worktree_base_repo_failure():
|
||||
"""Test creating worktree when base repo setup fails"""
|
||||
url = "https://github.com/owner/repo"
|
||||
work_order_id = "wo-test123"
|
||||
branch_name = "feat-test"
|
||||
mock_logger = MagicMock()
|
||||
|
||||
with patch(
|
||||
"src.agent_work_orders.utils.worktree_operations.ensure_base_repository",
|
||||
return_value=(None, "Base repo error"),
|
||||
):
|
||||
worktree_path, error = create_worktree(
|
||||
url, work_order_id, branch_name, mock_logger
|
||||
)
|
||||
|
||||
assert worktree_path is None
|
||||
assert error == "Base repo error"
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_validate_worktree_success():
|
||||
"""Test validating worktree when everything is correct"""
|
||||
url = "https://github.com/owner/repo"
|
||||
work_order_id = "wo-test123"
|
||||
worktree_path = get_worktree_path(url, work_order_id)
|
||||
|
||||
state = {"worktree_path": worktree_path}
|
||||
|
||||
mock_result = MagicMock()
|
||||
mock_result.returncode = 0
|
||||
mock_result.stdout = worktree_path # Git knows about it
|
||||
|
||||
with patch("os.path.exists", return_value=True), patch(
|
||||
"subprocess.run", return_value=mock_result
|
||||
):
|
||||
is_valid, error = validate_worktree(url, work_order_id, state)
|
||||
|
||||
assert is_valid is True
|
||||
assert error is None
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_validate_worktree_no_path_in_state():
|
||||
"""Test validating worktree when state has no path"""
|
||||
url = "https://github.com/owner/repo"
|
||||
work_order_id = "wo-test123"
|
||||
state = {} # No worktree_path
|
||||
|
||||
is_valid, error = validate_worktree(url, work_order_id, state)
|
||||
|
||||
assert is_valid is False
|
||||
assert "No worktree_path" in error
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_validate_worktree_directory_not_found():
|
||||
"""Test validating worktree when directory doesn't exist"""
|
||||
url = "https://github.com/owner/repo"
|
||||
work_order_id = "wo-test123"
|
||||
worktree_path = get_worktree_path(url, work_order_id)
|
||||
|
||||
state = {"worktree_path": worktree_path}
|
||||
|
||||
with patch("os.path.exists", return_value=False):
|
||||
is_valid, error = validate_worktree(url, work_order_id, state)
|
||||
|
||||
assert is_valid is False
|
||||
assert "not found" in error
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_validate_worktree_not_registered_with_git():
|
||||
"""Test validating worktree when git doesn't know about it"""
|
||||
url = "https://github.com/owner/repo"
|
||||
work_order_id = "wo-test123"
|
||||
worktree_path = get_worktree_path(url, work_order_id)
|
||||
|
||||
state = {"worktree_path": worktree_path}
|
||||
|
||||
mock_result = MagicMock()
|
||||
mock_result.returncode = 0
|
||||
mock_result.stdout = "/some/other/path" # Doesn't contain our path
|
||||
|
||||
with patch("os.path.exists", return_value=True), patch(
|
||||
"subprocess.run", return_value=mock_result
|
||||
):
|
||||
is_valid, error = validate_worktree(url, work_order_id, state)
|
||||
|
||||
assert is_valid is False
|
||||
assert "not registered" in error
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_remove_worktree_success():
|
||||
"""Test removing worktree successfully"""
|
||||
url = "https://github.com/owner/repo"
|
||||
work_order_id = "wo-test123"
|
||||
mock_logger = MagicMock()
|
||||
|
||||
mock_result = MagicMock()
|
||||
mock_result.returncode = 0
|
||||
|
||||
with patch("os.path.exists", return_value=True), patch(
|
||||
"subprocess.run", return_value=mock_result
|
||||
):
|
||||
success, error = remove_worktree(url, work_order_id, mock_logger)
|
||||
|
||||
assert success is True
|
||||
assert error is None
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_remove_worktree_fallback_to_manual():
|
||||
"""Test removing worktree with fallback to manual removal"""
|
||||
url = "https://github.com/owner/repo"
|
||||
work_order_id = "wo-test123"
|
||||
mock_logger = MagicMock()
|
||||
|
||||
mock_result = MagicMock()
|
||||
mock_result.returncode = 1
|
||||
mock_result.stderr = "Git remove failed"
|
||||
|
||||
with patch("os.path.exists", return_value=True), patch(
|
||||
"subprocess.run", return_value=mock_result
|
||||
), patch("shutil.rmtree"):
|
||||
success, error = remove_worktree(url, work_order_id, mock_logger)
|
||||
|
||||
# Should succeed via manual cleanup
|
||||
assert success is True
|
||||
assert error is None
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_remove_worktree_no_base_repo():
|
||||
"""Test removing worktree when base repo doesn't exist"""
|
||||
url = "https://github.com/owner/repo"
|
||||
work_order_id = "wo-test123"
|
||||
mock_logger = MagicMock()
|
||||
|
||||
def mock_exists(path):
|
||||
# Base repo doesn't exist, but worktree directory does
|
||||
return "main" not in path
|
||||
|
||||
with patch("os.path.exists", side_effect=mock_exists), patch("shutil.rmtree"):
|
||||
success, error = remove_worktree(url, work_order_id, mock_logger)
|
||||
|
||||
assert success is True
|
||||
assert error is None
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_setup_worktree_environment(tmp_path):
|
||||
"""Test setting up worktree environment"""
|
||||
worktree_path = str(tmp_path)
|
||||
backend_port = 9107
|
||||
frontend_port = 9207
|
||||
mock_logger = MagicMock()
|
||||
|
||||
setup_worktree_environment(worktree_path, backend_port, frontend_port, mock_logger)
|
||||
|
||||
ports_env_path = tmp_path / ".ports.env"
|
||||
assert ports_env_path.exists()
|
||||
|
||||
content = ports_env_path.read_text()
|
||||
assert "BACKEND_PORT=9107" in content
|
||||
assert "FRONTEND_PORT=9207" in content
|
||||
Reference in New Issue
Block a user