Skip to content
Open
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
121 changes: 102 additions & 19 deletions src/aggregation/fetcher.py
Original file line number Diff line number Diff line change
Expand Up @@ -11,23 +11,6 @@
from .sphinx_builder import build_sphinx_markdown


def _convert_to_git_pattern(pattern: str) -> str:
"""
Convert Python glob pattern to git sparse-checkout compatible pattern.

Git doesn't support **, so convert to folder prefix.
e.g., "features/**/*.md" -> "features/*"
"""
if "**" in pattern:
parts = pattern.split("/")
for i, part in enumerate(parts):
if "**" in part:
parts[i] = "*"
return "/".join(parts[: i + 1])
return pattern
return pattern


class DocsFetcher:
"""Handles fetching documentation from remote or local repositories."""

Expand Down Expand Up @@ -86,10 +69,29 @@ def _fetch_remote(
cwd=temp_dir,
)

# Enable sparse-checkout in cone-mode
subprocess.run(
["git", "config", "core.sparseCheckout", "true"],
check=True,
capture_output=True,
cwd=temp_dir,
)

# Fetch the ref (full history to support any locked commit)
print(" Fetching repository...")
subprocess.run(
["git", "fetch", "--all"],
["git", "sparse-checkout", "init", "--cone"],
check=True,
capture_output=True,
cwd=temp_dir,
)

# Build sparse-checkout pattern
sparse_patterns = self._build_sparse_patterns(repo)

