1import base64
2from contextlib import redirect_stderr, redirect_stdout
3from datetime import datetime, timezone
4import hashlib
5import io
6import json
7from pathlib import Path
8import runpy
9import stat
10import subprocess
11import sys
12import tempfile
13import traceback
14import unittest
15from unittest.mock import Mock, patch
16
17release = runpy.run_path(str(Path(__file__).resolve().parent / 'release.py'))
18
19
20def utc(text):
21 return datetime.fromisoformat(text).replace(tzinfo=timezone.utc)
22
23
24class ReleaseTest(unittest.TestCase):
25 def test_notary_arguments_reach_child_without_entering_logs_or_tracebacks(self):
26 for status in (0, 7):
27 with self.subTest(status=status), tempfile.TemporaryDirectory() as folder:
28 record = Path(folder) / 'arguments.json'
29 secrets = ['PRIVATE-KEY-PATH', 'PRIVATE-KEY-ID', 'PRIVATE-ISSUER']
30 command = ['xcrun', 'notarytool', 'submit', Path('archive.zip'),
31 '--key', secrets[0], '--key-id', secrets[1], '--issuer', secrets[2], '--wait']
32 child = ('import json, pathlib, sys; '
33 'pathlib.Path(sys.argv[1]).write_text(json.dumps(sys.argv[3:])); '
34 'print("notary output"); print("notary diagnostic", file=sys.stderr); '
35 'sys.exit(int(sys.argv[2]))')
36 actual_run = subprocess.run
37
38 def notary(arguments, **kwargs):
39 return actual_run([sys.executable, '-c', child, str(record), str(status),
40 *map(str, arguments)], **kwargs)
41
42 printed = io.StringIO()
43 with tempfile.TemporaryFile(mode='w+') as output, tempfile.TemporaryFile(mode='w+') as errors, \
44 patch.object(release['subprocess'], 'run', side_effect=notary), \
45 redirect_stdout(printed):
46 if status:
47 try:
48 release['run'](command, stdout=output, stderr=errors)
49 except subprocess.CalledProcessError as error:
50 self.assertIs(type(error), subprocess.CalledProcessError)
51 self.assertEqual(error.returncode, status)
52 try:
53 raise RuntimeError('Release failed') from error
54 except RuntimeError as chained:
55 trace = ''.join(traceback.format_exception(type(chained), chained, chained.__traceback__))
56 trace += repr(chained.__cause__) + repr(error.args)
57 else:
58 self.fail('The notary child did not fail')
59 else:
60 self.assertIsNone(release['run'](command, stdout=output, stderr=errors))
61 trace = ''
62 output.seek(0)
63 errors.seek(0)
64 self.assertEqual(output.read(), 'notary output\n')
65 self.assertEqual(errors.read(), 'notary diagnostic\n')
66 self.assertEqual(json.loads(record.read_text()), list(map(str, command)))
67 self.assertIn('xcrun notarytool', printed.getvalue())
68 for secret in secrets:
69 self.assertNotIn(secret, printed.getvalue() + trace)
70
71 def test_notary_timeout_redacts_arguments_without_changing_failure(self):
72 command = ['xcrun', 'notarytool', 'submit', '--password', 'PRIVATE-PASSWORD']
73 error = subprocess.TimeoutExpired(command, 3, output=b'output', stderr=b'diagnostic')
74 with patch.object(release['subprocess'], 'run', side_effect=error), redirect_stdout(io.StringIO()):
75 with self.assertRaises(subprocess.TimeoutExpired) as raised:
76 release['run'](command, timeout=3)
77 self.assertIs(raised.exception, error)
78 self.assertEqual(error.timeout, 3)
79 self.assertEqual(error.output, b'output')
80 self.assertEqual(error.stderr, b'diagnostic')
81 self.assertNotIn('PRIVATE-PASSWORD', repr(error) + repr(error.args) +
82 ''.join(traceback.format_exception(type(error), error, error.__traceback__)))
83
84 def test_other_command_failure_keeps_its_arguments_and_output(self):
85 command = ['codesign', '--verify', Path('Snowbound.app')]
86 error = subprocess.CalledProcessError(2, command, output=b'output', stderr=b'diagnostic')
87 printed = io.StringIO()
88 with patch.object(release['subprocess'], 'run', side_effect=error), redirect_stdout(printed):
89 with self.assertRaises(subprocess.CalledProcessError) as raised:
90 release['run'](command)
91 self.assertIs(raised.exception, error)
92 self.assertIs(error.cmd, command)
93 self.assertEqual(error.output, b'output')
94 self.assertEqual(error.stderr, b'diagnostic')
95 self.assertIn('codesign --verify Snowbound.app', printed.getvalue())
96
97 def test_staple_retries_only_a_lost_connection_and_propagates_last_failure(self):
98 network = 'error is Error Domain=NSURLErrorDomain Code=-1005 "The network connection was lost."\n'
99 for failures in ([(68, network)], [(68, network)] * 4, [(68, network)] * 5,
100 [(68, 'The bundle has not been notarized.\n')],
101 [(77, network)], [(68, network.replace('-1005', '-10050'))]):
102 with self.subTest(failures=failures), tempfile.TemporaryDirectory() as folder:
103 calls = []
104 errors = []
105
106 def failing_tool(command, **kwargs):
107 calls.append((command, kwargs))
108 if len(calls) <= len(failures):
109 status, diagnostic = failures[len(calls) - 1]
110 kwargs['stdout'].write(diagnostic)
111 error = subprocess.CalledProcessError(status, command)
112 errors.append(error)
113 raise error
114 kwargs['stdout'].write('The staple and validate action worked!\n')
115
116 printed = io.StringIO()
117 sleep = Mock()
118 with patch.dict(release['staple'].__globals__, {'run': failing_tool}), \
119 patch.object(release['time'], 'sleep', sleep), redirect_stdout(printed):
120 if failures[0] == (68, network) and len(failures) < 5:
121 self.assertIsNone(release['staple'](Path(folder) / 'Snowbound.app'))
122 else:
123 with self.assertRaises(subprocess.CalledProcessError) as raised:
124 release['staple'](Path(folder) / 'Snowbound.app')
125 self.assertIs(raised.exception, errors[-1])
126 retried = failures[0] == (68, network)
127 self.assertEqual(len(calls), min(len(failures) + 1, 5) if retried else 1)
128 self.assertEqual([wait for (wait,), _ in sleep.call_args_list], [10, 20, 40, 80][:len(calls) - 1])
129 self.assertEqual(printed.getvalue(), ''.join(text for _, text in failures) +
130 ('The staple and validate action worked!\n' if len(errors) < len(calls) else ''))
131 for command, kwargs in calls:
132 self.assertEqual(command, ['xcrun', 'stapler', 'staple', Path(folder) / 'Snowbound.app'])
133 self.assertEqual(kwargs['timeout'], 120)
134 self.assertEqual(kwargs['stderr'], subprocess.STDOUT)
135
136 def test_staple_timeout_preserves_diagnostic_without_retrying(self):
137 command = ['xcrun', 'stapler', 'staple', Path('Snowbound.app')]
138 error = subprocess.TimeoutExpired(command, 120)
139
140 def timeout(_, **kwargs):
141 kwargs['stdout'].write('Incomplete ticket download\n')
142 raise error
143
144 printed = io.StringIO()
145 with patch.dict(release['staple'].__globals__, {'run': Mock(side_effect=timeout)}), redirect_stdout(printed):
146 with self.assertRaises(subprocess.TimeoutExpired) as raised:
147 release['staple'](command[-1])
148 release['staple'].__globals__['run'].assert_called_once()
149 self.assertIs(raised.exception, error)
150 self.assertEqual(printed.getvalue(), 'Incomplete ticket download\n')
151
152 def test_download_bytes_survive_rejected_permissions(self):
153 with tempfile.TemporaryDirectory() as folder:
154 source, destination = Path(folder) / 'source', Path(folder) / 'download'
155 source.write_bytes(b'an executable')
156 warning = io.StringIO()
157 with patch.object(release['shutil'], 'copymode', side_effect=PermissionError('chmod denied')), \
158 redirect_stderr(warning):
159 release['copy_download'](source, destination)
160 self.assertEqual(destination.read_bytes(), source.read_bytes())
161 self.assertIn(str(destination), warning.getvalue())
162 self.assertIn('could not preserve file permissions', warning.getvalue())
163
164 def test_download_preserves_supported_permissions(self):
165 with tempfile.TemporaryDirectory() as folder:
166 source, destination = Path(folder) / 'source', Path(folder) / 'download'
167 source.write_bytes(b'an executable')
168 source.chmod(0o750)
169 release['copy_download'](source, destination)
170 self.assertEqual(destination.read_bytes(), source.read_bytes())
171 self.assertEqual(stat.S_IMODE(destination.stat().st_mode), 0o750)
172
173 def test_download_content_failure_stops_before_permissions(self):
174 for error in (PermissionError('write denied'), OSError('disk full')):
175 with self.subTest(error=error), \
176 patch.object(release['shutil'], 'copyfile', side_effect=error), \
177 patch.object(release['shutil'], 'copymode') as copymode:
178 with self.assertRaises(type(error)) as raised:
179 release['copy_download'](Path('source'), Path('download'))
180 self.assertIs(raised.exception, error)
181 copymode.assert_not_called()
182
183 def test_download_other_mode_copy_errors_propagate(self):
184 with tempfile.TemporaryDirectory() as folder:
185 source, destination = Path(folder) / 'source', Path(folder) / 'download'
186 source.write_bytes(b'an executable')
187 with patch.object(release['shutil'], 'copymode', side_effect=OSError('missing source')):
188 with self.assertRaises(OSError):
189 release['copy_download'](source, destination)
190
191 def test_publication_finishes_when_share_rejects_permissions(self):
192 main, scope = release['main'], release['main'].__globals__
193 with tempfile.TemporaryDirectory() as folder:
194 root = Path(folder)
195 published = root / 'published'
196 published.mkdir()
197 source = root / 'snowbound'
198 source.write_bytes(b'an executable')
199 source.chmod(0o750)
200 denied = []
201 copymode = release['shutil'].copymode
202
203 def share_modes(source, destination, **kwargs):
204 if Path(destination).is_relative_to(published):
205 denied.append(Path(destination))
206 raise PermissionError('share rejects chmod')
207 return copymode(source, destination, **kwargs)
208
209 def minisign(files):
210 for file in files:
211 Path(f'{file}.minisig').write_bytes(b'archive signature')
212
213 def split_debug(executable, debug):
214 debug.write_bytes(b'debug symbols')
215
216 overrides = {
217 'ROOT': root, 'PUBLISHED': published, 'FIRST': 'a',
218 'jj': Mock(return_value='a'), 'clean': Mock(), 'clean_local': Mock(), 'run': Mock(),
219 'history': lambda: {'a': ([], utc('2026-10-05T12:00:00'), 'fix: publish')},
220 'build_linux': lambda architectures: {'linux-x86_64': source},
221 'split_debug': split_debug, 'sign': lambda files: ['00' for _ in files],
222 'minisign': minisign,
223 }
224 warning = io.StringIO()
225 with patch.dict(scope, overrides), \
226 patch.object(release['sys'], 'argv', ['release.py', '--platforms', 'linux-x86_64']), \
227 patch.dict(release['os'].environ), \
228 patch.object(release['subprocess'], 'run', return_value=Mock(stdout='')), \
229 patch.object(release['shutil'], 'copymode', side_effect=share_modes), \
230 redirect_stderr(warning), redirect_stdout(io.StringIO()):
231 main()
232 target = published / '2026-10-05.r1'
233 archive = 'snowbound-2026-10-05-r1-linux-x86_64'
234 stable = published / 'latest' / 'snowbound-linux-x86_64'
235 self.assertEqual((target / archive).read_bytes(), source.read_bytes())
236 self.assertEqual((target / f'{archive}.minisig').read_bytes(), b'archive signature')
237 build = json.loads((target / 'build.json').read_text())
238 self.assertEqual(build['archives']['linux-x86_64']['sha256'],
239 hashlib.sha256(source.read_bytes()).hexdigest())
240 self.assertEqual((target / 'build.json.sig').read_text(), '00\n')
241 self.assertEqual(stable.read_bytes(), source.read_bytes())
242 self.assertEqual(Path(f'{stable}.minisig').read_bytes(), b'archive signature')
243 self.assertEqual(json.loads((published / 'latest.json').read_text()),
244 {'linux-x86_64': '2026-10-05-r1'})
245 self.assertIn(published / '.2026-10-05.r1.partial' / archive, denied)
246 self.assertIn(stable.with_name(f'.{stable.name}.partial'), denied)
247 self.assertEqual(warning.getvalue().count('could not preserve file permissions'), len(denied))
248
249 def test_a_build_counts_the_commits_of_its_day_in_los_angeles(self):
250 # 07:34 UTC on the 29th is 00:34 in Los Angeles; 05:00 UTC on the 30th is 22:00 on the 29th.
251 commits = [utc('2026-09-30T05:00:00'), utc('2026-09-29T19:23:00'), utc('2026-09-29T07:34:00'),
252 utc('2026-09-29T06:29:00'), utc('2026-09-28T12:43:00')]
253 self.assertEqual(release['derive'](commits[0], commits), ('2026-09-29', 3))
254 self.assertEqual(release['derive'](commits[3], commits[3:]), ('2026-09-28', 2))
255 self.assertEqual(release['name'](('2026-09-29', 10)), '2026-09-29-r10')
256 self.assertEqual(release['parse']('2026-09-29-r10'), ('2026-09-29', 10))
257 self.assertEqual(release['folder'](('2026-09-29', 10)), '2026-09-29.r10')
258
259 def test_a_commit_counts_once_by_its_prefix_or_once_per_bullet(self):
260 entries = release['entries']
261 self.assertEqual(entries('fix(update): accept an archive of exactly its size\n\nureq refuses.\n'),
262 [('fix', 'Accept an archive of exactly its size')])
263 self.assertEqual(entries('docs: center the row'), [('other', 'Center the row')])
264 self.assertEqual(entries('Initial import'), [('other', 'Initial import')])
265 self.assertEqual(entries('feat!: macOS icon'), [('feature', 'macOS icon')])
266 self.assertEqual(
267 entries('feat: icon, builds\n\nIntro.\n\n- macOS ships an icon so Tahoe\n shows it.\n'
268 '* pinch zoom.\n\nAssisted-by: claude-opus-5.5\n'),
269 [('feature', 'macOS ships an icon so Tahoe shows it'), ('feature', 'Pinch zoom')])
270
271 def test_changes_start_after_the_build_before_and_carry_their_commits_versions(self):
272 commits = {'a': ([], utc('2026-09-30T10:00:00'), 'fix: published first'),
273 'b': (['a'], utc('2026-09-30T11:00:00'), 'feat: pinch zoom'),
274 'c': (['b'], utc('2026-09-30T12:00:00'), 'fix: two\n\n- one\n- two\n')}
275 self.assertEqual(release['changes'](commits, 'c', 'a'), [
276 {'version': '2026-09-30-r2', 'kind': 'feature', 'title': 'Pinch zoom'},
277 {'version': '2026-09-30-r3', 'kind': 'fix', 'title': 'One'},
278 {'version': '2026-09-30-r3', 'kind': 'fix', 'title': 'Two'}])
279 self.assertEqual(release['changes'](commits, 'c', 'b'), [
280 {'version': '2026-09-30-r3', 'kind': 'fix', 'title': 'One'},
281 {'version': '2026-09-30-r3', 'kind': 'fix', 'title': 'Two'}])
282
283 def test_builds_are_the_published_folders_oldest_first(self):
284 with tempfile.TemporaryDirectory() as published:
285 for entry in ['2026-10-02.r34', '2026-09-30.r2', '2026-10-02.r7', '.2026-10-03.r1.partial', 'latest']:
286 (Path(published) / entry).mkdir()
287 (Path(published) / 'latest.json').touch()
288 self.assertEqual(release['builds'](Path(published)),
289 [('2026-09-30', 2), ('2026-10-02', 7), ('2026-10-02', 34)])
290
291 def test_latest_only_moves_forward(self):
292 latest = {'macos-aarch64': '2026-09-29-r9', 'linux-x86_64': '2026-09-30-r1'}
293 moved = release['newest'](latest, {'macos-aarch64': {}, 'linux-x86_64': {}, 'macos-10.6': {}},
294 ('2026-09-29', 10))
295 self.assertEqual(moved, {'macos-aarch64': '2026-09-29-r10', 'linux-x86_64': '2026-09-30-r1',
296 'macos-10.6': '2026-09-29-r10'})
297 self.assertEqual(release['newest'](latest, {'macos-aarch64': {}}, ('2026-09-29', 9)), latest)
298
299 def test_minisig_signs_the_files_blake2b_then_that_with_its_comment(self):
300 minisign, scope = release['minisign'], release['minisign'].__globals__
301 messages = []
302
303 def sign(paths):
304 messages.extend(Path(path).read_bytes() for path in paths)
305 return [bytes([len(messages)]).hex() * 64 for _ in paths]
306
307 real, scope['sign'] = scope['sign'], sign
308 try:
309 with tempfile.TemporaryDirectory() as folder:
310 file = Path(folder) / 'snowbound-linux-x86_64'
311 file.write_bytes(b'an executable')
312 minisign([file])
313 untrusted, signature, trusted, global_signature = Path(f'{file}.minisig').read_text().splitlines()
314 finally:
315 scope['sign'] = real
316 key_id = base64.b64decode(release['MINISIGN'].read_text().splitlines()[1])[2:10]
317 self.assertTrue(untrusted.startswith('untrusted comment: '))
318 self.assertEqual(base64.b64decode(signature), b'ED' + key_id + bytes([1]) * 64)
319 self.assertRegex(trusted, r'^trusted comment: timestamp:\d+\tfile:snowbound-linux-x86_64\thashed$')
320 self.assertEqual(base64.b64decode(global_signature), bytes([2]) * 64)
321 self.assertEqual(messages, [hashlib.blake2b(b'an executable').digest(),
322 bytes([1]) * 64 + trusted.removeprefix('trusted comment: ').encode()])
323
324 def test_build_macos_signs_each_mac_app(self):
325 build_mac, scope = release['build_mac'], release['build_mac'].__globals__
326 commands = []
327
328 def record(command, **_):
329 commands.append(list(map(str, command)))
330 if command[0] == 'ditto':
331 Path(command[-1]).touch()
332
333 run, scope['run'] = scope['run'], record
334 try:
335 def signing(platform, developer_id, notarize):
336 commands.clear()
337 with tempfile.TemporaryDirectory() as stage:
338 build_mac(platform, Path(stage), developer_id, notarize, Path(stage) / 'symbols.zip')
339 build = commands[0]
340 return (build[build.index(f'{stage}/Snowbound.dSYM') + 1:],
341 [command[1] for command in commands[1:]])
342
343 self.assertEqual(signing('macos-aarch64', True, ['--key-id', 'K']),
344 (['--sign', 'developer-id', '--sign-identity', release['IDENTITY'], '--arch', 'aarch64'],
345 ['-c', '-c', 'notarytool', 'stapler', '-c']))
346 self.assertEqual(signing('macos-x86_64', True, ['--key-id', 'K']),
347 (['--sign', 'developer-id', '--sign-identity', release['IDENTITY'], '--arch', 'x86_64'],
348 ['-c', '-c', 'notarytool', 'stapler', '-c']))
349 self.assertEqual(signing('macos-aarch64', False, None), (['--sign', 'ad-hoc', '--arch', 'aarch64'], ['-c', '-c']))
350 self.assertEqual(signing('macos-10.6', True, ['--key-id', 'K']), (['--snow-leopard'], ['-c', '-c']))
351 finally:
352 scope['run'] = run
353
354
355if __name__ == '__main__':
356 unittest.main()