import os
import re
import subprocess
import tempfile
import textwrap
import git
import pytest
from helpers import (
NonExistentPath,
TempGitFileRemote,
TempGitRepository,
checksum_directory,
grm,
shell,
)
templates = {
"repo_simple": {
"toml": """
[[trees]]
root = "{root}"
[[trees.repos]]
name = "test"
""",
"yaml": """
trees:
- root: "{root}"
repos:
- name: "test"
""",
},
"repo_with_remote": {
"toml": """
[[trees]]
root = "{root}"
[[trees.repos]]
name = "test"
[[trees.repos.remotes]]
name = "{remotename}"
url = "file://{remote}"
type = "file"
""",
"yaml": textwrap.dedent(
"""
trees:
- root: "{root}"
repos:
- name: test
remotes:
- name: "{remotename}"
url: "file://{remote}"
type: "file"
"""
),
},
"repo_with_two_remotes": {
"toml": """
[[trees]]
root = "{root}"
[[trees.repos]]
name = "test"
[[trees.repos.remotes]]
name = "origin"
url = "file://{remote1}"
type = "file"
[[trees.repos.remotes]]
name = "origin2"
url = "file://{remote2}"
type = "file"
""",
"yaml": textwrap.dedent(
"""
trees:
- root: "{root}"
repos:
- name: "test"
remotes:
- name: "origin"
url: "file://{remote1}"
type: "file"
- name: "origin2"
url: "file://{remote2}"
type: "file"
"""
),
},
"worktree_repo_simple": {
"toml": """
[[trees]]
root = "{root}"
[[trees.repos]]
name = "test"
worktree_setup = true
""",
"yaml": textwrap.dedent(
"""
trees:
- root: "{root}"
repos:
- name: test
worktree_setup: true
"""
),
},
"worktree_repo_with_remote": {
"toml": """
[[trees]]
root = "{root}"
[[trees.repos]]
name = "test"
worktree_setup = true
[[trees.repos.remotes]]
name = "origin"
url = "file://{remote}"
type = "file"
""",
"yaml": textwrap.dedent(
"""
trees:
- root: "{root}"
repos:
- name: test
worktree_setup: true
remotes:
- name: origin
url: "file://{remote}"
type: "file"
"""
),
},
"repo_in_subdirectory": {
"toml": """
[[trees]]
root = "{root}"
[[trees.repos]]
name = "outer/inner"
[[trees.repos.remotes]]
name = "origin"
url = "file://{remote}"
type = "file"
""",
"yaml": textwrap.dedent(
"""
trees:
- root: "{root}"
repos:
- name: outer/inner
remotes:
- name: origin
url: "file://{remote}"
type: "file"
"""
),
},
"nested_trees": {
"toml": """
[[trees]]
root = "{root}"
[[trees.repos]]
name = "outer"
[[trees.repos.remotes]]
name = "origin"
url = "file://{remote1}"
type = "file"
[[trees]]
root = "{root}/subdir"
[[trees.repos]]
name = "inner"
[[trees.repos.remotes]]
name = "origin"
url = "file://{remote2}"
type = "file"
""",
"yaml": textwrap.dedent(
"""
trees:
- root: "{root}"
repos:
- name: outer
remotes:
- name: origin
url: "file://{remote1}"
type: "file"
- root: "{root}/subdir"
repos:
- name: inner
remotes:
- name: origin
url: "file://{remote2}"
type: "file"
"""
),
},
}
@pytest.mark.parametrize("configtype", ["toml", "yaml"])
def test_repos_sync_config_is_valid_symlink(configtype):
with tempfile.TemporaryDirectory() as target:
with TempGitFileRemote() as (remote, head_commit_sha):
with tempfile.NamedTemporaryFile() as config:
with tempfile.TemporaryDirectory() as config_dir:
config_symlink = os.path.join(config_dir, "cfglink")
os.symlink(config.name, config_symlink)
with open(config.name, "w") as f:
f.write(
templates["repo_with_remote"][configtype].format(
root=target, remote=remote, remotename="origin"
)
)
subprocess.run(["cat", config.name])
cmd = grm(["repos", "sync", "config", "--config", config_symlink])
assert cmd.returncode == 0
git_dir = os.path.join(target, "test")
assert os.path.exists(git_dir)
with git.Repo(git_dir) as repo:
assert not repo.bare
assert not repo.is_dirty()
assert set([str(r) for r in repo.remotes]) == {"origin"}
assert str(repo.active_branch) == "master"
assert str(repo.head.commit) == head_commit_sha
def test_repos_sync_config_is_invalid_symlink():
with tempfile.TemporaryDirectory() as target:
with tempfile.TemporaryDirectory() as config_dir:
with NonExistentPath() as nonexistent_dir:
config_symlink = os.path.join(config_dir, "cfglink")
os.symlink(nonexistent_dir, config_symlink)
cmd = grm(["repos", "sync", "config", "--config", config_symlink])
assert cmd.returncode != 0
assert len(cmd.stdout) == 0
assert "not found" in cmd.stderr.lower()
assert not os.path.exists(os.path.join(target, "test"))
assert not os.path.exists(os.path.join(target, "test"))
def test_repos_sync_config_is_directory():
with tempfile.TemporaryDirectory() as config:
cmd = grm(["repos", "sync", "config", "--config", config])
assert cmd.returncode != 0
assert len(cmd.stdout) == 0
assert "is a directory" in cmd.stderr.lower()
def test_repos_sync_config_is_unreadable():
with tempfile.TemporaryDirectory() as config_dir:
config_path = os.path.join(config_dir, "cfg")
open(config_path, "w")
os.chmod(config_path, 0o0000)
cmd = grm(["repos", "sync", "config", "--config", config_path])
assert os.path.exists(config_path)
assert cmd.returncode != 0
assert len(cmd.stdout) == 0
@pytest.mark.parametrize("configtype", ["toml", "yaml"])
def test_repos_sync_unmanaged_repos(configtype):
with tempfile.TemporaryDirectory() as root:
with TempGitRepository(dir=root) as unmanaged_repo:
with tempfile.NamedTemporaryFile() as config:
with open(config.name, "w") as f:
f.write(templates["repo_simple"][configtype].format(root=root))
cmd = grm(["repos", "sync", "config", "--config", config.name])
assert cmd.returncode == 0
git_dir = os.path.join(root, "test")
assert os.path.exists(git_dir)
unmanaged_repo_name = os.path.relpath(unmanaged_repo, root)
regex = f".*unmanaged.*{unmanaged_repo_name}"
assert any(
[re.match(regex, line) for line in cmd.stderr.lower().split("\n")]
)
@pytest.mark.parametrize("configtype", ["toml", "yaml"])
def test_repos_sync_root_is_file(configtype):
with tempfile.NamedTemporaryFile() as target:
with tempfile.NamedTemporaryFile() as config:
with open(config.name, "w") as f:
f.write(templates["repo_simple"][configtype].format(root=target.name))
cmd = grm(["repos", "sync", "config", "--config", config.name])
assert cmd.returncode != 0
assert "notadirectory" in cmd.stderr.lower()
@pytest.mark.parametrize("configtype", ["toml", "yaml"])
def test_repos_sync_normal_clone(configtype):
with tempfile.TemporaryDirectory() as target:
with TempGitFileRemote() as (remote1, remote1_head_commit_sha):
with TempGitFileRemote() as (remote2, remote2_head_commit_sha):
with tempfile.NamedTemporaryFile() as config:
with open(config.name, "w") as f:
f.write(
templates["repo_with_two_remotes"][configtype].format(
root=target, remote1=remote1, remote2=remote2
)
)
cmd = grm(["repos", "sync", "config", "--config", config.name])
assert cmd.returncode == 0
git_dir = os.path.join(target, "test")
assert os.path.exists(git_dir)
with git.Repo(git_dir) as repo:
assert not repo.bare
assert not repo.is_dirty()
assert set([str(r) for r in repo.remotes]) == {
"origin",
"origin2",
}
assert str(repo.active_branch) == "master"
assert str(repo.head.commit) == remote1_head_commit_sha
assert len(repo.remotes) == 2
urls = list(repo.remote("origin").urls)
assert len(urls) == 1
assert urls[0] == f"file://{remote1}"
urls = list(repo.remote("origin2").urls)
assert len(urls) == 1
assert urls[0] == f"file://{remote2}"
@pytest.mark.parametrize("configtype", ["toml", "yaml"])
def test_repos_sync_repo_in_subdirectory(configtype):
with tempfile.TemporaryDirectory() as target:
with TempGitFileRemote() as (remote, remote_head_commit_sha):
with tempfile.NamedTemporaryFile() as config:
with open(config.name, "w") as f:
f.write(
templates["repo_in_subdirectory"][configtype].format(
root=target, remote=remote
)
)
cmd = grm(["repos", "sync", "config", "--config", config.name])
assert cmd.returncode == 0
git_dir = os.path.join(target, "outer", "inner")
assert os.path.exists(git_dir)
with git.Repo(git_dir) as repo:
assert not repo.bare
assert not repo.is_dirty()
assert set([str(r) for r in repo.remotes]) == {"origin"}
assert str(repo.active_branch) == "master"
assert str(repo.head.commit) == remote_head_commit_sha
assert len(repo.remotes) == 1
urls = list(repo.remote("origin").urls)
assert len(urls) == 1
assert urls[0] == f"file://{remote}"
cmd = grm(["repos", "sync", "config", "--config", config.name])
assert "found unmanaged repository" not in cmd.stderr.lower()
@pytest.mark.parametrize("configtype", ["toml", "yaml"])
def test_repos_sync_nested_clone(configtype):
with tempfile.TemporaryDirectory() as target:
with TempGitFileRemote() as (remote1, remote1_head_commit_sha):
with TempGitFileRemote() as (remote2, remote2_head_commit_sha):
with tempfile.NamedTemporaryFile() as config:
with open(config.name, "w") as f:
f.write(
templates["nested_trees"][configtype].format(
root=target, remote1=remote1, remote2=remote2
)
)
cmd = grm(["repos", "sync", "config", "--config", config.name])
assert cmd.returncode == 0
def validate(git_dir, sha, remote):
assert os.path.exists(git_dir)
with git.Repo(git_dir) as repo:
assert not repo.bare
assert not repo.is_dirty()
assert set([str(r) for r in repo.remotes]) == {"origin"}
assert str(repo.active_branch) == "master"
assert str(repo.head.commit) == sha
assert len(repo.remotes) == 1
urls = list(repo.remote("origin").urls)
assert len(urls) == 1
assert urls[0] == f"file://{remote}"
validate(
os.path.join(target, "outer"), remote1_head_commit_sha, remote1
)
validate(
os.path.join(target, "subdir", "inner"),
remote2_head_commit_sha,
remote2,
)
cmd = grm(["repos", "sync", "config", "--config", config.name])
print(cmd.stdout)
print(cmd.stderr)
assert "found unmanaged repository" not in cmd.stderr.lower()
@pytest.mark.parametrize("configtype", ["toml", "yaml"])
def test_repos_sync_normal_init(configtype):
with tempfile.TemporaryDirectory() as target:
with tempfile.NamedTemporaryFile() as config:
with open(config.name, "w") as f:
f.write(templates["repo_simple"][configtype].format(root=target))
cmd = grm(["repos", "sync", "config", "--config", config.name])
assert cmd.returncode == 0
git_dir = os.path.join(target, "test")
assert os.path.exists(git_dir)
with git.Repo(git_dir) as repo:
assert not repo.bare
assert not repo.is_dirty()
assert not repo.head.is_valid()
@pytest.mark.parametrize("configtype", ["toml", "yaml"])
def test_repos_sync_normal_add_remote(configtype):
with tempfile.TemporaryDirectory() as target:
with TempGitFileRemote() as (remote1, remote1_head_commit_sha):
with TempGitFileRemote() as (remote2, remote2_head_commit_sha):
with tempfile.NamedTemporaryFile() as config:
with open(config.name, "w") as f:
f.write(
templates["repo_with_remote"][configtype].format(
root=target, remote=remote1, remotename="origin"
)
)
cmd = grm(["repos", "sync", "config", "--config", config.name])
assert cmd.returncode == 0
git_dir = os.path.join(target, "test")
assert os.path.exists(git_dir)
with git.Repo(git_dir) as repo:
assert not repo.bare
assert not repo.is_dirty()
assert set([str(r) for r in repo.remotes]) == {"origin"}
assert str(repo.active_branch) == "master"
assert str(repo.head.commit) == remote1_head_commit_sha
with open(config.name, "w") as f:
f.write(
templates["repo_with_two_remotes"][configtype].format(
root=target, remote1=remote1, remote2=remote2
)
)
cmd = grm(["repos", "sync", "config", "--config", config.name])
assert cmd.returncode == 0
with git.Repo(git_dir) as repo:
assert set([str(r) for r in repo.remotes]) == {
"origin",
"origin2",
}
urls = list(repo.remote("origin").urls)
assert len(urls) == 1
assert urls[0] == f"file://{remote1}"
urls = list(repo.remote("origin2").urls)
assert len(urls) == 1
assert urls[0] == f"file://{remote2}"
@pytest.mark.parametrize("configtype", ["toml", "yaml"])
def test_repos_sync_normal_remove_remote(configtype):
with tempfile.TemporaryDirectory() as target:
with TempGitFileRemote() as (remote1, remote1_head_commit_sha):
with TempGitFileRemote() as (remote2, remote2_head_commit_sha):
with tempfile.NamedTemporaryFile() as config:
with open(config.name, "w") as f:
f.write(
templates["repo_with_two_remotes"][configtype].format(
root=target, remote1=remote1, remote2=remote2
)
)
cmd = grm(["repos", "sync", "config", "--config", config.name])
assert cmd.returncode == 0
git_dir = os.path.join(target, "test")
assert os.path.exists(git_dir)
with git.Repo(git_dir) as repo:
assert not repo.bare
assert not repo.is_dirty()
assert set([str(r) for r in repo.remotes]) == {
"origin",
"origin2",
}
assert str(repo.active_branch) == "master"
assert str(repo.head.commit) == remote1_head_commit_sha
with open(config.name, "w") as f:
f.write(
templates["repo_with_remote"][configtype].format(
root=target, remote=remote2, remotename="origin2"
)
)
cmd = grm(["repos", "sync", "config", "--config", config.name])
assert cmd.returncode == 0
shell(f"cd {git_dir} && git remote -v")
with git.Repo(git_dir) as repo:
with pytest.raises(git.exc.GitCommandError):
list(repo.remote("origin").urls)
urls = list(repo.remote("origin2").urls)
assert len(urls) == 1
assert urls[0] == f"file://{remote2}"
@pytest.mark.parametrize("configtype", ["toml", "yaml"])
def test_repos_sync_normal_change_remote_url(configtype):
with tempfile.TemporaryDirectory() as target:
with TempGitFileRemote() as (remote1, remote1_head_commit_sha):
with TempGitFileRemote() as (remote2, remote2_head_commit_sha):
with tempfile.NamedTemporaryFile() as config:
with open(config.name, "w") as f:
f.write(
templates["repo_with_remote"][configtype].format(
root=target, remote=remote1, remotename="origin"
)
)
cmd = grm(["repos", "sync", "config", "--config", config.name])
assert cmd.returncode == 0
git_dir = os.path.join(target, "test")
assert os.path.exists(git_dir)
with git.Repo(git_dir) as repo:
assert not repo.bare
assert not repo.is_dirty()
assert set([str(r) for r in repo.remotes]) == {"origin"}
assert str(repo.active_branch) == "master"
assert str(repo.head.commit) == remote1_head_commit_sha
with open(config.name, "w") as f:
f.write(
templates["repo_with_remote"][configtype].format(
root=target, remote=remote2, remotename="origin"
)
)
cmd = grm(["repos", "sync", "config", "--config", config.name])
assert cmd.returncode == 0
with git.Repo(git_dir) as repo:
assert set([str(r) for r in repo.remotes]) == {"origin"}
urls = list(repo.remote("origin").urls)
assert len(urls) == 1
assert urls[0] == f"file://{remote2}"
@pytest.mark.parametrize("configtype", ["toml", "yaml"])
def test_repos_sync_normal_change_remote_name(configtype):
with tempfile.TemporaryDirectory() as target:
with TempGitFileRemote() as (remote1, remote1_head_commit_sha):
with TempGitFileRemote() as (remote2, remote2_head_commit_sha):
with tempfile.NamedTemporaryFile() as config:
with open(config.name, "w") as f:
f.write(
templates["repo_with_remote"][configtype].format(
root=target, remote=remote1, remotename="origin"
)
)
cmd = grm(["repos", "sync", "config", "--config", config.name])
assert cmd.returncode == 0
git_dir = os.path.join(target, "test")
assert os.path.exists(git_dir)
with git.Repo(git_dir) as repo:
assert not repo.bare
assert not repo.is_dirty()
assert set([str(r) for r in repo.remotes]) == {"origin"}
assert str(repo.active_branch) == "master"
assert str(repo.head.commit) == remote1_head_commit_sha
with open(config.name, "w") as f:
f.write(
templates["repo_with_remote"][configtype].format(
root=target, remote=remote1, remotename="origin2"
)
)
cmd = grm(["repos", "sync", "config", "--config", config.name])
assert cmd.returncode == 0
with git.Repo(git_dir) as repo:
with pytest.raises(git.exc.GitCommandError):
list(repo.remote("origin").urls)
urls = list(repo.remote("origin2").urls)
assert len(urls) == 1
assert urls[0] == f"file://{remote1}"
@pytest.mark.parametrize("configtype", ["toml", "yaml"])
@pytest.mark.parametrize("init_worktree", [True, False, "default"])
def test_repos_sync_worktree_clone(configtype, init_worktree):
with tempfile.TemporaryDirectory() as target:
with TempGitFileRemote() as (remote, head_commit_sha):
with tempfile.NamedTemporaryFile() as config:
with open(config.name, "w") as f:
f.write(
templates["worktree_repo_with_remote"][configtype].format(
root=target, remote=remote, remotename="origin"
)
)
args = ["repos", "sync", "config", "--config", config.name]
if init_worktree is True:
args.append("--init-worktree=true")
if init_worktree is False:
args.append("--init-worktree=false")
for i in [1, 2]:
cmd = grm(args)
assert cmd.returncode == 0
worktree_dir = f"{target}/test"
assert os.path.exists(worktree_dir)
if init_worktree is True or init_worktree == "default":
assert set(os.listdir(worktree_dir)) == {
".git-main-working-tree",
"master",
}
else:
assert set(os.listdir(worktree_dir)) == {
".git-main-working-tree"
}
with git.Repo(
os.path.join(worktree_dir, ".git-main-working-tree")
) as repo:
assert repo.bare
assert set([str(r) for r in repo.remotes]) == {"origin"}
assert str(repo.active_branch) == "master"
assert str(repo.head.commit) == head_commit_sha
@pytest.mark.parametrize("configtype", ["toml", "yaml"])
def test_repos_sync_worktree_init(configtype):
with tempfile.TemporaryDirectory() as target:
with tempfile.NamedTemporaryFile() as config:
with open(config.name, "w") as f:
f.write(
templates["worktree_repo_simple"][configtype].format(root=target)
)
cmd = grm(["repos", "sync", "config", "--config", config.name])
assert cmd.returncode == 0
worktree_dir = f"{target}/test"
assert os.path.exists(worktree_dir)
assert set(os.listdir(worktree_dir)) == {".git-main-working-tree"}
with git.Repo(os.path.join(worktree_dir, ".git-main-working-tree")) as repo:
assert repo.bare
assert not repo.head.is_valid()
@pytest.mark.parametrize("configtype", ["toml", "yaml"])
def test_repos_sync_invalid_syntax(configtype):
with tempfile.NamedTemporaryFile() as config:
with open(config.name, "w") as f:
if configtype == "toml":
f.write(
"""
[[trees]]
root = invalid as there are no quotes ;)
"""
)
elif configtype == "yaml":
f.write(
"""
trees:
wrong:
indentation:
"""
)
else:
raise NotImplementedError()
cmd = grm(["repos", "sync", "config", "--config", config.name])
assert cmd.returncode != 0
@pytest.mark.parametrize("configtype", ["toml", "yaml"])
def test_repos_sync_unchanged(configtype):
with tempfile.TemporaryDirectory() as target:
with TempGitFileRemote() as (remote1, remote1_head_commit_sha):
with TempGitFileRemote() as (remote2, remote2_head_commit_sha):
with tempfile.NamedTemporaryFile() as config:
with open(config.name, "w") as f:
f.write(
templates["repo_with_two_remotes"][configtype].format(
root=target, remote1=remote1, remote2=remote2
)
)
cmd = grm(["repos", "sync", "config", "--config", config.name])
assert cmd.returncode == 0
before = checksum_directory(target)
cmd = grm(["repos", "sync", "config", "--config", config.name])
after = checksum_directory(target)
assert cmd.returncode == 0
assert before == after
@pytest.mark.parametrize("configtype", ["toml", "yaml"])
def test_repos_sync_normal_change_to_worktree(configtype):
with tempfile.TemporaryDirectory() as target:
with TempGitFileRemote() as (remote1, remote1_head_commit_sha):
with TempGitFileRemote() as (remote2, remote2_head_commit_sha):
with tempfile.NamedTemporaryFile() as config:
with open(config.name, "w") as f:
f.write(
templates["repo_with_remote"][configtype].format(
root=target, remote=remote1, remotename="origin"
)
)
cmd = grm(["repos", "sync", "config", "--config", config.name])
assert cmd.returncode == 0
with open(config.name, "w") as f:
f.write(
templates["worktree_repo_with_remote"][configtype].format(
root=target, remote=remote1, remotename="origin"
)
)
cmd = grm(["repos", "sync", "config", "--config", config.name])
assert cmd.returncode != 0
assert "already exists" in cmd.stderr
assert "not using a worktree setup" in cmd.stderr
@pytest.mark.parametrize("configtype", ["toml", "yaml"])
def test_repos_sync_worktree_change_to_normal(configtype):
with tempfile.TemporaryDirectory() as target:
with TempGitFileRemote() as (remote1, remote1_head_commit_sha):
with TempGitFileRemote() as (remote2, remote2_head_commit_sha):
with tempfile.NamedTemporaryFile() as config:
with open(config.name, "w") as f:
f.write(
templates["worktree_repo_with_remote"][configtype].format(
root=target, remote=remote1, remotename="origin"
)
)
cmd = grm(["repos", "sync", "config", "--config", config.name])
assert cmd.returncode == 0
with open(config.name, "w") as f:
f.write(
templates["repo_with_remote"][configtype].format(
root=target, remote=remote1, remotename="origin"
)
)
cmd = grm(["repos", "sync", "config", "--config", config.name])
assert cmd.returncode != 0
assert "already exists" in cmd.stderr
assert "using a worktree setup" in cmd.stderr