# Set sparse-checkout patterns
subprocess.run(
["git", "sparse-checkout", "set"] + sparse_patterns,
check=True,
capture_output=True,
cwd=temp_dir,
Expand All @@ -99,11 +101,21 @@ def _fetch_remote(
# - In update-locks mode: always checkout the ref (latest)
# - In normal mode: use locked commit if available, otherwise checkout ref
if self.update_locks or not repo.commit:
fetch_depth_args = ["--depth", "1"]
checkout_ref = repo.ref
else:
fetch_depth_args = ["--filter=blob:none"]
checkout_ref = repo.commit

print(f" Checking out: {checkout_ref}")
print(f" Fetching sparse checkout: {checkout_ref}")
print(f" Patterns: {', '.join(sparse_patterns)}")
print(
f" Strategy: {'shallow' if '--depth' in fetch_depth_args else 'partial'}"
)

fetch_cmd = ["git", "fetch"] + fetch_depth_args + ["origin", checkout_ref]
subprocess.run(fetch_cmd, check=True, capture_output=True, cwd=temp_dir)

subprocess.run(
["git", "checkout", checkout_ref],
check=True,
Expand Down Expand Up @@ -164,6 +176,77 @@ def _fetch_remote(
# Cleanup
shutil.rmtree(temp_dir, ignore_errors=True)

def _build_sparse_patterns(self, repo: RepoConfig) -> list[str]:
"""
Build sparse-checkout patterns for a repo.

Uses cone mode patterns which are directory based and not glob patterns.

Args:
repo: Repository configuration

Returns:
List of directory patterns for sparse-checkout
"""
patterns: list[str] = []

# Always include docs path
if repo.docs_path:
patterns.append(repo.docs_path)

# Add root files if necessary
if repo.root_files:
for file_pattern in repo.root_files:
cone_pattern = self._to_cone_pattern(file_pattern)
if cone_pattern and cone_pattern not in patterns:
patterns.append(cone_pattern)

return patterns

def _to_cone_pattern(self, pattern: str) -> str | None:
"""
Convert a glob pattern to a cone-mode sparse-checkout pattern.

Cone mode doesn't support globs.

Examples:
"README.md" -> "README.md" (file at root)
"features/*/README.md" -> "features" (directory)
"features/foo/info.yaml" -> "features/foo" (specific directory)
"docs/**/*.md" -> "docs" (directory)

Args:
pattern: Glob pattern from root_files

Returns:
Cone-compatible pattern as string
"""
pattern = pattern.rstrip("/")

# If globs, extract directory
if any(char in pattern for char in ["*", "?", "[", "]"]):
parts = pattern.split("/")
for i, part in enumerate(parts):
if any(char in part for char in ["*", "?", "[", "]"]):
# Return path up to the glob part
if i == 0:
# Glob at root level; Manually filter later.
return None
return "/".join(parts[:i])
return pattern

# No globs, check if file or directory
# For cone pattern, include parent directory
if "/" in pattern:
# Path: return deepest directory
parts = pattern.split("/")
# If last part looks like a file with extension, return parent dir
if "." in parts[-1]:
return "/".join(parts[:-1]) if len(parts) > 1 else parts[-1]
return pattern

return pattern

def _fetch_local(
self,
repo: RepoConfig,
Expand Down
158 changes: 158 additions & 0 deletions tests/unit/test_fetcher.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,158 @@
"""Unit tests for fetcher module."""

from pathlib import Path

from aggregation.fetcher import DocsFetcher
from aggregation.models import RepoConfig


class TestSparseCheckoutPatternGeneration:
"""Test sparse-checkout pattern building."""

def test_build_sparse_patterns_simple_path(self):
fetcher = DocsFetcher(Path("."))
repo = RepoConfig(
name="test",
url="https://github.com/test/repo",
docs_path="docs",
ref="main",
)

patterns = fetcher._build_sparse_patterns(repo)

assert patterns == ["docs"]

def test_build_sparse_patterns_with_root_files(self):
fetcher = DocsFetcher(Path("."))
repo = RepoConfig(
name="test",
url="https://github.com/test/repo",
docs_path="docs",
ref="main",
root_files=["README.md", "LICENSE"],
)

patterns = fetcher._build_sparse_patterns(repo)

assert "docs" in patterns
assert "README.md" in patterns
assert "LICENSE" in patterns

def test_build_sparse_patterns_with_glob_patterns(self):
fetcher = DocsFetcher(Path("."))
repo = RepoConfig(
name="test",
url="https://github.com/test/repo",
docs_path="docs",
ref="main",
root_files=["features/*/README.md", "features/*/info.yaml"],
)
patterns = fetcher._build_sparse_patterns(repo)

assert "docs" in patterns
assert "features" in patterns
assert patterns.count("features") == 1

def test_build_sparse_patterns_no_duplicates(self):
fetcher = DocsFetcher(Path("."))
repo = RepoConfig(
name="test",
url="https://github.com/test/repo",
docs_path="docs",
ref="main",
root_files=["docs/extra.md", "docs/subfolder/file.md"],
)

patterns = fetcher._build_sparse_patterns(repo)

assert patterns.count("docs") == 1


class TestConePatternConversion:
def test_to_cone_pattern_simple_file(self):
"""Test conversion of simple filename at root."""
fetcher = DocsFetcher(Path("."))

result = fetcher._to_cone_pattern("README.md")

assert result == "README.md"

def test_to_cone_pattern_directory(self):
"""Test conversion of directory path."""
fetcher = DocsFetcher(Path("."))

result = fetcher._to_cone_pattern("features")

assert result == "features"

def test_to_cone_pattern_file_in_directory(self):
"""Test conversion of file path."""
fetcher = DocsFetcher(Path("."))

result = fetcher._to_cone_pattern("features/foo/info.yaml")

assert result == "features/foo"

def test_to_cone_pattern_glob_wildcard(self):
"""Test conversion of glob pattern with wildcard."""
fetcher = DocsFetcher(Path("."))

result = fetcher._to_cone_pattern("features/*/README.md")

assert result == "features"

def test_to_cone_pattern_double_wildcard(self):
"""Test conversion of glob pattern with double wildcard."""
fetcher = DocsFetcher(Path("."))

result = fetcher._to_cone_pattern("docs/**/*.md")

assert result == "docs"

def test_to_cone_pattern_root_glob_returns_none(self):
"""Test that glob at root level returns None."""
fetcher = DocsFetcher(Path("."))

result = fetcher._to_cone_pattern("*.md")

assert result is None

def test_to_cone_pattern_question_mark(self):
"""Test conversion of pattern with ? wildcard."""
fetcher = DocsFetcher(Path("."))

result = fetcher._to_cone_pattern("features/foo?/README.md")

assert result == "features"

def test_to_cone_pattern_bracket_glob(self):
"""Test conversion of pattern with bracket glob."""
fetcher = DocsFetcher(Path("."))

result = fetcher._to_cone_pattern("features/[ab]*/README.md")

assert result == "features"

def test_to_cone_pattern_trailing_slash(self):
"""Test that trailing slashes are removed."""
fetcher = DocsFetcher(Path("."))

result = fetcher._to_cone_pattern("features/")

assert result == "features"

def test_to_cone_pattern_nested_path(self):
"""Test conversion of nested directory path."""
fetcher = DocsFetcher(Path("."))

result = fetcher._to_cone_pattern("a/b/c/d")

assert result == "a/b/c/d"

def test_to_cone_pattern_nested_file(self):
"""Test conversion of nested file path."""
fetcher = DocsFetcher(Path("."))

result = fetcher._to_cone_pattern("a/b/c/file.txt")

assert result == "a/b/c"
Loading