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
|
"""A wrapper for the `rsync` command."""
import json
import os
import pathlib
import sys
from . import util
def check_path(path: pathlib.Path, base: pathlib.Path) -> None:
"""Validate a source or destination path."""
print(f"Validating {path} under {base}")
if not path.is_absolute():
sys.exit(f"rsync wrapper invoked with non-absolute path {path}")
if (path.exists() or path.is_symlink()) and not path.is_dir():
sys.exit(f"rsync wrapper invoked with bad path {path}")
try:
path.relative_to(base)
except ValueError:
sys.exit(f"rsync wrapper invoked with path {path} not under {base}")
def main() -> None:
"""Wrap the `rsync` command, check for some things before and after."""
print("rsync wrapper invoked as " + util.cmdstr(sys.argv))
state_file = pathlib.Path(os.environ["REPOPUSH_TEST_STATE"])
state_data = json.loads(state_file.read_text(encoding="UTF-8"))
state_data["steps"] += 1
state_file.write_text(json.dumps(state_data), encoding="UTF-8")
if state_data["noop"]:
if "-n" not in sys.argv:
sys.exit("rsync wrapper invoked without `-n`")
else:
if "-n" in sys.argv:
sys.exit("rsync wrapper invoked with `-n")
if "-az" not in sys.argv:
sys.exit("rsync wrapper invoked without `-az`")
if len(sys.argv) < 5:
sys.exit("rsync wrapper invoked with too few arguments")
if sys.argv[-3] != "--":
sys.exit("rsync wrapper invoked without `-- path path`")
src_path = pathlib.Path(sys.argv[-2])
dst_path = pathlib.Path(sys.argv[-1])
check_path(src_path, state_data["src"])
check_path(dst_path, state_data["dst"])
expected_steps = util.expected_steps(pathlib.Path(state_data["src"]))
step = state_data["steps"]
print(f"Step {step} of {expected_steps}")
if step > expected_steps:
sys.exit(
f"rsync wrapper invoked for step {step}, "
f"only expected {expected_steps}"
)
if step == expected_steps:
if "--delete" not in sys.argv:
sys.exit(
"rsync wrapper invoked without `--delete` on the last step"
)
else:
if "--delete" in sys.argv:
sys.exit(
"rsync wrapper invoked with `--delete` before the last step"
)
rsync = os.environ["REPOPUSH_TEST_RSYNC"]
os.execv(rsync, [rsync] + sys.argv[1:])
if __name__ == "__main__":
main()
|
