Skip to content
Closed
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
166 changes: 155 additions & 11 deletions src/dda/env/dev/types/linux_container.py
Original file line number Diff line number Diff line change
Expand Up @@ -11,7 +11,7 @@
import msgspec

from dda.env.dev.interface import DeveloperEnvironmentConfig, DeveloperEnvironmentInterface
from dda.utils.fs import cp_r, temp_directory
from dda.utils.fs import Path, cp_r, temp_directory
from dda.utils.git.constants import GitEnvVars

if TYPE_CHECKING:
Expand All @@ -20,7 +20,6 @@
from dda.tools.docker import Docker
from dda.utils.container.model import Mount
from dda.utils.editors.interface import EditorInterface
from dda.utils.fs import Path


class LinuxContainerConfig(DeveloperEnvironmentConfig):
Expand Down Expand Up @@ -187,16 +186,12 @@ def start(self) -> None:
command.extend(("--mount", mount.as_csv()))

if not self.config.clone:
from dda.utils.fs import Path

repos_path = Path.cwd().parent
for repo_spec in self.config.repos:
repo = repo_spec.split("@")[0]
repo_path = repos_path / repo
if not repo_path.is_dir():
self.app.abort(f"Local repository not found: {repo}")

command.extend(("-v", f"{repo_path}:{self.repo_path(repo)}"))
repo_path = self._resolve_repository_path(repo_spec)
# Mount to base repo path (without worktree subdirectory)
mount_dest = f"{self.home_dir}/repos/{repo}"
command.extend(("-v", f"{repo_path}:{mount_dest}"))

for mount_spec in self.config.extra_mount_specs:
command.extend(("--mount", mount_spec))
Expand Down Expand Up @@ -448,7 +443,156 @@ def repo_path(self, repo: str | None) -> str:
if repo is None:
repo = self.default_repo

return f"{self.home_dir}/repos/{repo}"
base_path = f"{self.home_dir}/repos/{repo}"

# Check if we're currently in a worktree - if so, append the worktree subdirectory
worktree_subdir = self._get_worktree_subdirectory()
if worktree_subdir:
return f"{base_path}/{worktree_subdir}"

return base_path

def _get_worktree_subdirectory(self) -> str | None:
"""
If the current directory is a worktree inside the mounted main repository,
returns the relative path from the main repo to the worktree.

Returns:
Relative path like ".worktrees/branch-name" or None if not in a worktree
"""
cwd = Path.cwd()
git_dir = cwd / ".git"

# Not a git worktree if .git is not a file
if not git_dir.is_file():
return None

try:
# Read the .git file to find the main repo
gitdir_content = git_dir.read_text().strip()
if not gitdir_content.startswith("gitdir: "):
return None

# Extract path like: gitdir: /path/to/main/.git/worktrees/branch
worktree_git_path = gitdir_content[8:] # Remove "gitdir: " prefix
# Navigate up to find main repo: /path/to/main/.git/worktrees/branch -> /path/to/main
main_repo_path = Path(worktree_git_path).parent.parent.parent

# Calculate relative path from main repo to current worktree directory
try:
relative_path = cwd.relative_to(main_repo_path)
return str(relative_path)
except ValueError:
# Worktree is outside the main repo - not supported
self.app.abort(
f"Git worktree at {cwd} is outside the main repository at {main_repo_path}.\n"
f"External worktrees are not currently supported. Please run dda commands from:\n"
f" - The main repository directory, or\n"
f" - A worktree inside the main repository (e.g., {main_repo_path}/.worktrees/branch-name)"
)
except Exception: # noqa: BLE001
# If anything fails, assume not a worktree
return None

def _resolve_repository_path(self, repo_spec: str) -> Path:
"""
Resolve the local path for a repository specification.

Tries multiple strategies:
1. Check if current directory is the requested repo (git-aware, handles worktrees)
2. Check parent directory for repo (backward compatible)

Args:
repo_spec: Repository specification (e.g., "datadog-agent" or "datadog-agent@branch")

Returns:
Path to the repository

Raises:
Aborts if repository cannot be found
"""
from dda.utils.fs import Path

repo_name = repo_spec.split("@")[0] # Strip @branch/@tag if present

# Strategy 1: Check if current directory is a git repository matching the repo name
cwd = Path.cwd()
is_match, resolved_path = self._is_matching_repository(cwd, repo_name)
if is_match:
return resolved_path

