1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
from __future__ import annotations
from pathlib import Path
import pygit2
from pygit2 import Repository, Signature, init_repository
from pygittools.hooks import HookResult
from pygittools.hooks_push import PostCommitPushRemotes, force_with_lease_push
SIG = Signature("test", "test@example.com")
def _init_client_with_remote(tmp_path: Path) -> tuple[Repository, Path, Path, Path]:
bare_path: Path = tmp_path / "remote.git"
bare_path.mkdir()
init_repository(str(bare_path), bare=True)
backup_path: Path = tmp_path / "backup.git"
backup_path.mkdir()
init_repository(str(backup_path), bare=True)
client_path: Path = tmp_path / "client"
client: Repository = init_repository(str(client_path), bare=False)
client.remotes.create("origin", str(bare_path))
client.remotes.create("backup", str(backup_path))
(client_path / "f").write_text("a\n", encoding="utf-8")
index = client.index
index.add("f")
index.write()
tree = index.write_tree()
client.create_commit("HEAD", SIG, SIG, "init", tree, [])
client.references.create("refs/heads/wip/topic", client.head.target, force=True)
client.checkout(f"refs/heads/wip/topic")
return client, client_path, bare_path, backup_path
def test_skips_non_matching_branch(tmp_path: Path) -> None:
client, client_path, _bare, _backup = _init_client_with_remote(tmp_path)
client.references.create("refs/heads/main", client.head.target, force=True)
client.checkout("refs/heads/main")
repo = pygit2.Repository(str(client_path))
assert PostCommitPushRemotes(repo, r"^wip/").run() == HookResult.SUCCESS
assert repo.references.get("refs/remotes/origin/main") is None
def test_pushes_matching_branch_to_all_remotes(tmp_path: Path) -> None:
client, client_path, bare_path, backup_path = _init_client_with_remote(tmp_path)
repo = pygit2.Repository(str(client_path))
assert PostCommitPushRemotes(repo, r"^wip/").run() == HookResult.SUCCESS
local_target = client.references["refs/heads/wip/topic"].target
assert pygit2.Repository(str(bare_path)).references["refs/heads/wip/topic"].target == local_target
assert pygit2.Repository(str(backup_path)).references["refs/heads/wip/topic"].target == local_target
def test_force_with_lease_rejects_stale_tracking_ref(tmp_path: Path) -> None:
client, client_path, bare_path, _backup = _init_client_with_remote(tmp_path)
repo = pygit2.Repository(str(client_path))
assert force_with_lease_push(repo, "origin", "wip/topic") is None
(client_path / "f").write_text("b\n", encoding="utf-8")
index = repo.index
index.add("f")
index.write()
tree = index.write_tree()
parent = repo.head.target
repo.create_commit("refs/heads/wip/topic", SIG, SIG, "second", tree, [parent])
repo.checkout("refs/heads/wip/topic")
bare = pygit2.Repository(str(bare_path))
tb = bare.TreeBuilder()
tree = tb.write()
current = bare.references["refs/heads/wip/topic"].target
other = bare.create_commit("refs/heads/wip/topic", SIG, SIG, "other", tree, [current])
assert other != repo.references["refs/heads/wip/topic"].target
error = force_with_lease_push(repo, "origin", "wip/topic")
assert error is not None
assert "rejected" in error.lower() or "failed" in error.lower()
def test_skips_detached_head(tmp_path: Path) -> None:
client, client_path, _bare, _backup = _init_client_with_remote(tmp_path)
client.set_head(client.head.target)
repo = pygit2.Repository(str(client_path))
assert repo.head_is_detached is True
assert PostCommitPushRemotes(repo, r".*").run() == HookResult.SUCCESS