"""Exercise artifact integrity and deployment failure boundaries with synthetic data.""" import contextlib import hashlib import io import json import os import runpy from pathlib import Path import sys import tarfile import tempfile import unittest from unittest.mock import Mock, patch import urllib.error import urllib.request import zipfile sys.path.insert(0, str(Path(__file__).resolve().parents[1])) import deploy import release cleanup = runpy.run_path(str(Path(__file__).resolve().parents[1] / 'ci-cleanup.py'))['cleanup'] SHA = 'a' * 40 DIGEST = 'sha256:' + 'b' * 64 class BundleTests(unittest.TestCase): def setUp(self): self.tmp = tempfile.TemporaryDirectory() self.addCleanup(self.tmp.cleanup) self.cwd = Path.cwd() os.chdir(self.tmp.name) self.addCleanup(os.chdir, self.cwd) Path('VERSION').write_text('0.1.1\n') Path('downloads').mkdir() Path('downloads/rawmark_0.1.1_linux_amd64.tar.gz').write_bytes(b'archive') self.environment = patch.dict(os.environ, {'CI_RUN_ID': '42'}) self.environment.start() self.addCleanup(self.environment.stop) self.git = patch.object(release, 'run', return_value=SHA) self.git.start() self.addCleanup(self.git.stop) def test_development_has_only_one_target_and_exact_checksums(self): release.bundle('development') build = release.verify_bundle() self.assertEqual(build['platforms'], ['linux/amd64']) self.assertEqual(build['run_id'], 42) self.assertEqual(Path('downloads/SHA256SUMS').read_text(), hashlib.sha256(b'archive').hexdigest() + ' rawmark_0.1.1_linux_amd64.tar.gz\n') def test_corrupted_download_is_rejected(self): release.bundle('development') Path('downloads/rawmark_0.1.1_linux_amd64.tar.gz').write_bytes(b'corrupt') with self.assertRaisesRegex(ValueError, 'Checksum mismatch'): release.verify_bundle() def test_archive_at_cdn_budget_is_rejected(self): with patch.object(release, 'UPLOAD_LIMIT', 7): with self.assertRaisesRegex(ValueError, 'CDN upload budget'): release.bundle('development') def test_receipt_from_other_commit_is_rejected(self): release.bundle('development') with patch.object(release, 'run', return_value='c' * 40): with self.assertRaisesRegex(ValueError, 'commit mismatch'): release.verify_bundle() def test_missing_release_platform_is_rejected(self): release.bundle('development') path = Path('downloads/build.json') build = json.loads(path.read_text()) build['mode'] = 'release' release.write_json(path, build) with self.assertRaisesRegex(ValueError, 'platforms mismatch'): release.verify_bundle() class PublicationTests(unittest.TestCase): def test_cross_host_redirect_strips_credentials(self): request = urllib.request.Request('https://api.github.com/asset', headers={'Authorization': 'Bearer private'}) redirected = release.SafeRedirect().redirect_request(request, None, 302, 'Found', {}, 'https://objects.example/asset') self.assertFalse(redirected.has_header('Authorization')) def test_anonymous_client_cannot_write(self): api = release.API('https://api.github.com', '', github=True) with self.assertRaisesRegex(ValueError, 'cannot mutate'): api.repo('/releases', data={}, method='POST') def test_existing_asset_conflict_is_not_overwritten(self): with tempfile.TemporaryDirectory() as directory: path = Path(directory) / 'archive.zip' path.write_bytes(b'new') api = release.API('https://git.example/api/v1', 'private') api.assets = Mock(return_value=[{'name': path.name}]) api.download = Mock(return_value=b'old') api.request = Mock() with self.assertRaisesRegex(ValueError, 'conflicts'): api.upload({'draft': True}, path) api.request.assert_not_called() def test_identical_asset_is_reused_even_after_publication(self): with tempfile.TemporaryDirectory() as directory: path = Path(directory) / 'archive.zip' path.write_bytes(b'same') api = release.API('https://git.example/api/v1', 'private') api.assets = Mock(return_value=[{'name': path.name}]) api.download = Mock(return_value=b'same') api.request = Mock() api.upload({'draft': False}, path) api.request.assert_not_called() def test_incomplete_release_cannot_be_deployed(self): gitea, github = Mock(), Mock() gitea.release.return_value = {'draft': True, 'prerelease': False} github.release.return_value = {'draft': False, 'prerelease': False} with self.assertRaisesRegex(ValueError, 'not completely published'): release.release_receipt(gitea, github, '20261030-aaaaaaaa') def test_registry_permission_error_is_not_treated_as_missing_image(self): result = Mock(returncode=1, stderr='403 Forbidden', stdout='') with patch.object(release.subprocess, 'run', return_value=result): with self.assertRaisesRegex(RuntimeError, '403 Forbidden'): release.registry_digest(release.GITEA_IMAGE, '0.1.1') def test_registry_missing_tag_is_distinct_from_other_failures(self): result = Mock(returncode=1, stderr='manifest unknown', stdout='') with patch.object(release.subprocess, 'run', return_value=result): self.assertIsNone(release.registry_digest(release.GITEA_IMAGE, '0.1.1')) class ResumeTests(unittest.TestCase): def setUp(self): self.tmp = tempfile.TemporaryDirectory() self.addCleanup(self.tmp.cleanup) self.cwd = Path.cwd() os.chdir(self.tmp.name) self.addCleanup(os.chdir, self.cwd) Path('VERSION').write_text('0.1.1\n') Path('downloads').mkdir() for platform in release.PLATFORMS: Path('downloads', release.archive_name('0.1.1', platform)).write_bytes(platform.encode()) Path('web/dist-demo').mkdir(parents=True) Path('web/dist-demo/index.html').write_text('Demo') with patch.object(release, 'run', return_value=SHA), patch.dict(os.environ, {'CI_RUN_ID': '42'}): release.bundle('release') self.original = {p.name: p.read_bytes() for p in Path('downloads').iterdir()} for path in Path('downloads').iterdir(): path.unlink() self.gitea = Mock() self.gitea.assets.return_value = [{'name': name} for name in self.original] self.gitea.download.side_effect = lambda asset: self.original[asset['name']] self.draft = {'draft': True, 'body': 'Build run: `42`.'} def test_resume_from_release_assets_does_not_require_old_ci_runs(self): with patch.object(release, 'run', return_value=SHA): release.restore(self.gitea, self.draft) self.gitea.repo.assert_not_called() self.gitea.upload.assert_not_called() self.assertEqual({p.name: p.read_bytes() for p in Path('downloads').iterdir()}, self.original) def test_expired_artifact_fails_instead_of_rebuilding(self): self.gitea.assets.return_value = [{'name': 'build.json'}] self.gitea.repo.return_value = {'artifacts': []} with patch.object(release, 'run', return_value=SHA) as command: with self.assertRaisesRegex(ValueError, 'refusing to rebuild'): release.restore(self.gitea, self.draft) self.assertFalse(any('build' in str(call) for call in command.call_args_list)) def test_interrupted_first_receipt_upload_uses_original_run(self): self.gitea.assets.return_value = [] names = {'checksums': ['build.json', 'SHA256SUMS'], 'demo': ['demo.zip']} names.update({platform.replace('/', '-'): [release.archive_name('0.1.1', platform)] for platform in release.PLATFORMS}) archives = {} for identifier, (name, files) in enumerate(names.items(), 1): data = io.BytesIO() with zipfile.ZipFile(data, 'w') as archive: for filename in files: archive.writestr(filename, self.original[filename]) archives[identifier] = (name, data.getvalue()) def request(path, **kwargs): if path == '/actions/runs/42/artifacts': return {'artifacts': [{'id': identifier, 'name': name, 'expired': False} for identifier, (name, _) in archives.items()]} return archives[int(path.split('/')[-2])][1] self.gitea.repo.side_effect = request with patch.object(release, 'run', return_value=SHA): release.restore(self.gitea, self.draft) self.gitea.upload.assert_called_once_with(self.draft, Path('downloads/build.json')) self.assertEqual({p.name: p.read_bytes() for p in Path('downloads').iterdir()}, self.original) class DeploymentTests(unittest.TestCase): def setUp(self): self.tmp = tempfile.TemporaryDirectory() self.addCleanup(self.tmp.cleanup) self.directory = Path(self.tmp.name).resolve() (self.directory / 'data').mkdir() (self.directory / 'data/rawmark.db').write_bytes(b'database') (self.directory / 'data/rawmark.db-wal').write_bytes(b'wal') (self.directory / 'compose.yml').write_text('services: {}\n') (self.directory / '.env').write_text( f'RAWMARK_CHANNEL=development\nRAWMARK_PROJECT=rawmark-ci-deploy\n' f'RAWMARK_BIND=127.0.0.1\nRAWMARK_PORT=21899\nRAWMARK_DATA_DIR={self.directory}/data\n') self.image = release.GITEA_DEV_IMAGE + '@' + DIGEST self.previous = {'Id': 'old-id', 'Image': 'old-image', 'State': {'Running': True}, 'Mounts': [{'Destination': '/data', 'Type': 'bind', 'Source': str(self.directory / 'data')}]} self.image_info = {'Id': 'new-image', 'Config': {'User': '65532:65532', 'Labels': { 'org.opencontainers.image.revision': SHA, 'org.opencontainers.image.version': '0.1.1'}}} self.commands = [] self.api = patch.object(deploy, 'API') self.api.start().return_value.repo.return_value = {'commit': {'id': SHA}} self.addCleanup(self.api.stop) self.env = patch.dict(os.environ, {'GITEA_API_URL': 'https://git.example/api/v1', 'GITEA_RELEASE_TOKEN': 'private'}) self.env.start() self.addCleanup(self.env.stop) def command(self, *args, **kwargs): self.commands.append(args) if args[:3] == ('docker', 'image', 'inspect'): return json.dumps([self.image_info]) if args[:3] == ('docker', 'run', '--rm'): return 'RawMark 0.1.1' if args[:3] == ('docker', 'ps', '-aq'): return 'old-id' if args[:3] == ('docker', 'ps', '-q'): return 'old-id' if args[:3] == ('docker', 'inspect', 'old-id'): return json.dumps([self.previous]) if args[0:2] == ('docker', 'compose') and 'ps' in args: return 'new-id' if args[:3] == ('docker', 'inspect', 'new-id'): return json.dumps([{'Image': 'new-image', 'NetworkSettings': {'Networks': { 'test': {'NetworkID': 'network', 'IPAddress': '127.0.0.1'}}}}]) return '' def test_channel_mismatch_cannot_touch_docker(self): with patch.object(deploy, 'run') as docker: with self.assertRaisesRegex(ValueError, 'channel'): deploy.deployment(self.directory, 'production', self.image, SHA, '0.1.1') docker.assert_not_called() def test_unresolved_pending_deployment_cannot_be_overwritten(self): (self.directory / 'deployment.pending.json').write_text('{"backup":"keep"}') with patch.object(deploy, 'run') as docker: with self.assertRaisesRegex(ValueError, 'pending deployment'): deploy.deployment(self.directory, 'development', self.image, SHA, '0.1.1') docker.assert_not_called() def test_stale_dev_skips_without_stopping_existing_service(self): with patch.object(deploy, 'API') as api, patch.object(deploy, 'run', side_effect=self.command): api.return_value.repo.return_value = {'commit': {'id': 'c' * 40}} deploy.deployment(self.directory, 'development', self.image, SHA, '0.1.1') self.assertFalse(any(c[:2] == ('docker', 'stop') for c in self.commands)) self.assertFalse((self.directory / 'deployment.json').exists()) def test_wrong_mount_is_rejected_before_stopping(self): self.previous['Mounts'][0]['Source'] = '/some-other-data' with patch.object(deploy, 'run', side_effect=self.command): with self.assertRaisesRegex(ValueError, 'mount differs'): deploy.deployment(self.directory, 'development', self.image, SHA, '0.1.1') self.assertFalse(any(c[:2] == ('docker', 'stop') for c in self.commands)) def test_backup_includes_wal_and_precedes_replacement(self): response = contextlib.nullcontext(io.BytesIO(b'RawMark')) opener = Mock() opener.open.side_effect = [response, contextlib.nullcontext(io.BytesIO(b'{"code":0}'))] with patch.object(deploy, 'run', side_effect=self.command), patch.object(deploy.urllib.request, 'build_opener', return_value=opener): deploy.deployment(self.directory, 'development', self.image, SHA, '0.1.1') receipt = json.loads((self.directory / 'deployment.json').read_text()) with tarfile.open(receipt['backup']) as archive: self.assertEqual(archive.extractfile('data/rawmark.db-wal').read(), b'wal') self.assertEqual(Path(receipt['backup']).stat().st_mode & 0o777, 0o600) stop = next(i for i,c in enumerate(self.commands) if c[:2] == ('docker', 'stop')) start = next(i for i,c in enumerate(self.commands) if c[:2] == ('docker', 'compose') and 'up' in c) self.assertLess(stop, start) self.assertIn('RAWMARK_IMAGE=' + self.image, (self.directory / '.env').read_text()) def test_existing_data_is_backed_up_without_a_container(self): def without_container(*args, **kwargs): if args[:2] == ('docker', 'ps'): return '' return self.command(*args, **kwargs) opener = Mock() opener.open.side_effect = [contextlib.nullcontext(io.BytesIO(b'RawMark')), contextlib.nullcontext(io.BytesIO(b'{"code":0}'))] with patch.object(deploy, 'run', side_effect=without_container), patch.object( deploy.urllib.request, 'build_opener', return_value=opener): deploy.deployment(self.directory, 'development', self.image, SHA, '0.1.1') receipt = json.loads((self.directory / 'deployment.json').read_text()) self.assertTrue(Path(receipt['backup']).exists()) self.assertIsNone(receipt['previous_image']) def test_failed_replacement_keeps_backup_and_does_not_downgrade_data(self): def failure(*args, **kwargs): if args[:2] == ('docker', 'compose') and 'up' in args: raise RuntimeError('candidate failed') return self.command(*args, **kwargs) with patch.object(deploy, 'run', side_effect=failure), patch.object(deploy.subprocess, 'run') as stop: with self.assertRaisesRegex(RuntimeError, 'candidate failed'): deploy.deployment(self.directory, 'development', self.image, SHA, '0.1.1') stop.assert_called_once() pending = json.loads((self.directory / 'deployment.pending.json').read_text()) self.assertTrue(Path(pending['backup']).exists()) self.assertFalse(any(c[:2] == ('docker', 'start') for c in self.commands)) self.assertNotIn('RAWMARK_IMAGE=', (self.directory / '.env').read_text()) class DevelopmentVersionTests(unittest.TestCase): def test_version_uses_utc_commit_date_and_exact_eight_character_hash(self): with patch.dict(os.environ, {'RAWMARK_BUILD_CHANNEL': 'development'}), patch.object( release, 'run', side_effect=[SHA, '0']): self.assertEqual(release.build_version(), '19700101-aaaaaaaa') self.assertEqual(release.archive_name('19700101-aaaaaaaa', 'linux/amd64'), 'RawMark_19700101-aaaaaaaa_linux_amd64.tar.gz') class CleanupTests(unittest.TestCase): def test_cleanup_rejects_broad_filter(self): with patch('subprocess.check_output') as docker: with self.assertRaises(ValueError): cleanup('') docker.assert_not_called() if __name__ == '__main__': unittest.main() class ReleaseReservationTests(unittest.TestCase): def setUp(self): self.directory = tempfile.TemporaryDirectory() self.addCleanup(self.directory.cleanup) self.patch = patch.object(release, 'STATE_DIRECTORY', Path(self.directory.name)) self.patch.start() self.addCleanup(self.patch.stop) def test_other_candidate_cannot_replace_pending_release(self): release.reserve(SHA) with self.assertRaisesRegex(ValueError, 'Another release'): release.reserve('c' * 40) def test_fixed_notes_and_candidate_are_required_on_retry(self): release.reserve(SHA) notes = Path(self.directory.name) / 'notes.md' notes.write_text('English\n\n中文') with patch.object(release, 'version_for', return_value='20261030-aaaaaaaa'): release.select_release(SHA, notes) with patch.dict(os.environ, {'RELEASE_ID': SHA, 'RELEASE_COMMIT': SHA}): self.assertEqual(release.selected_state()['notes'], notes.read_text()) with patch.dict(os.environ, {'RELEASE_COMMIT': 'b' * 40}): with self.assertRaisesRegex(ValueError, 'reserved candidate'): release.selected_state() def test_release_matrix_includes_windows_arm64(self): self.assertEqual(set(release.PLATFORMS), {'linux/amd64', 'linux/arm64', 'darwin/amd64', 'darwin/arm64', 'windows/amd64', 'windows/arm64'}) def test_release_version_uses_utc_commit_date_and_fixed_sha(self): with patch.object(release, 'run', return_value='0'): self.assertEqual(release.version_for(SHA), '19700101-aaaaaaaa')