# Strategy 2: Check parent directory (existing behavior, backward compatible)
parent_repo_path = cwd.parent / repo_name
if parent_repo_path.is_dir():
is_match, resolved_path = self._is_matching_repository(parent_repo_path, repo_name)
if is_match:
return resolved_path
# Fallback: If not a git repo but directory exists, use it for backward compat
return parent_repo_path

self.app.abort(f"Local repository not found: {repo_name}")
return None # abort() never returns, but ruff requires an explicit return

def _is_matching_repository(self, path: Path, expected_repo_name: str) -> tuple[bool, Path]:
"""
Check if the given path is a git repository matching the expected repository name.

Uses git remote URL to determine the repository name, which works for:
- Regular repositories
- Git worktrees (regardless of directory name)
- Nested repository structures

For worktrees, returns the main repository root path instead of the worktree path,
since worktrees need access to the main repository's .git directory to function.

Args:
path: Path to check
expected_repo_name: Expected repository name (e.g., "datadog-agent")

Returns:
Tuple of (is_match, resolved_path) where:
- is_match: True if the path is a git repository matching the expected name
- resolved_path: For worktrees, the main repo root; for regular repos, the path itself
"""
if not path.is_dir():
return False, path

git_dir = path / ".git"
if not git_dir.exists():
return False, path

# Use git to get the repository name from the remote URL
try:
# Change to the target directory temporarily to get its remote
import os

original_cwd = os.getcwd()
try:
os.chdir(path)
remote = self.app.tools.git.get_remote()

if remote.repo != expected_repo_name:
return False, path

# If this is a worktree, find the main repository root
if git_dir.is_file():
# This is a worktree - read the .git file to find the main repo
gitdir_content = git_dir.read_text().strip()
if gitdir_content.startswith("gitdir: "):
# Extract path like: gitdir: /path/to/main/.git/worktrees/branch
worktree_git_path = gitdir_content[8:] # Remove "gitdir: " prefix
# Navigate up to find main repo: /path/to/main/.git/worktrees/branch -> /path/to/main
main_repo_path = Path(worktree_git_path).parent.parent.parent
return True, main_repo_path

# Regular repository
return True, path
finally:
os.chdir(original_cwd)
# Catch all git-related exceptions (subprocess errors, missing git, no remote, etc.)
except Exception: # noqa: BLE001
# Not a git repository or no remote configured
return False, path

def _container_cp(self, source: str, destination: str, *args: Any) -> None:
"""Runs a `cp -r` command inside the context of the container"""
Expand Down
43 changes: 43 additions & 0 deletions tests/env/dev/types/test_linux_container.py
Original file line number Diff line number Diff line change
Expand Up @@ -1415,6 +1415,49 @@ def test_directory_to_existing_file_fails(self, linux_container, test_target_dir
linux_container.export_path(source="/folder1", destination=existing_file)


class TestRepositoryResolution:
"""Tests for git-aware repository resolution logic."""

def test_git_worktree_resolves_to_main_repo(self, app, mocker, temp_dir):
"""Test git worktree resolves to main repository root, not the worktree directory."""
# Create main repository structure
main_repo = temp_dir / "datadog-agent"
main_repo.ensure_dir()
(main_repo / ".git").ensure_dir()

# Create worktree with arbitrary directory name
worktree_dir = temp_dir / "my_feature_branch"
worktree_dir.ensure_dir()
# Write .git file that points to main repo (like real worktrees do)
(worktree_dir / ".git").write_text(f"gitdir: {main_repo}/.git/worktrees/my_feature_branch")

mock_remote = mocker.MagicMock()
mock_remote.repo = "datadog-agent"
mocker.patch.object(app.tools.git, "get_remote", return_value=mock_remote)

container = LinuxContainer(app=app, name="test", instance="test")
with worktree_dir.as_cwd():
# Testing internal repository resolution logic directly
resolved = container._resolve_repository_path("datadog-agent") # noqa: SLF001

# Should resolve to main repo, not worktree, so git commands work in container
assert resolved == main_repo

def test_backward_compatibility_parent_directory(self, app, mocker, temp_dir):
"""Test non-git directory is found via parent/repo_name (backward compatible)."""
repo_dir = temp_dir / "datadog-agent"
repo_dir.ensure_dir()

mocker.patch.object(app.tools.git, "get_remote", side_effect=Exception("Not a git repo"))

container = LinuxContainer(app=app, name="test", instance="test")
with repo_dir.as_cwd():
# Testing internal repository resolution logic directly
resolved = container._resolve_repository_path("datadog-agent") # noqa: SLF001

assert resolved == repo_dir


class TestImportFiles:
"""Basic import operations."""

Expand Down