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
|
import re
from argparse import ArgumentParser
from os.path import dirname, abspath, join
from subprocess import check_output, call
def git(command, repo):
return check_output('git '+command, cwd=repo, shell=True).decode()
def repo_state_bad(mock_repo):
status = git('status', mock_repo)
if 'You are in the middle of an am session' in status:
print(f'Mock repo at {mock_repo} needs cleanup:\n')
call('git status', shell=True)
return True
def cleanup_old_patches(mock_repo):
print('cleaning up old patches:')
call('rm -vf /tmp/*.mock.patch', shell=True)
call('find . -name "*.rej" -print -delete', shell=True, cwd=mock_repo)
def find_initial_cpython_rev():
with open('lastsync.txt') as source:
return source.read().strip()
def cpython_revs_affecting_mock(cpython_repo, start):
revs = git(f'log --no-merges --format=%H {start}.. '
f'-- Lib/unittest/mock.py Lib/unittest/test/testmock/',
repo=cpython_repo).split()
revs.reverse()
print(f'{len(revs)} patches that may need backporting')
return revs
def has_been_backported(mock_repo, cpython_rev):
backport_rev = git(f'log --format=%H --grep "Backports: {cpython_rev}"',
repo=mock_repo).strip()
if backport_rev:
print(f'{cpython_rev} backported in {backport_rev}')
return True
print(f'{cpython_rev} has not been backported')
def extract_patch_for(cpython_repo, rev):
return git(f'format-patch -1 --no-stat --keep-subject --signoff --stdout {rev}',
repo=cpython_repo)
def munge(rev, patch):
sign_off = 'Signed-off-by:'
patch = patch.replace(sign_off, f'Backports: {rev}\n{sign_off}', 1)
for pattern, sub in (
('(a|b)/Lib/unittest/mock.py', r'\1/mock/mock.py'),
(r'(a|b)/Lib/unittest/test/testmock/(\S+)', r'\1/mock/tests/\2'),
('(a|b)/Misc/NEWS', r'\1/NEWS'),
('(a|b)/NEWS.d/next/[^/]+/(.+\.rst)', r'\1/NEWS.d/\2'),
):
patch = re.sub(pattern, sub, patch)
return patch
def apply_patch(mock_repo, rev, patch):
patch_path = f'/tmp/{rev}.mock.patch'
with open(patch_path, 'w') as target:
target.write(patch)
print(f'wrote {patch_path}')
call(f'git am -k '
f'--include "mock/*" --include NEWS --include "NEWS.d/*" '
f'--reject {patch_path} ',
cwd=mock_repo, shell=True)
def update_last_sync(mock_repo, rev):
with open(join(mock_repo, 'lastsync.txt'), 'w') as target:
target.write(rev+'\n')
print(f'update lastsync.txt to {rev}')
def rev_from_mock_patch(text):
match = re.search('Backports: ([a-z0-9]+)', text)
return match.group(1)
def skip_current(mock_repo, reason):
text = git('am --show-current-patch', repo=mock_repo)
rev = rev_from_mock_patch(text)
git('am --abort', repo=mock_repo)
print(f'skipping {rev}')
update_last_sync(mock_repo, rev)
call(f'git commit -m "Backports: {rev}, skipped: {reason}" lastsync.txt', shell=True, cwd=mock_repo)
cleanup_old_patches(mock_repo)
def commit_last_sync(revs, mock_repo):
print('Yay! All caught up!')
if len(revs):
git('commit -m "latest sync point" lastsync.txt', repo=mock_repo)
def main():
args = parse_args()
if args.skip_current:
return skip_current(args.mock, args.skip_reason)
if repo_state_bad(args.mock):
return
cleanup_old_patches(args.mock)
initial_cpython_rev = find_initial_cpython_rev()
revs = cpython_revs_affecting_mock(args.cpython, initial_cpython_rev)
for rev in revs:
if has_been_backported(args.mock, rev):
update_last_sync(args.mock, rev)
continue
patch = extract_patch_for(args.cpython, rev)
patch = munge(rev, patch)
apply_patch(args.mock, rev, patch)
break
else:
commit_last_sync(revs, args.mock)
def parse_args():
parser = ArgumentParser()
parser.add_argument('--cpython', default='../cpython')
parser.add_argument('--mock', default=abspath(dirname(__file__)))
parser.add_argument('--skip-current', action='store_true')
parser.add_argument('--skip-reason', default='it has no changes needed here.')
return parser.parse_args()
if __name__ == '__main__':
main()
|
