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
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
|
"""
Synchronize or validate a sparse subrepository export. The sparse export
configuration is stored in the `.sparse-export` file in the root of the
repository. If the file does not exist, then there is nothing to update
or verify.
"""
import argparse
import hashlib
import logging
import os
import shutil
import subprocess
import sys
logger = logging.getLogger(__name__)
def setup_argparser(argparser):
argparser.add_argument(
"--log-level", default="warning",
choices=["debug", "info", "warning", "error"])
argparser.add_argument(
"command", choices=["update", "verify"],
help="whether to update or verify the export")
argparser.add_argument("--force")
def get_argdict(args):
out = {}
for key, value in vars(args).items():
if key.startswith("_"):
continue
out[key] = value
return out
def copy_file(sourcepath, destpath):
destdir = os.path.dirname(destpath)
if not os.path.exists(destdir):
logger.debug("Making directory %s", destdir)
os.makedirs(destdir)
logger.debug("Copying %s -> %s", sourcepath, destpath)
shutil.copy2(sourcepath, destpath)
return 0
def read_chunks(infile, chunksize=4096):
while True:
chunk = infile.read(chunksize)
if chunk == b'':
break
yield chunk
def hash_file(filepath, hashname="sha1", chunksize=4096):
hashobj = hashlib.new(hashname)
with open(filepath, "rb") as infile:
for chunk in read_chunks(infile):
hashobj.update(chunk)
return hashobj
def verify_file(sourcepath, destpath):
if not os.path.exists(destpath):
logger.error("Missing %s", destpath)
return 1
if hash_file(sourcepath).digest() == hash_file(destpath).digest():
logger.info("File %s up to date", destpath)
return 0
logger.error("File %s differs from source", destpath)
return 1
def inner_main(command, force):
if command == "update":
operation = copy_file
elif command == "verify":
operation = verify_file
else:
raise ValueError("Unknown command {}".format(command))
repodir = subprocess.check_output(
["git", "rev-parse", "--show-toplevel"]).strip().decode("utf-8")
exportsdir = os.path.join(repodir, "tangent/tooling/sparse-exports")
if force:
logger.debug("Forcing export of %s", force)
dirname = force
else:
if os.path.exists(os.path.join(exportsdir, "iamgroot.txt")):
# This is the monorepo
return 0
sparse_sentinel = os.path.join(repodir, ".sparse-export")
if not os.path.exists(exportsdir) and os.path.exists(sparse_sentinel):
# This is the public export repo
return 0
dirnames = os.listdir(exportsdir)
if len(dirnames) != 1:
logger.error(
"Invalid sparse export, %s contains too many directories", exportsdir)
return 1
dirname = dirnames[0]
export_dir = os.path.join(exportsdir, dirname)
returncode = 0
for directory, _dirnames, filenames in os.walk(export_dir):
reldir = os.path.relpath(directory, export_dir)
for filename in filenames:
if reldir == ".":
relpath_file = filename
else:
relpath_file = os.path.join(reldir, filename)
sourcepath = os.path.join(directory, filename)
if relpath_file == "sparse-checkout":
gitdir = subprocess.check_output(
["git", "rev-parse", "--git-dir"]).decode("utf-8").strip()
destpath = os.path.join(gitdir, "info/sparse-checkout")
else:
destpath = os.path.join(repodir, relpath_file)
returncode |= operation(sourcepath, destpath)
return returncode
def main():
logging.basicConfig(level=logging.INFO)
argparser = argparse.ArgumentParser(description=__doc__)
setup_argparser(argparser)
try:
import argcomplete
argcomplete.autocomplete(argparser)
except ImportError:
pass
args = argparser.parse_args()
argdict = get_argdict(args)
log_level = argdict.pop("log_level")
logging.getLogger().setLevel(getattr(logging, log_level.upper()))
return inner_main(**argdict)
if __name__ == "__main__":
sys.exit(main())
|
