test: add comprehensive tests for config profile git mounts
- Add git mount merge function tests - Add profile resolution tests with git mounts - Add integration tests for CRUD with git mounts - Add glob expansion tests (patterns, limits, repo boundary) - Add branch checkout tests (success and failure) - Add error handling tests for missing repos/invalid UUIDs All 52 tests pass.
This commit is contained in:
@@ -133,6 +133,65 @@ def authenticated_client(test_client) -> Generator[TestClient, None, None]:
|
|||||||
yield test_client
|
yield test_client
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.fixture
|
||||||
|
def test_project_and_repo(authenticated_client) -> tuple[str, str]:
|
||||||
|
"""Create a project and repository directly in the database."""
|
||||||
|
import uuid
|
||||||
|
from src.models.project import Project
|
||||||
|
from src.models.git_repository import GitRepository
|
||||||
|
|
||||||
|
project_id = uuid.uuid4()
|
||||||
|
repo_id = uuid.uuid4()
|
||||||
|
user_id = None
|
||||||
|
|
||||||
|
# Get user ID from session
|
||||||
|
async def get_user_id():
|
||||||
|
nonlocal user_id
|
||||||
|
from src.auth.session import decode_session_cookie
|
||||||
|
settings = Settings()
|
||||||
|
session_cookie = authenticated_client.cookies.get("session")
|
||||||
|
if session_cookie:
|
||||||
|
session = decode_session_cookie(settings=settings, cookie_value=session_cookie)
|
||||||
|
if session:
|
||||||
|
user_id = uuid.UUID(session["user_id"])
|
||||||
|
|
||||||
|
asyncio.run(get_user_id())
|
||||||
|
|
||||||
|
if not user_id:
|
||||||
|
raise RuntimeError("Could not get user ID from authenticated client")
|
||||||
|
|
||||||
|
async def create_project_and_repo():
|
||||||
|
override_fn = app.dependency_overrides.get(get_db_session)
|
||||||
|
if override_fn:
|
||||||
|
gen = override_fn()
|
||||||
|
session = await gen.asend(None)
|
||||||
|
try:
|
||||||
|
project = Project(
|
||||||
|
id=project_id,
|
||||||
|
name="test-project",
|
||||||
|
description="Test project",
|
||||||
|
owner_id=user_id,
|
||||||
|
)
|
||||||
|
session.add(project)
|
||||||
|
|
||||||
|
repo = GitRepository(
|
||||||
|
id=repo_id,
|
||||||
|
name="test-repo",
|
||||||
|
path="/tmp/test-repo",
|
||||||
|
project_id=project_id,
|
||||||
|
owner_id=user_id,
|
||||||
|
remote_url="https://github.com/test/repo.git",
|
||||||
|
)
|
||||||
|
session.add(repo)
|
||||||
|
await session.commit()
|
||||||
|
finally:
|
||||||
|
await gen.aclose()
|
||||||
|
|
||||||
|
asyncio.run(create_project_and_repo())
|
||||||
|
|
||||||
|
return str(project_id), str(repo_id)
|
||||||
|
|
||||||
|
|
||||||
@pytest.fixture
|
@pytest.fixture
|
||||||
def admin_client(test_client) -> Generator[TestClient, None, None]:
|
def admin_client(test_client) -> Generator[TestClient, None, None]:
|
||||||
"""Provide an authenticated test client with an admin user."""
|
"""Provide an authenticated test client with an admin user."""
|
||||||
|
|||||||
@@ -320,3 +320,134 @@ class TestConfigProfilesAPI:
|
|||||||
assert response.status_code == 200
|
assert response.status_code == 200
|
||||||
data = response.json()
|
data = response.json()
|
||||||
assert data["profile_id"] is None
|
assert data["profile_id"] is None
|
||||||
|
|
||||||
|
def test_create_config_profile_with_git_mounts(self, authenticated_client: TestClient, test_project_and_repo) -> None:
|
||||||
|
"""Test creating a config profile with git mounts."""
|
||||||
|
_project_id, repo_id = test_project_and_repo
|
||||||
|
|
||||||
|
response = authenticated_client.post(
|
||||||
|
"/config-profiles",
|
||||||
|
json={
|
||||||
|
"name": "git-mount-profile",
|
||||||
|
"env_vars": {},
|
||||||
|
"files": {},
|
||||||
|
"git_mounts": [
|
||||||
|
{
|
||||||
|
"repo_id": repo_id,
|
||||||
|
"source_path": ".",
|
||||||
|
"target_path": "/app",
|
||||||
|
"branch": "main",
|
||||||
|
}
|
||||||
|
],
|
||||||
|
},
|
||||||
|
)
|
||||||
|
assert response.status_code == 201
|
||||||
|
data = response.json()
|
||||||
|
assert data["name"] == "git-mount-profile"
|
||||||
|
assert len(data["git_mounts"]) == 1
|
||||||
|
assert data["git_mounts"][0]["target_path"] == "/app"
|
||||||
|
assert data["git_mounts"][0]["branch"] == "main"
|
||||||
|
|
||||||
|
def test_update_config_profile_git_mounts(self, authenticated_client: TestClient, test_project_and_repo) -> None:
|
||||||
|
"""Test updating git mounts on a config profile."""
|
||||||
|
_project_id, repo_id = test_project_and_repo
|
||||||
|
|
||||||
|
# Create profile first
|
||||||
|
create_response = authenticated_client.post(
|
||||||
|
"/config-profiles",
|
||||||
|
json={
|
||||||
|
"name": "update-git-mounts",
|
||||||
|
"env_vars": {},
|
||||||
|
"files": {},
|
||||||
|
},
|
||||||
|
)
|
||||||
|
profile_id = create_response.json()["id"]
|
||||||
|
|
||||||
|
# Update with git mounts
|
||||||
|
response = authenticated_client.put(
|
||||||
|
f"/config-profiles/{profile_id}",
|
||||||
|
json={
|
||||||
|
"git_mounts": [
|
||||||
|
{
|
||||||
|
"repo_id": repo_id,
|
||||||
|
"source_path": "config",
|
||||||
|
"target_path": "/config",
|
||||||
|
}
|
||||||
|
],
|
||||||
|
},
|
||||||
|
)
|
||||||
|
assert response.status_code == 200
|
||||||
|
data = response.json()
|
||||||
|
assert len(data["git_mounts"]) == 1
|
||||||
|
assert data["git_mounts"][0]["source_path"] == "config"
|
||||||
|
|
||||||
|
def test_create_config_profile_invalid_git_mount_source_path(self, authenticated_client: TestClient, test_project_and_repo) -> None:
|
||||||
|
"""Test that invalid git mount source paths are rejected."""
|
||||||
|
_project_id, repo_id = test_project_and_repo
|
||||||
|
|
||||||
|
response = authenticated_client.post(
|
||||||
|
"/config-profiles",
|
||||||
|
json={
|
||||||
|
"name": "bad-git-mount",
|
||||||
|
"env_vars": {},
|
||||||
|
"files": {},
|
||||||
|
"git_mounts": [
|
||||||
|
{
|
||||||
|
"repo_id": repo_id,
|
||||||
|
"source_path": "/absolute/path",
|
||||||
|
"target_path": "/app",
|
||||||
|
}
|
||||||
|
],
|
||||||
|
},
|
||||||
|
)
|
||||||
|
assert response.status_code == 422
|
||||||
|
|
||||||
|
def test_create_config_profile_invalid_git_mount_target_path(self, authenticated_client: TestClient, test_project_and_repo) -> None:
|
||||||
|
"""Test that invalid git mount target paths are rejected."""
|
||||||
|
_project_id, repo_id = test_project_and_repo
|
||||||
|
|
||||||
|
response = authenticated_client.post(
|
||||||
|
"/config-profiles",
|
||||||
|
json={
|
||||||
|
"name": "bad-git-mount-target",
|
||||||
|
"env_vars": {},
|
||||||
|
"files": {},
|
||||||
|
"git_mounts": [
|
||||||
|
{
|
||||||
|
"repo_id": repo_id,
|
||||||
|
"source_path": ".",
|
||||||
|
"target_path": "relative/path",
|
||||||
|
}
|
||||||
|
],
|
||||||
|
},
|
||||||
|
)
|
||||||
|
assert response.status_code == 422
|
||||||
|
|
||||||
|
def test_preview_config_profile_with_git_mounts(self, authenticated_client: TestClient, test_project_and_repo) -> None:
|
||||||
|
"""Test previewing a profile with git mounts."""
|
||||||
|
_project_id, repo_id = test_project_and_repo
|
||||||
|
|
||||||
|
# Create profile with git mounts
|
||||||
|
create_response = authenticated_client.post(
|
||||||
|
"/config-profiles",
|
||||||
|
json={
|
||||||
|
"name": "preview-git-mounts",
|
||||||
|
"env_vars": {},
|
||||||
|
"files": {},
|
||||||
|
"git_mounts": [
|
||||||
|
{
|
||||||
|
"repo_id": repo_id,
|
||||||
|
"source_path": ".",
|
||||||
|
"target_path": "/app",
|
||||||
|
}
|
||||||
|
],
|
||||||
|
},
|
||||||
|
)
|
||||||
|
profile_id = create_response.json()["id"]
|
||||||
|
|
||||||
|
# Preview
|
||||||
|
response = authenticated_client.get(f"/config-profiles/{profile_id}/preview")
|
||||||
|
assert response.status_code == 200
|
||||||
|
data = response.json()
|
||||||
|
assert len(data["git_mounts"]) == 1
|
||||||
|
assert data["git_mounts"][0]["repo_id"] == repo_id
|
||||||
|
|||||||
@@ -13,6 +13,7 @@ from src.services.config_profile_resolver import (
|
|||||||
_merge_files,
|
_merge_files,
|
||||||
_merge_mounts,
|
_merge_mounts,
|
||||||
_merge_runtime_hints,
|
_merge_runtime_hints,
|
||||||
|
_merge_git_mounts,
|
||||||
)
|
)
|
||||||
|
|
||||||
|
|
||||||
@@ -97,6 +98,39 @@ class TestMergeFunctions:
|
|||||||
assert result["/app"].mode == "ro"
|
assert result["/app"].mode == "ro"
|
||||||
assert overrides == {"/app": "source"}
|
assert overrides == {"/app": "source"}
|
||||||
|
|
||||||
|
def test_merge_git_mounts_basic(self) -> None:
|
||||||
|
"""Test basic git mount merging."""
|
||||||
|
result = _merge_git_mounts(
|
||||||
|
[],
|
||||||
|
[{"repo_id": "repo1", "source_path": ".", "target_path": "/app"}],
|
||||||
|
"source",
|
||||||
|
)
|
||||||
|
assert len(result) == 1
|
||||||
|
assert result[0]["repo_id"] == "repo1"
|
||||||
|
assert result[0]["target_path"] == "/app"
|
||||||
|
|
||||||
|
def test_merge_git_mounts_override_same_repo_target(self) -> None:
|
||||||
|
"""Test that git mounts with same repo+target override."""
|
||||||
|
result = _merge_git_mounts(
|
||||||
|
[{"repo_id": "repo1", "source_path": ".", "target_path": "/app", "branch": "main"}],
|
||||||
|
[{"repo_id": "repo1", "source_path": "src", "target_path": "/app", "branch": "dev"}],
|
||||||
|
"source",
|
||||||
|
)
|
||||||
|
assert len(result) == 1
|
||||||
|
assert result[0]["source_path"] == "src"
|
||||||
|
assert result[0]["branch"] == "dev"
|
||||||
|
|
||||||
|
def test_merge_git_mounts_different_targets(self) -> None:
|
||||||
|
"""Test that git mounts with different targets are preserved."""
|
||||||
|
result = _merge_git_mounts(
|
||||||
|
[{"repo_id": "repo1", "source_path": ".", "target_path": "/app"}],
|
||||||
|
[{"repo_id": "repo2", "source_path": ".", "target_path": "/config"}],
|
||||||
|
"source",
|
||||||
|
)
|
||||||
|
assert len(result) == 2
|
||||||
|
targets = {m["target_path"] for m in result}
|
||||||
|
assert targets == {"/app", "/config"}
|
||||||
|
|
||||||
|
|
||||||
class TestResolveProfile:
|
class TestResolveProfile:
|
||||||
"""Unit tests for profile resolution."""
|
"""Unit tests for profile resolution."""
|
||||||
@@ -250,6 +284,76 @@ class TestResolveProfile:
|
|||||||
with pytest.raises(ConfigProfileCycleError):
|
with pytest.raises(ConfigProfileCycleError):
|
||||||
await resolve_profile(db_session, profile_a.id)
|
await resolve_profile(db_session, profile_a.id)
|
||||||
|
|
||||||
|
@pytest.mark.asyncio
|
||||||
|
async def test_resolve_profile_with_git_mounts(self, db_session: AsyncSession) -> None:
|
||||||
|
"""Test resolving a profile with git mounts."""
|
||||||
|
user_id = uuid.uuid4()
|
||||||
|
|
||||||
|
profile = ConfigProfile(
|
||||||
|
id=uuid.uuid4(),
|
||||||
|
user_id=user_id,
|
||||||
|
name="with-git-mounts",
|
||||||
|
env_vars={},
|
||||||
|
files={},
|
||||||
|
git_mounts=[
|
||||||
|
{"repo_id": "repo1", "source_path": ".", "target_path": "/app"},
|
||||||
|
],
|
||||||
|
)
|
||||||
|
db_session.add(profile)
|
||||||
|
await db_session.commit()
|
||||||
|
|
||||||
|
result = await resolve_profile(db_session, profile.id)
|
||||||
|
assert len(result.git_mounts) == 1
|
||||||
|
assert result.git_mounts[0]["repo_id"] == "repo1"
|
||||||
|
assert result.git_mounts[0]["target_path"] == "/app"
|
||||||
|
|
||||||
|
@pytest.mark.asyncio
|
||||||
|
async def test_resolve_profile_with_git_mount_includes(self, db_session: AsyncSession) -> None:
|
||||||
|
"""Test resolving a profile that includes another with git mounts."""
|
||||||
|
user_id = uuid.uuid4()
|
||||||
|
|
||||||
|
# Create base profile with git mount
|
||||||
|
base = ConfigProfile(
|
||||||
|
id=uuid.uuid4(),
|
||||||
|
user_id=user_id,
|
||||||
|
name="base",
|
||||||
|
env_vars={},
|
||||||
|
files={},
|
||||||
|
git_mounts=[
|
||||||
|
{"repo_id": "repo1", "source_path": ".", "target_path": "/app"},
|
||||||
|
],
|
||||||
|
)
|
||||||
|
db_session.add(base)
|
||||||
|
|
||||||
|
# Create child profile with its own git mount
|
||||||
|
child = ConfigProfile(
|
||||||
|
id=uuid.uuid4(),
|
||||||
|
user_id=user_id,
|
||||||
|
name="child",
|
||||||
|
env_vars={},
|
||||||
|
files={},
|
||||||
|
git_mounts=[
|
||||||
|
{"repo_id": "repo2", "source_path": "config", "target_path": "/config"},
|
||||||
|
],
|
||||||
|
)
|
||||||
|
db_session.add(child)
|
||||||
|
await db_session.commit()
|
||||||
|
|
||||||
|
# Create include relationship
|
||||||
|
include = ConfigProfileInclude(
|
||||||
|
id=uuid.uuid4(),
|
||||||
|
profile_id=child.id,
|
||||||
|
included_profile_id=base.id,
|
||||||
|
order_index=0,
|
||||||
|
)
|
||||||
|
db_session.add(include)
|
||||||
|
await db_session.commit()
|
||||||
|
|
||||||
|
result = await resolve_profile(db_session, child.id)
|
||||||
|
assert len(result.git_mounts) == 2
|
||||||
|
targets = {m["target_path"] for m in result.git_mounts}
|
||||||
|
assert targets == {"/app", "/config"}
|
||||||
|
|
||||||
@pytest.mark.asyncio
|
@pytest.mark.asyncio
|
||||||
async def test_resolve_profile_not_found(self, db_session: AsyncSession) -> None:
|
async def test_resolve_profile_not_found(self, db_session: AsyncSession) -> None:
|
||||||
"""Test resolving a non-existent profile."""
|
"""Test resolving a non-existent profile."""
|
||||||
|
|||||||
@@ -0,0 +1,139 @@
|
|||||||
|
"""Unit tests for git mount resolution in tool instances."""
|
||||||
|
|
||||||
|
import os
|
||||||
|
import tempfile
|
||||||
|
from pathlib import Path
|
||||||
|
|
||||||
|
import pytest
|
||||||
|
|
||||||
|
from src.api.tool_instances import (
|
||||||
|
_checkout_branch,
|
||||||
|
_expand_glob_source,
|
||||||
|
_resolve_single_git_mount,
|
||||||
|
)
|
||||||
|
from src.services.config_profile_resolver import ResolvedProfile
|
||||||
|
|
||||||
|
|
||||||
|
class TestExpandGlobSource:
|
||||||
|
"""Unit tests for glob pattern expansion."""
|
||||||
|
|
||||||
|
def test_no_glob_single_file(self, tmp_path: Path) -> None:
|
||||||
|
"""Test non-glob path returns single file."""
|
||||||
|
test_file = tmp_path / "test.txt"
|
||||||
|
test_file.write_text("content")
|
||||||
|
|
||||||
|
result = _expand_glob_source(str(test_file), str(tmp_path))
|
||||||
|
assert len(result) == 1
|
||||||
|
assert result[0] == str(test_file)
|
||||||
|
|
||||||
|
def test_no_glob_missing_file(self, tmp_path: Path) -> None:
|
||||||
|
"""Test non-glob missing file returns empty list."""
|
||||||
|
missing_file = tmp_path / "missing.txt"
|
||||||
|
|
||||||
|
result = _expand_glob_source(str(missing_file), str(tmp_path))
|
||||||
|
assert len(result) == 0
|
||||||
|
|
||||||
|
def test_glob_pattern(self, tmp_path: Path) -> None:
|
||||||
|
"""Test glob pattern matches files."""
|
||||||
|
(tmp_path / "file1.txt").write_text("content1")
|
||||||
|
(tmp_path / "file2.txt").write_text("content2")
|
||||||
|
(tmp_path / "other.py").write_text("code")
|
||||||
|
|
||||||
|
result = _expand_glob_source(str(tmp_path / "*.txt"), str(tmp_path))
|
||||||
|
assert len(result) == 2
|
||||||
|
assert all(f.endswith(".txt") for f in result)
|
||||||
|
|
||||||
|
def test_glob_recursive(self, tmp_path: Path) -> None:
|
||||||
|
"""Test recursive glob pattern."""
|
||||||
|
subdir = tmp_path / "subdir"
|
||||||
|
subdir.mkdir()
|
||||||
|
(subdir / "nested.txt").write_text("content")
|
||||||
|
|
||||||
|
result = _expand_glob_source(str(tmp_path / "**" / "*.txt"), str(tmp_path))
|
||||||
|
assert len(result) == 1
|
||||||
|
assert "nested.txt" in result[0]
|
||||||
|
|
||||||
|
def test_glob_limit_enforced(self, tmp_path: Path) -> None:
|
||||||
|
"""Test that glob matches are limited to prevent abuse."""
|
||||||
|
# Create more than 100 files
|
||||||
|
for i in range(105):
|
||||||
|
(tmp_path / f"file{i}.txt").write_text("content")
|
||||||
|
|
||||||
|
result = _expand_glob_source(str(tmp_path / "*.txt"), str(tmp_path))
|
||||||
|
assert len(result) == 100 # MAX_GLOB_MATCHES limit
|
||||||
|
|
||||||
|
def test_glob_escapes_repo(self, tmp_path: Path) -> None:
|
||||||
|
"""Test that glob results outside repo are filtered."""
|
||||||
|
other_dir = tmp_path.parent / "other"
|
||||||
|
other_dir.mkdir(exist_ok=True)
|
||||||
|
(other_dir / "outside.txt").write_text("content")
|
||||||
|
|
||||||
|
result = _expand_glob_source(str(tmp_path.parent / "*" / "*.txt"), str(tmp_path))
|
||||||
|
# Should only include files within tmp_path, not other_dir
|
||||||
|
assert all(r.startswith(str(tmp_path)) for r in result)
|
||||||
|
|
||||||
|
|
||||||
|
class TestCheckoutBranch:
|
||||||
|
"""Unit tests for branch checkout."""
|
||||||
|
|
||||||
|
def test_checkout_existing_branch(self, tmp_path: Path) -> None:
|
||||||
|
"""Test checking out an existing branch."""
|
||||||
|
# Initialize git repo
|
||||||
|
os.system(f"cd {tmp_path} && git init && git config user.email 'test@test.com' && git config user.name 'Test'")
|
||||||
|
(tmp_path / "file.txt").write_text("content")
|
||||||
|
os.system(f"cd {tmp_path} && git add . && git commit -m 'initial'")
|
||||||
|
os.system(f"cd {tmp_path} && git branch feature")
|
||||||
|
|
||||||
|
_checkout_branch(str(tmp_path), "feature")
|
||||||
|
|
||||||
|
# Verify we're on feature branch
|
||||||
|
result = os.popen(f"cd {tmp_path} && git branch --show-current").read().strip()
|
||||||
|
assert result == "feature"
|
||||||
|
|
||||||
|
def test_checkout_nonexistent_branch(self, tmp_path: Path) -> None:
|
||||||
|
"""Test checking out a non-existent branch raises error."""
|
||||||
|
os.system(f"cd {tmp_path} && git init && git config user.email 'test@test.com' && git config user.name 'Test'")
|
||||||
|
(tmp_path / "file.txt").write_text("content")
|
||||||
|
os.system(f"cd {tmp_path} && git add . && git commit -m 'initial'")
|
||||||
|
|
||||||
|
with pytest.raises(RuntimeError, match="Failed to checkout branch"):
|
||||||
|
_checkout_branch(str(tmp_path), "nonexistent")
|
||||||
|
|
||||||
|
|
||||||
|
class TestResolveSingleGitMount:
|
||||||
|
"""Unit tests for resolving a single git mount."""
|
||||||
|
|
||||||
|
@pytest.mark.asyncio
|
||||||
|
async def test_resolve_missing_repo(self, db_session) -> None:
|
||||||
|
"""Test that missing repo returns empty list."""
|
||||||
|
git_mount = {
|
||||||
|
"repo_id": "12345678-1234-1234-1234-123456789abc",
|
||||||
|
"source_path": ".",
|
||||||
|
"target_path": "/app",
|
||||||
|
}
|
||||||
|
|
||||||
|
result = await _resolve_single_git_mount(db_session, git_mount)
|
||||||
|
assert result == []
|
||||||
|
|
||||||
|
@pytest.mark.asyncio
|
||||||
|
async def test_resolve_missing_target_path(self, db_session) -> None:
|
||||||
|
"""Test that missing target path returns empty list."""
|
||||||
|
git_mount = {
|
||||||
|
"repo_id": "12345678-1234-1234-1234-123456789abc",
|
||||||
|
"source_path": ".",
|
||||||
|
}
|
||||||
|
|
||||||
|
result = await _resolve_single_git_mount(db_session, git_mount)
|
||||||
|
assert result == []
|
||||||
|
|
||||||
|
@pytest.mark.asyncio
|
||||||
|
async def test_resolve_invalid_repo_id(self, db_session) -> None:
|
||||||
|
"""Test that invalid repo_id returns empty list."""
|
||||||
|
git_mount = {
|
||||||
|
"repo_id": "not-a-uuid",
|
||||||
|
"source_path": ".",
|
||||||
|
"target_path": "/app",
|
||||||
|
}
|
||||||
|
|
||||||
|
result = await _resolve_single_git_mount(db_session, git_mount)
|
||||||
|
assert result == []
|
||||||
Reference in New Issue
Block a user