578 lines
19 KiB
Python
578 lines
19 KiB
Python
# This is free software for the public good of a permacomputer hosted at
|
|
# permacomputer.com, an always-on computer by the people, for the people.
|
|
# One which is durable, easy to repair, & distributed like tap water
|
|
# for machine learning intelligence.
|
|
#
|
|
# The permacomputer is community-owned infrastructure optimized around
|
|
# four values:
|
|
#
|
|
# TRUTH First principles, math & science, open source code freely distributed
|
|
# FREEDOM Voluntary partnerships, freedom from tyranny & corporate control
|
|
# HARMONY Minimal waste, self-renewing systems with diverse thriving connections
|
|
# LOVE Be yourself without hurting others, cooperation through natural law
|
|
#
|
|
# This software contributes to that vision by archiving the web, preserving digital knowledge before it disappears.
|
|
# Code is seeds to sprout on any abandoned technology.
|
|
|
|
"""
|
|
Tests for repo module.
|
|
|
|
Tests VCS detection, repository cloning, and file walking.
|
|
"""
|
|
|
|
import pytest
|
|
from unittest.mock import Mock, AsyncMock, patch, MagicMock
|
|
import tempfile
|
|
import shutil
|
|
import os
|
|
from pathlib import Path
|
|
|
|
from neopig.repo import (
|
|
detect_vcs,
|
|
get_repo_path,
|
|
run_cmd,
|
|
run_cmd_async,
|
|
clone_repo,
|
|
clone_repo_async,
|
|
pull_repo,
|
|
pull_repo_async,
|
|
get_commit_hash,
|
|
walk_files,
|
|
is_binary_file,
|
|
get_file_language,
|
|
VCS_HOSTS,
|
|
VCS_URL_PATTERNS,
|
|
VCS_DIRS,
|
|
BINARY_EXTENSIONS,
|
|
)
|
|
|
|
|
|
class TestVCSDetection:
|
|
"""Test VCS type detection from URLs."""
|
|
|
|
def test_detect_github(self):
|
|
"""Test detecting GitHub as git."""
|
|
vcs, uri = detect_vcs("https://github.com/user/repo")
|
|
assert vcs == "git"
|
|
assert uri.endswith(".git")
|
|
|
|
def test_detect_github_with_git_suffix(self):
|
|
"""Test GitHub URL already ending in .git."""
|
|
vcs, uri = detect_vcs("https://github.com/user/repo.git")
|
|
assert vcs == "git"
|
|
assert uri == "https://github.com/user/repo.git"
|
|
|
|
def test_detect_gitlab(self):
|
|
"""Test detecting GitLab as git."""
|
|
vcs, uri = detect_vcs("https://gitlab.com/user/repo")
|
|
assert vcs == "git"
|
|
assert uri.endswith(".git")
|
|
|
|
def test_detect_bitbucket(self):
|
|
"""Test detecting Bitbucket as git."""
|
|
vcs, uri = detect_vcs("https://bitbucket.org/user/repo")
|
|
assert vcs == "git"
|
|
|
|
def test_detect_codeberg(self):
|
|
"""Test detecting Codeberg as git."""
|
|
vcs, uri = detect_vcs("https://codeberg.org/user/repo")
|
|
assert vcs == "git"
|
|
|
|
def test_detect_hg_mozilla(self):
|
|
"""Test detecting Mozilla HG as mercurial."""
|
|
vcs, uri = detect_vcs("https://hg.mozilla.org/mozilla-central")
|
|
assert vcs == "hg"
|
|
assert uri == "https://hg.mozilla.org/mozilla-central"
|
|
|
|
def test_detect_hg_python(self):
|
|
"""Test detecting Python HG as mercurial."""
|
|
vcs, uri = detect_vcs("https://hg.python.org/cpython")
|
|
assert vcs == "hg"
|
|
|
|
def test_detect_svn_apache(self):
|
|
"""Test detecting Apache SVN."""
|
|
vcs, uri = detect_vcs("https://svn.apache.org/repos/asf/project")
|
|
assert vcs == "svn"
|
|
|
|
def test_detect_ssh_git_url(self):
|
|
"""Test detecting SSH git URL."""
|
|
vcs, uri = detect_vcs("git@github.com:user/repo.git")
|
|
assert vcs == "git"
|
|
assert uri == "git@github.com:user/repo.git"
|
|
|
|
def test_detect_ssh_hg_url(self):
|
|
"""Test detecting SSH hg URL."""
|
|
vcs, uri = detect_vcs("hg@bitbucket.org:user/repo")
|
|
assert vcs == "hg"
|
|
|
|
def test_detect_fossil_extension(self):
|
|
"""Test detecting fossil from .fossil extension."""
|
|
vcs, uri = detect_vcs("https://example.com/project.fossil")
|
|
assert vcs == "fossil"
|
|
|
|
def test_detect_svn_trunk_pattern(self):
|
|
"""Test detecting SVN from /trunk pattern."""
|
|
vcs, uri = detect_vcs("https://svn.example.com/project/trunk")
|
|
assert vcs == "svn"
|
|
|
|
def test_detect_svn_branches_pattern(self):
|
|
"""Test detecting SVN from /branches pattern."""
|
|
vcs, uri = detect_vcs("https://svn.example.com/project/branches/feature")
|
|
assert vcs == "svn"
|
|
|
|
def test_detect_unknown_url(self):
|
|
"""Test non-VCS URL returns None."""
|
|
vcs, uri = detect_vcs("https://example.com/page.html")
|
|
assert vcs is None
|
|
assert uri is None
|
|
|
|
def test_detect_non_repo_url(self):
|
|
"""Test regular website URL."""
|
|
vcs, uri = detect_vcs("https://google.com")
|
|
assert vcs is None
|
|
|
|
|
|
class TestGetRepoPath:
|
|
"""Test repository path generation."""
|
|
|
|
def test_github_repo_path(self):
|
|
"""Test path for GitHub repo."""
|
|
path = get_repo_path("https://github.com/user/repo", Path("vault"))
|
|
assert path == Path("vault/github.com/user/repo")
|
|
|
|
def test_github_repo_path_with_git_suffix(self):
|
|
"""Test path strips .git suffix."""
|
|
path = get_repo_path("https://github.com/user/repo.git", Path("vault"))
|
|
assert path == Path("vault/github.com/user/repo")
|
|
|
|
def test_nested_repo_path(self):
|
|
"""Test path for nested repo structure."""
|
|
path = get_repo_path("https://github.com/org/sub/repo", Path("vault"))
|
|
assert path == Path("vault/github.com/org/sub/repo")
|
|
|
|
def test_ssh_repo_path(self):
|
|
"""Test path for SSH URL."""
|
|
path = get_repo_path("git@github.com:user/repo.git", Path("vault"))
|
|
assert path == Path("vault/github.com/user/repo")
|
|
|
|
def test_repo_path_with_port(self):
|
|
"""Test path strips port from host."""
|
|
path = get_repo_path("https://git.example.com:8443/user/repo", Path("vault"))
|
|
assert "8443" not in str(path) or "git.example.com" in str(path)
|
|
|
|
|
|
class TestRunCmd:
|
|
"""Test command execution."""
|
|
|
|
def test_run_cmd_success(self):
|
|
"""Test successful command execution."""
|
|
ret, stdout, stderr = run_cmd(["echo", "hello"])
|
|
assert ret == 0
|
|
assert "hello" in stdout
|
|
|
|
def test_run_cmd_failure(self):
|
|
"""Test failed command execution."""
|
|
ret, stdout, stderr = run_cmd(["false"])
|
|
assert ret != 0
|
|
|
|
def test_run_cmd_nonexistent(self):
|
|
"""Test nonexistent command."""
|
|
ret, stdout, stderr = run_cmd(["nonexistent_command_xyz"])
|
|
assert ret != 0
|
|
|
|
def test_run_cmd_timeout(self):
|
|
"""Test command timeout."""
|
|
# This should timeout quickly
|
|
ret, stdout, stderr = run_cmd(["sleep", "10"], timeout=1)
|
|
assert ret == -1
|
|
assert "timed out" in stderr.lower() or "timeout" in stderr.lower()
|
|
|
|
|
|
class TestRunCmdAsync:
|
|
"""Test async command execution."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_run_cmd_async_success(self):
|
|
"""Test async successful command."""
|
|
ret, stdout, stderr = await run_cmd_async(["echo", "hello"])
|
|
assert ret == 0
|
|
assert "hello" in stdout
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_run_cmd_async_failure(self):
|
|
"""Test async failed command."""
|
|
ret, stdout, stderr = await run_cmd_async(["false"])
|
|
assert ret != 0
|
|
|
|
|
|
class TestCloneRepo:
|
|
"""Test repository cloning."""
|
|
|
|
def setup_method(self):
|
|
self.temp_dir = tempfile.mkdtemp()
|
|
|
|
def teardown_method(self):
|
|
if os.path.exists(self.temp_dir):
|
|
shutil.rmtree(self.temp_dir)
|
|
|
|
def test_clone_repo_unknown_vcs(self):
|
|
"""Test cloning with unknown VCS type."""
|
|
dest = Path(self.temp_dir) / "unknown"
|
|
success, msg = clone_repo("https://example.com/repo", dest, "unknown")
|
|
assert success is False
|
|
assert "Unknown VCS type" in msg
|
|
|
|
@patch('neopig.repo.run_cmd')
|
|
def test_clone_git_shallow(self, mock_run):
|
|
"""Test git clone with shallow flag."""
|
|
mock_run.return_value = (0, "", "")
|
|
dest = Path(self.temp_dir) / "repo"
|
|
|
|
success, msg = clone_repo("https://github.com/user/repo.git", dest, "git", shallow=True)
|
|
|
|
mock_run.assert_called_once()
|
|
call_args = mock_run.call_args[0][0]
|
|
assert "--depth" in call_args
|
|
assert "1" in call_args
|
|
|
|
@patch('neopig.repo.run_cmd')
|
|
def test_clone_git_full(self, mock_run):
|
|
"""Test git clone without shallow flag."""
|
|
mock_run.return_value = (0, "", "")
|
|
dest = Path(self.temp_dir) / "repo"
|
|
|
|
success, msg = clone_repo("https://github.com/user/repo.git", dest, "git", shallow=False)
|
|
|
|
mock_run.assert_called_once()
|
|
call_args = mock_run.call_args[0][0]
|
|
assert "--depth" not in call_args
|
|
|
|
@patch('neopig.repo.run_cmd')
|
|
def test_clone_hg(self, mock_run):
|
|
"""Test mercurial clone."""
|
|
mock_run.return_value = (0, "", "")
|
|
dest = Path(self.temp_dir) / "repo"
|
|
|
|
success, msg = clone_repo("https://hg.example.com/repo", dest, "hg")
|
|
|
|
mock_run.assert_called_once()
|
|
call_args = mock_run.call_args[0][0]
|
|
assert "hg" in call_args
|
|
assert "clone" in call_args
|
|
|
|
@patch('neopig.repo.run_cmd')
|
|
def test_clone_svn(self, mock_run):
|
|
"""Test SVN checkout."""
|
|
mock_run.return_value = (0, "", "")
|
|
dest = Path(self.temp_dir) / "repo"
|
|
|
|
success, msg = clone_repo("https://svn.example.com/repo", dest, "svn")
|
|
|
|
mock_run.assert_called_once()
|
|
call_args = mock_run.call_args[0][0]
|
|
assert "svn" in call_args
|
|
assert "checkout" in call_args
|
|
|
|
|
|
class TestCloneRepoAsync:
|
|
"""Test async repository cloning."""
|
|
|
|
@pytest.mark.asyncio
|
|
@patch('neopig.repo.clone_repo')
|
|
async def test_clone_repo_async(self, mock_clone):
|
|
"""Test async clone delegates to sync."""
|
|
mock_clone.return_value = (True, "Cloned")
|
|
|
|
success, msg = await clone_repo_async(
|
|
"https://github.com/user/repo.git",
|
|
Path("/tmp/repo"),
|
|
"git"
|
|
)
|
|
|
|
assert success is True
|
|
mock_clone.assert_called_once()
|
|
|
|
|
|
class TestPullRepo:
|
|
"""Test repository pulling/updating."""
|
|
|
|
def setup_method(self):
|
|
self.temp_dir = tempfile.mkdtemp()
|
|
self.repo_path = Path(self.temp_dir) / "repo"
|
|
self.repo_path.mkdir()
|
|
|
|
def teardown_method(self):
|
|
if os.path.exists(self.temp_dir):
|
|
shutil.rmtree(self.temp_dir)
|
|
|
|
def test_pull_no_vcs(self):
|
|
"""Test pull on directory without VCS."""
|
|
success, msg = pull_repo(self.repo_path)
|
|
assert success is False
|
|
assert "No VCS directory found" in msg
|
|
|
|
@patch('neopig.repo.run_cmd')
|
|
def test_pull_git(self, mock_run):
|
|
"""Test git pull."""
|
|
mock_run.return_value = (0, "Already up to date", "")
|
|
(self.repo_path / ".git").mkdir()
|
|
|
|
success, msg = pull_repo(self.repo_path)
|
|
|
|
assert success is True
|
|
mock_run.assert_called_once()
|
|
call_args = mock_run.call_args[0][0]
|
|
assert "git" in call_args
|
|
assert "pull" in call_args
|
|
|
|
@patch('neopig.repo.run_cmd')
|
|
def test_pull_hg(self, mock_run):
|
|
"""Test mercurial pull."""
|
|
mock_run.return_value = (0, "pulling from...", "")
|
|
(self.repo_path / ".hg").mkdir()
|
|
|
|
success, msg = pull_repo(self.repo_path)
|
|
|
|
assert success is True
|
|
call_args = mock_run.call_args[0][0]
|
|
assert "hg" in call_args
|
|
assert "pull" in call_args
|
|
|
|
@patch('neopig.repo.run_cmd')
|
|
def test_pull_svn(self, mock_run):
|
|
"""Test SVN update."""
|
|
mock_run.return_value = (0, "Updating...", "")
|
|
(self.repo_path / ".svn").mkdir()
|
|
|
|
success, msg = pull_repo(self.repo_path)
|
|
|
|
assert success is True
|
|
call_args = mock_run.call_args[0][0]
|
|
assert "svn" in call_args
|
|
assert "update" in call_args
|
|
|
|
|
|
class TestGetCommitHash:
|
|
"""Test getting commit/revision hash."""
|
|
|
|
def setup_method(self):
|
|
self.temp_dir = tempfile.mkdtemp()
|
|
self.repo_path = Path(self.temp_dir) / "repo"
|
|
self.repo_path.mkdir()
|
|
|
|
def teardown_method(self):
|
|
if os.path.exists(self.temp_dir):
|
|
shutil.rmtree(self.temp_dir)
|
|
|
|
def test_get_hash_no_vcs(self):
|
|
"""Test getting hash without VCS."""
|
|
result = get_commit_hash(self.repo_path)
|
|
assert result is None
|
|
|
|
@patch('neopig.repo.run_cmd')
|
|
def test_get_git_hash(self, mock_run):
|
|
"""Test getting git commit hash."""
|
|
mock_run.return_value = (0, "abc123def456\n", "")
|
|
(self.repo_path / ".git").mkdir()
|
|
|
|
result = get_commit_hash(self.repo_path)
|
|
|
|
assert result == "abc123def456"
|
|
|
|
@patch('neopig.repo.run_cmd')
|
|
def test_get_hg_hash(self, mock_run):
|
|
"""Test getting mercurial changeset hash."""
|
|
mock_run.return_value = (0, "abc123+\n", "")
|
|
(self.repo_path / ".hg").mkdir()
|
|
|
|
result = get_commit_hash(self.repo_path)
|
|
|
|
assert result == "abc123+"
|
|
|
|
|
|
class TestWalkFiles:
|
|
"""Test walking files in a repository."""
|
|
|
|
def setup_method(self):
|
|
self.temp_dir = tempfile.mkdtemp()
|
|
self.repo_path = Path(self.temp_dir) / "repo"
|
|
self.repo_path.mkdir()
|
|
|
|
def teardown_method(self):
|
|
if os.path.exists(self.temp_dir):
|
|
shutil.rmtree(self.temp_dir)
|
|
|
|
def test_walk_empty_repo(self):
|
|
"""Test walking empty directory."""
|
|
files = list(walk_files(self.repo_path))
|
|
assert files == []
|
|
|
|
def test_walk_with_files(self):
|
|
"""Test walking directory with files."""
|
|
# Create some files
|
|
(self.repo_path / "main.py").write_text("print('hello')")
|
|
(self.repo_path / "README.md").write_text("# Test")
|
|
|
|
files = list(walk_files(self.repo_path))
|
|
|
|
assert len(files) == 2
|
|
names = [f.name for f in files]
|
|
assert "main.py" in names
|
|
assert "README.md" in names
|
|
|
|
def test_walk_skips_vcs_dirs(self):
|
|
"""Test that VCS directories are skipped."""
|
|
(self.repo_path / ".git").mkdir()
|
|
(self.repo_path / ".git" / "config").write_text("[core]")
|
|
(self.repo_path / "main.py").write_text("print('hello')")
|
|
|
|
files = list(walk_files(self.repo_path))
|
|
|
|
# Should only have main.py, not .git/config
|
|
assert len(files) == 1
|
|
assert files[0].name == "main.py"
|
|
|
|
def test_walk_skips_binary_by_default(self):
|
|
"""Test that binary files are skipped by default."""
|
|
(self.repo_path / "image.png").write_bytes(b"\x89PNG\r\n\x1a\n")
|
|
(self.repo_path / "main.py").write_text("print('hello')")
|
|
|
|
files = list(walk_files(self.repo_path, include_binary=False))
|
|
|
|
names = [f.name for f in files]
|
|
assert "main.py" in names
|
|
assert "image.png" not in names
|
|
|
|
def test_walk_includes_binary_when_requested(self):
|
|
"""Test that binary files are included when requested."""
|
|
(self.repo_path / "image.png").write_bytes(b"\x89PNG\r\n\x1a\n")
|
|
(self.repo_path / "main.py").write_text("print('hello')")
|
|
|
|
files = list(walk_files(self.repo_path, include_binary=True))
|
|
|
|
names = [f.name for f in files]
|
|
assert "main.py" in names
|
|
assert "image.png" in names
|
|
|
|
def test_walk_skips_empty_files(self):
|
|
"""Test that empty files are skipped."""
|
|
(self.repo_path / "empty.txt").write_text("")
|
|
(self.repo_path / "main.py").write_text("print('hello')")
|
|
|
|
files = list(walk_files(self.repo_path))
|
|
|
|
names = [f.name for f in files]
|
|
assert "main.py" in names
|
|
assert "empty.txt" not in names
|
|
|
|
def test_walk_nested_directories(self):
|
|
"""Test walking nested directories."""
|
|
(self.repo_path / "src").mkdir()
|
|
(self.repo_path / "src" / "app.py").write_text("app code")
|
|
(self.repo_path / "tests").mkdir()
|
|
(self.repo_path / "tests" / "test_app.py").write_text("test code")
|
|
|
|
files = list(walk_files(self.repo_path))
|
|
|
|
names = [f.name for f in files]
|
|
assert "app.py" in names
|
|
assert "test_app.py" in names
|
|
|
|
|
|
class TestIsBinaryFile:
|
|
"""Test binary file detection."""
|
|
|
|
def setup_method(self):
|
|
self.temp_dir = tempfile.mkdtemp()
|
|
|
|
def teardown_method(self):
|
|
if os.path.exists(self.temp_dir):
|
|
shutil.rmtree(self.temp_dir)
|
|
|
|
def test_binary_by_extension(self):
|
|
"""Test detecting binary by extension."""
|
|
path = Path(self.temp_dir) / "image.png"
|
|
path.write_bytes(b"fake png")
|
|
|
|
assert is_binary_file(path) is True
|
|
|
|
def test_binary_by_content(self):
|
|
"""Test detecting binary by content (null bytes)."""
|
|
path = Path(self.temp_dir) / "data.bin"
|
|
path.write_bytes(b"hello\x00world")
|
|
|
|
assert is_binary_file(path) is True
|
|
|
|
def test_text_file(self):
|
|
"""Test detecting text file."""
|
|
path = Path(self.temp_dir) / "code.py"
|
|
path.write_text("print('hello')")
|
|
|
|
assert is_binary_file(path) is False
|
|
|
|
def test_binary_extensions_set(self):
|
|
"""Test that binary extensions are defined."""
|
|
assert ".png" in BINARY_EXTENSIONS
|
|
assert ".jpg" in BINARY_EXTENSIONS
|
|
assert ".exe" in BINARY_EXTENSIONS
|
|
assert ".zip" in BINARY_EXTENSIONS
|
|
|
|
|
|
class TestGetFileLanguage:
|
|
"""Test programming language detection."""
|
|
|
|
def test_python_extension(self):
|
|
"""Test detecting Python."""
|
|
assert get_file_language(Path("script.py")) == "python"
|
|
|
|
def test_javascript_extension(self):
|
|
"""Test detecting JavaScript."""
|
|
assert get_file_language(Path("app.js")) == "javascript"
|
|
|
|
def test_typescript_extension(self):
|
|
"""Test detecting TypeScript."""
|
|
assert get_file_language(Path("app.ts")) == "typescript"
|
|
|
|
def test_jsx_extension(self):
|
|
"""Test detecting JSX."""
|
|
assert get_file_language(Path("component.jsx")) == "javascript"
|
|
|
|
def test_tsx_extension(self):
|
|
"""Test detecting TSX."""
|
|
assert get_file_language(Path("component.tsx")) == "typescript"
|
|
|
|
def test_ruby_extension(self):
|
|
"""Test detecting Ruby."""
|
|
assert get_file_language(Path("app.rb")) == "ruby"
|
|
|
|
def test_unknown_extension(self):
|
|
"""Test unknown extension returns None."""
|
|
result = get_file_language(Path("file.xyz"))
|
|
assert result is None or result == ""
|
|
|
|
|
|
class TestVCSConstants:
|
|
"""Test VCS-related constants."""
|
|
|
|
def test_vcs_hosts_contains_major_hosts(self):
|
|
"""Test VCS hosts map contains major hosts."""
|
|
assert "github.com" in VCS_HOSTS
|
|
assert "gitlab.com" in VCS_HOSTS
|
|
assert "bitbucket.org" in VCS_HOSTS
|
|
|
|
def test_vcs_dirs_contains_standard_dirs(self):
|
|
"""Test VCS dirs contains standard directories."""
|
|
assert ".git" in VCS_DIRS
|
|
assert ".hg" in VCS_DIRS
|
|
assert ".svn" in VCS_DIRS
|
|
|
|
def test_vcs_url_patterns_is_list(self):
|
|
"""Test VCS URL patterns is a list of tuples."""
|
|
assert isinstance(VCS_URL_PATTERNS, list)
|
|
for pattern, vcs_type in VCS_URL_PATTERNS:
|
|
assert isinstance(pattern, str)
|
|
assert isinstance(vcs_type, str)
|
|
|
|
|
|
if __name__ == '__main__':
|
|
pytest.main([__file__, '-v'])
|