Alex Blank
0e6521e433
- Add mappings array support to git_mount entries - Clone repository once per git_mount entry, mount multiple subdirectories - Normalize legacy source_path+target_path to mappings on read - Update _merge_git_mounts to dedup by (remote_url, branch) and concatenate mappings - Add _normalize_git_mount, _clone_git_repo, _resolve_git_mount_mappings helpers - Update GitMountItem Pydantic model with GitMountMapping and model_validator - Update frontend GitMountEditor component with mappings UI - Auto-convert legacy git mount entries to mappings format on load - Add 15 backend unit tests for normalization, resolution, and glob expansion - Update existing config profile resolver tests for new merge behavior Quality gates: pytest 167 passed, frontend typecheck clean Addresses: config-profile-multi-repo-mounts
214 lines
8.3 KiB
Python
214 lines
8.3 KiB
Python
"""Unit tests for git mount resolution with multi-mapping support."""
|
|
|
|
import os
|
|
import tempfile
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
import pytest
|
|
|
|
from src.api.tool_instances import (
|
|
_clone_git_repo,
|
|
_expand_glob_source,
|
|
_normalize_git_mount,
|
|
_resolve_git_mount_mappings,
|
|
_resolve_single_git_mount,
|
|
)
|
|
|
|
|
|
class TestNormalizeGitMount:
|
|
"""Tests for _normalize_git_mount."""
|
|
|
|
def test_legacy_to_mappings(self) -> None:
|
|
"""Legacy source_path + target_path becomes mappings array."""
|
|
entry = {
|
|
"remote_url": "https://github.com/user/repo.git",
|
|
"source_path": "packages/api",
|
|
"target_path": "/app/api",
|
|
"branch": "main",
|
|
}
|
|
result = _normalize_git_mount(entry)
|
|
assert "mappings" in result
|
|
assert result["mappings"] == [
|
|
{"source_path": "packages/api", "target_path": "/app/api"}
|
|
]
|
|
assert "source_path" not in result
|
|
assert "target_path" not in result
|
|
assert result["remote_url"] == "https://github.com/user/repo.git"
|
|
assert result["branch"] == "main"
|
|
|
|
def test_already_mappings(self) -> None:
|
|
"""Entry already with mappings is left unchanged."""
|
|
entry = {
|
|
"remote_url": "https://github.com/user/repo.git",
|
|
"branch": "main",
|
|
"mappings": [
|
|
{"source_path": "a", "target_path": "/a"},
|
|
{"source_path": "b", "target_path": "/b"},
|
|
],
|
|
}
|
|
result = _normalize_git_mount(entry)
|
|
assert result["mappings"] == [
|
|
{"source_path": "a", "target_path": "/a"},
|
|
{"source_path": "b", "target_path": "/b"},
|
|
]
|
|
assert "source_path" not in result
|
|
assert "target_path" not in result
|
|
|
|
def test_missing_target_path_no_mappings(self) -> None:
|
|
"""Entry with source_path but no target_path creates empty mappings."""
|
|
entry = {
|
|
"remote_url": "https://github.com/user/repo.git",
|
|
"source_path": "src",
|
|
}
|
|
result = _normalize_git_mount(entry)
|
|
assert "mappings" not in result
|
|
|
|
|
|
class TestResolveGitMountMappings:
|
|
"""Tests for _resolve_git_mount_mappings."""
|
|
|
|
def test_single_mapping(self) -> None:
|
|
"""A single mapping produces one volume mount."""
|
|
with tempfile.TemporaryDirectory() as repo_path:
|
|
os.makedirs(os.path.join(repo_path, "packages", "api"))
|
|
mappings = [
|
|
{"source_path": "packages/api", "target_path": "/app/api"},
|
|
]
|
|
result = _resolve_git_mount_mappings(repo_path, mappings, None)
|
|
assert len(result) == 1
|
|
assert result[0]["source"] == os.path.join(repo_path, "packages", "api")
|
|
assert result[0]["target"] == "/app/api"
|
|
assert result[0]["type"] == "bind"
|
|
|
|
def test_multiple_mappings(self) -> None:
|
|
"""Multiple mappings from same repo produce multiple mounts."""
|
|
with tempfile.TemporaryDirectory() as repo_path:
|
|
os.makedirs(os.path.join(repo_path, "packages", "api"))
|
|
os.makedirs(os.path.join(repo_path, "packages", "web"))
|
|
mappings = [
|
|
{"source_path": "packages/api", "target_path": "/app/api"},
|
|
{"source_path": "packages/web", "target_path": "/app/web"},
|
|
]
|
|
result = _resolve_git_mount_mappings(repo_path, mappings, None)
|
|
assert len(result) == 2
|
|
targets = {r["target"] for r in result}
|
|
assert targets == {"/app/api", "/app/web"}
|
|
|
|
def test_relative_target_path(self) -> None:
|
|
"""Relative target_path is resolved against working_directory."""
|
|
with tempfile.TemporaryDirectory() as repo_path:
|
|
os.makedirs(os.path.join(repo_path, "src"))
|
|
mappings = [
|
|
{"source_path": "src", "target_path": "code"},
|
|
]
|
|
result = _resolve_git_mount_mappings(repo_path, mappings, "/workspace")
|
|
assert len(result) == 1
|
|
assert result[0]["target"] == "/workspace/code"
|
|
|
|
def test_glob_expansion(self) -> None:
|
|
"""Glob patterns in source_path are expanded."""
|
|
with tempfile.TemporaryDirectory() as repo_path:
|
|
os.makedirs(os.path.join(repo_path, "packages", "api"))
|
|
os.makedirs(os.path.join(repo_path, "packages", "web"))
|
|
mappings = [
|
|
{"source_path": "packages/*", "target_path": "/app/packages"},
|
|
]
|
|
result = _resolve_git_mount_mappings(repo_path, mappings, None)
|
|
assert len(result) == 2
|
|
targets = {r["target"] for r in result}
|
|
assert targets == {
|
|
os.path.join("/app/packages", "packages", "api"),
|
|
os.path.join("/app/packages", "packages", "web"),
|
|
}
|
|
|
|
def test_missing_target_path_skipped(self) -> None:
|
|
"""Mapping without target_path is skipped."""
|
|
with tempfile.TemporaryDirectory() as repo_path:
|
|
mappings = [
|
|
{"source_path": "src"},
|
|
]
|
|
result = _resolve_git_mount_mappings(repo_path, mappings, None)
|
|
assert len(result) == 0
|
|
|
|
def test_no_working_directory_for_relative_target(self) -> None:
|
|
"""Relative target without working_directory is skipped."""
|
|
with tempfile.TemporaryDirectory() as repo_path:
|
|
os.makedirs(os.path.join(repo_path, "src"))
|
|
mappings = [
|
|
{"source_path": "src", "target_path": "code"},
|
|
]
|
|
result = _resolve_git_mount_mappings(repo_path, mappings, None)
|
|
assert len(result) == 0
|
|
|
|
|
|
class TestResolveSingleGitMount:
|
|
"""Tests for _resolve_single_git_mount."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_missing_remote_url(self) -> None:
|
|
"""Git mount without remote_url returns empty list."""
|
|
result = await _resolve_single_git_mount(
|
|
MagicMock(), {"mappings": [{"source_path": ".", "target_path": "/app"}]}, "/tmp", None
|
|
)
|
|
assert result == []
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_missing_instance_dir(self) -> None:
|
|
"""Git mount without instance_dir returns empty list."""
|
|
result = await _resolve_single_git_mount(
|
|
MagicMock(),
|
|
{"remote_url": "https://github.com/user/repo.git", "mappings": [{"source_path": ".", "target_path": "/app"}]},
|
|
None,
|
|
None,
|
|
)
|
|
assert result == []
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_legacy_format_normalized(self) -> None:
|
|
"""Legacy format is normalized and resolved."""
|
|
with tempfile.TemporaryDirectory() as instance_dir:
|
|
with patch(
|
|
"src.api.tool_instances._clone_git_repo",
|
|
return_value=os.path.join(instance_dir, "repo-clone"),
|
|
):
|
|
os.makedirs(os.path.join(instance_dir, "repo-clone", "src"))
|
|
result = await _resolve_single_git_mount(
|
|
MagicMock(),
|
|
{
|
|
"remote_url": "https://github.com/user/repo.git",
|
|
"source_path": "src",
|
|
"target_path": "/app/src",
|
|
},
|
|
instance_dir,
|
|
None,
|
|
)
|
|
assert len(result) == 1
|
|
assert result[0]["target"] == "/app/src"
|
|
|
|
|
|
class TestExpandGlobSource:
|
|
"""Tests for _expand_glob_source."""
|
|
|
|
def test_no_glob(self) -> None:
|
|
"""Non-glob path returns single item if exists."""
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
path = os.path.join(tmp, "file.txt")
|
|
open(path, "w").close()
|
|
result = _expand_glob_source(path, tmp)
|
|
assert result == [path]
|
|
|
|
def test_no_glob_missing(self) -> None:
|
|
"""Non-glob path that doesn't exist returns empty list."""
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
path = os.path.join(tmp, "missing.txt")
|
|
result = _expand_glob_source(path, tmp)
|
|
assert result == []
|
|
|
|
def test_glob_pattern(self) -> None:
|
|
"""Glob pattern expands to matched paths."""
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
open(os.path.join(tmp, "a.txt"), "w").close()
|
|
open(os.path.join(tmp, "b.txt"), "w").close()
|
|
result = _expand_glob_source(os.path.join(tmp, "*.txt"), tmp)
|
|
assert len(result) == 2
|