Files
git-repo-manager/e2e_tests/test_repos_sync.py
2023-01-26 16:51:46 +01:00

831 lines
32 KiB
Python

#!/usr/bin/env python3
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)
# this removes the prefix (root) from the path (unmanaged_repo)
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 "not a directory" 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()
# as there are no commits yet, HEAD does not point to anything
# valid
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:
"""
There is some bug(?) in GitPython. It does not properly
detect removed remotes. It will still report the old
remove in repo.remotes.
So instead, we make sure that we get an Exception when
we try to access the old remove via repo.remote().
Note that repo.remote() checks the actual repo lazily.
Even `exists()` seems to just check against repo.remotes
and will return True even if the remote is not actually
configured. So we have to force GitPython to hit the filesystem.
calling Remotes.urls does. But it returns an iterator
that first has to be unwrapped via list(). Only THEN
do we actually get an exception of the remotes does not
exist.
"""
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:
# See the note in `test_repos_sync_normal_remove_remote()`
# about repo.remotes
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
# as there are no commits yet, HEAD does not point to anything
# valid
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