"""Synthetic guards for this one fixed preview upload; no GitHub mutations.""" from copy import deepcopy from pathlib import Path import json import tempfile import unittest from unittest.mock import patch import upload_draft_release as upload class FrozenDraftUploadTests(unittest.TestCase): def test_release_guard_rejects_any_changed_identity_or_published_state(self): good = {'id': upload.RELEASE_ID, 'tag_name': upload.TAG, 'target_commitish': upload.COMMIT, 'draft': True, 'prerelease': True} self.assertEqual(upload.checked_release(good), good) for field, value in (('id', 1), ('tag_name', 'wrong'), ('target_commitish', 'main'), ('draft', False), ('prerelease', False)): bad = dict(good) bad[field] = value with self.subTest(field=field), self.assertRaises(ValueError): upload.checked_release(bad) def test_direct_and_annotated_tag_must_peel_to_exact_commit(self): direct = {'ref': 'refs/tags/' + upload.TAG, 'object': {'type': 'commit', 'sha': upload.COMMIT}} upload.checked_tag(direct) reference = {'ref': direct['ref'], 'object': {'type': 'tag', 'sha': 'a' * 40}} annotation = {'sha': 'a' * 40, 'tag': upload.TAG, 'object': direct['object']} upload.checked_tag(reference, annotation) for bad in (None, dict(annotation, sha='b' * 40), dict(annotation, object={'type': 'commit', 'sha': 'b' * 40})): with self.assertRaises(ValueError): upload.checked_tag(reference, bad) def test_remote_asset_requires_exact_hash_size_name_and_uploaded_state(self): good = {'name': 'one.zip', 'size': 42, 'state': 'uploaded', 'digest': 'sha256:' + 'a' * 64} upload.checked_remote(good, 'one.zip', 42, 'a' * 64) for field, value in (('name', 'two.zip'), ('size', 41), ('state', 'starter'), ('digest', None)): with self.subTest(field=field), self.assertRaises(ValueError): upload.checked_remote(dict(good, **{field: value}), 'one.zip', 42, 'a' * 64) def test_receipt_must_bind_same_plan_checkpoint_kind_and_every_payload(self): plan = {'snapshot': {'commit': upload.COMMIT}} asset = {'name': 'one.zip', 'kind': 'shared-data', 'files': [{'name': 'data/pic', 'sha256': 'b' * 64}]} good = {'status': 'ZIP_CRC_AND_COMPLETE_PAYLOAD_SHA256_PASS', 'asset': 'one.zip', 'kind': 'shared-data', 'checkpoint': plan['snapshot'], 'plan_sha256': 'c' * 64, 'payload_files': 1, 'payload_sha256': {'data/pic': 'b' * 64}, 'bytes': 42, 'sha256': 'd' * 64} upload.checked_receipt(plan, 'c' * 64, asset, good) for field, value in (('checkpoint', {}), ('plan_sha256', 'a' * 64), ('kind', 'linux'), ('payload_files', 0), ('payload_sha256', {}), ('bytes', 2 * 1024**3), ('sha256', 'x')): with self.subTest(field=field), self.assertRaises(ValueError): upload.checked_receipt(plan, 'c' * 64, asset, dict(good, **{field: value})) def test_remote_members_rejects_unplanned_missing_or_duplicate_assets(self): controller = object.__new__(upload.DraftExporter) controller.verified = {'one.zip': {'bytes': 42, 'sha256': 'a' * 64}} asset = {'name': 'one.zip', 'size': 42, 'state': 'uploaded', 'digest': 'sha256:' + 'a' * 64} controller.remote_members({'assets': [asset]}) for assets in ([], [asset, asset], [dict(asset, name='extra.zip')]): with self.assertRaises(ValueError): controller.remote_members({'assets': assets}) def test_upload_failure_or_remote_digest_mismatch_never_deletes_local_zip(self): with tempfile.TemporaryDirectory() as temp: root = Path(temp) package = root / 'frozen' package.mkdir() controller = upload.DraftExporter(package, root / 'state', root / 'single') archive = root / 'single/one.zip' archive.write_bytes(b'complete ZIP fixture') expected = upload.digest(archive) before = {'assets': []} wrong = {'id': 123, 'name': 'one.zip', 'size': archive.stat().st_size, 'state': 'uploaded', 'digest': 'sha256:' + '0' * 64} with patch.object(controller, 'snapshot'), patch.object(controller, 'release', return_value=before), \ patch.object(controller, 'log'), patch.object(upload.subprocess, 'run') as run: run.return_value.returncode = 0 run.return_value.stdout = json.dumps(wrong) run.return_value.stderr = '' with self.assertRaises(ValueError): controller.upload(archive, 'one.zip', expected, archive.stat().st_size) self.assertTrue(archive.exists()) self.assertEqual(controller.verified, {}) run.return_value.returncode = 1 with self.assertRaises(RuntimeError): controller.upload(archive, 'one.zip', expected, archive.stat().st_size) self.assertTrue(archive.exists()) def test_output_directories_must_be_new_and_outside_frozen_tree(self): with tempfile.TemporaryDirectory() as temp: root = Path(temp) package = root / 'frozen' package.mkdir() with self.assertRaises(ValueError): upload.DraftExporter(package, package / 'state', root / 'single') state = root / 'already-existing-state' state.mkdir() with self.assertRaises(FileExistsError): upload.DraftExporter(package, state, root / 'single') class ConfigurableReleaseTargetTests(unittest.TestCase): def test_stable_and_preview_guards_require_the_explicit_expected_state(self): stable = upload.ReleaseTarget('c' * 40, 'v1.0.0', 12345, False) release = {'id': 12345, 'tag_name': 'v1.0.0', 'target_commitish': 'c' * 40, 'draft': True, 'prerelease': False} upload.checked_release(release, stable) with self.assertRaises(ValueError): upload.checked_release(dict(release, prerelease=True), stable) with self.assertRaises(ValueError): upload.checked_release(release) # Legacy preview defaults cannot match stable. reference = {'ref': 'refs/tags/v1.0.0', 'object': {'type': 'commit', 'sha': 'c' * 40}} upload.checked_tag(reference, target=stable) with self.assertRaises(ValueError): upload.checked_tag(dict(reference, object={'type': 'commit', 'sha': upload.COMMIT}), target=stable) def test_invalid_exact_release_inputs_fail_before_any_directory_or_upload(self): for values in (("main", 'v1.0.0', 123, False), ('A' * 40, 'v1.0.0', 123, False), ('c' * 40, '../v1', 123, False), ('c' * 40, 'v1/other', 123, False), ('c' * 40, 'v1.0.0', 0, False), ('c' * 40, 'v1.0.0', True, False), ('c' * 40, 'v1.0.0', 123, 'false')): with self.subTest(values=values), self.assertRaises(ValueError): upload.ReleaseTarget(*values) def test_explicit_release_routes_api_and_upload_to_its_exact_id_and_tag(self): target = upload.ReleaseTarget('c' * 40, 'v1.0.0', 12345, False) with tempfile.TemporaryDirectory() as temp: root = Path(temp) package = root / 'frozen' package.mkdir() controller = upload.DraftExporter(package, root / 'state', root / 'single', target) reference = {'ref': 'refs/tags/v1.0.0', 'object': {'type': 'commit', 'sha': 'c' * 40}} release = {'id': 12345, 'tag_name': 'v1.0.0', 'target_commitish': 'c' * 40, 'draft': True, 'prerelease': False, 'assets': []} with patch.object(controller, 'api', side_effect=[reference, release]) as api: self.assertEqual(controller.release(), release) self.assertEqual(api.call_args_list[0].args[0], 'repos/Quintine/RustRaptor/git/ref/tags/v1.0.0') self.assertEqual(api.call_args_list[1].args[0], 'repos/Quintine/RustRaptor/releases/12345') archive = root / 'single/one.zip' archive.write_bytes(b'fixture') wrong = {'name': 'one.zip', 'size': archive.stat().st_size, 'state': 'uploaded', 'digest': 'sha256:' + '0' * 64} with patch.object(controller, 'snapshot'), patch.object(controller, 'release', return_value=release), \ patch.object(controller, 'log'), patch.object(upload.subprocess, 'run') as run: run.return_value.returncode = 0 run.return_value.stdout = json.dumps(wrong) run.return_value.stderr = '' with self.assertRaises(ValueError): controller.upload(archive, 'one.zip', upload.digest(archive), archive.stat().st_size) self.assertIn('/releases/12345/assets?name=one.zip', run.call_args.args[0][4]) self.assertTrue(archive.exists()) if __name__ == '__main__': unittest.main()