test_muse_artifact.py
file-level
1
files
1
commits
0
hotspots
0
🧊 dead
0
💥 blast risk
| 1 | """Recovery packaging guards run in the Git-only environment without Muse.""" |
| 2 | import base64 |
| 3 | import csv |
| 4 | import hashlib |
| 5 | import io |
| 6 | import json |
| 7 | import os |
| 8 | from pathlib import Path |
| 9 | import zipfile |
| 10 | |
| 11 | import pytest |
| 12 | |
| 13 | from tools.ci import muse_artifact as artifact |
| 14 | |
| 15 | |
| 16 | @pytest.fixture |
| 17 | def payload(tmp_path): |
| 18 | files = {'muse/__init__.py': b'VERSION = "0.2.1rc5"\n', |
| 19 | 'muse/data.bin': b'\x00\xffpayload', |
| 20 | artifact.DIST + '/METADATA': b'Name: muse\nVersion: 0.2.1rc5\n', |
| 21 | artifact.DIST + '/WHEEL': b'original generator\n'} |
| 22 | manifest = {n: {'sha256': artifact.sha(b), 'size': len(b)} for n, b in files.items()} |
| 23 | source = hashlib.sha256(b'__init__.py\0' + files['muse/__init__.py'] + b'\0').hexdigest() |
| 24 | lock = {'origin': 'Local recovery, not an upstream release', |
| 25 | 'package_source_sha256': source, |
| 26 | 'payload_manifest_sha256': artifact.sha(artifact.json_bytes(manifest)), |
| 27 | 'original_sdist_sha256': '1' * 64} |
| 28 | for name, raw in files.items(): |
| 29 | path = tmp_path / name |
| 30 | path.parent.mkdir(parents=True, exist_ok=True) |
| 31 | path.write_bytes(raw) |
| 32 | return tmp_path, files, manifest, lock |
| 33 | |
| 34 | |
| 35 | def test_rebuild_independent_of_order_time_mode_and_roundtrip(payload): |
| 36 | root, files, manifest, lock = payload |
| 37 | expected = artifact.artifacts(files, lock) |
| 38 | for name in files: |
| 39 | (root / name).chmod(0o600) |
| 40 | os.utime(root / name, (1_800_000_000, 1_800_000_000)) |
| 41 | actual = artifact.read_installed(root, manifest) |
| 42 | artifact.validate_payload(actual, manifest, lock['package_source_sha256']) |
| 43 | assert artifact.artifacts(dict(reversed(list(actual.items()))), lock) == expected |
| 44 | recovered = artifact.read_archive(expected[artifact.SOURCE_NAME], manifest) |
| 45 | assert artifact.artifacts(recovered, lock) == expected |
| 46 | with zipfile.ZipFile(io.BytesIO(expected[artifact.WHEEL_NAME])) as wheel: |
| 47 | provenance = json.loads(wheel.read(artifact.DIST + '/OVERSEER_RECOVERY.json')) |
| 48 | assert provenance['upstream_release'] is False |
| 49 | assert provenance['original_sdist_recovered'] is False |
| 50 | assert provenance['recovery_source_archive_sha256'] == artifact.sha(expected[artifact.SOURCE_NAME]) |
| 51 | for name, digest, size in csv.reader(io.StringIO(wheel.read(artifact.DIST + '/RECORD').decode())): |
| 52 | if name.endswith('/RECORD'): |
| 53 | assert digest == size == '' |
| 54 | else: |
| 55 | raw = wheel.read(name) |
| 56 | assert int(size) == len(raw) |
| 57 | assert digest == 'sha256=' + base64.urlsafe_b64encode(hashlib.sha256(raw).digest()).decode().rstrip('=') |
| 58 | |
| 59 | |
| 60 | @pytest.mark.parametrize('path', ['muse/__init__.py', 'muse/data.bin', artifact.DIST + '/METADATA']) |
| 61 | def test_changed_source_data_or_metadata_refuses(payload, path): |
| 62 | _, files, manifest, lock = payload |
| 63 | files[path] += b'changed' |
| 64 | with pytest.raises(ValueError, match='payload differs'): |
| 65 | artifact.validate_payload(files, manifest, lock['package_source_sha256']) |
| 66 | |
| 67 | |
| 68 | def test_added_installed_code_refuses_but_generated_bytecode_is_ignored(payload): |
| 69 | root, files, manifest, _ = payload |
| 70 | (root / 'muse/cache.pyc').write_bytes(b'generated') |
| 71 | assert artifact.read_installed(root, manifest) == files |
| 72 | (root / 'muse/extra.py').write_bytes(b'unreviewed') |
| 73 | with pytest.raises(ValueError, match='extra payload'): |
| 74 | artifact.read_installed(root, manifest) |
| 75 | |
| 76 | |
| 77 | def test_linked_input_refuses(payload): |
| 78 | root, _, manifest, _ = payload |
| 79 | (root / 'muse/data.bin').unlink() |
| 80 | (root / 'muse/data.bin').symlink_to('__init__.py') |
| 81 | with pytest.raises(ValueError, match='linked input'): |
| 82 | artifact.read_installed(root, manifest) |
| 83 | |
| 84 | |
| 85 | def test_source_pin_drift_refuses(payload): |
| 86 | _, files, manifest, _ = payload |
| 87 | with pytest.raises(ValueError, match='source differs'): |
| 88 | artifact.validate_payload(files, manifest, '0' * 64) |
| 89 | |
| 90 | |
| 91 | def test_duplicate_archive_entries_refuse(payload): |
| 92 | _, files, manifest, _ = payload |
| 93 | raw = io.BytesIO(artifact.archive_bytes(files)) |
| 94 | with zipfile.ZipFile(raw, 'a') as archive, pytest.warns(UserWarning): |
| 95 | archive.writestr('muse/data.bin', b'duplicate') |
| 96 | with pytest.raises(ValueError, match='paths differ'): |
| 97 | artifact.read_archive(raw.getvalue(), manifest) |
| 98 | |
| 99 | |
| 100 | def test_manifest_tamper_refuses(tmp_path, monkeypatch): |
| 101 | (tmp_path / 'muse-rc5-artifact.json').write_text(json.dumps({'payload_manifest_sha256': '0' * 64})) |
| 102 | (tmp_path / 'muse-rc5-source-files.json').write_text('{}') |
| 103 | monkeypatch.setattr(artifact, 'HERE', tmp_path) |
| 104 | with pytest.raises(ValueError, match='manifest digest'): |
| 105 | artifact.contract() |
| 106 | |
| 107 | |
| 108 | @pytest.mark.parametrize('changed', ['wheel', 'source']) |
| 109 | def test_cli_rejects_artifact_substitution_before_output(payload, monkeypatch, changed): |
| 110 | root, files, manifest, lock = payload |
| 111 | built = artifact.artifacts(files, lock) |
| 112 | lock['artifacts'] = {n: artifact.sha(b) for n, b in built.items()} |
| 113 | monkeypatch.setattr(artifact, 'contract', lambda: (lock, manifest)) |
| 114 | name = artifact.WHEEL_NAME if changed == 'wheel' else artifact.SOURCE_NAME |
| 115 | path = root / name |
| 116 | path.write_bytes(built[name] + b'tampered') |
| 117 | output = root / 'output' |
| 118 | args = (['verify', '--wheel', str(path)] if changed == 'wheel' else |
| 119 | ['rebuild', '--source-archive', str(path), '--output-dir', str(output)]) |
| 120 | monkeypatch.setattr('sys.argv', ['muse_artifact.py', *args]) |
| 121 | with pytest.raises(SystemExit) as error: |
| 122 | artifact.main() |
| 123 | assert error.value.code == 1 |
| 124 | assert not output.exists() |
| 125 | |
| 126 | |
| 127 | def test_committed_contract_matches_supported_runtime_pin(): |
| 128 | lock, manifest = artifact.contract() |
| 129 | assert len(manifest) == 388 |
| 130 | assert all(name.startswith(('muse/', artifact.DIST + '/')) for name in manifest) |
| 131 | # Keep the runtime's independent installed-source check aligned with the |
| 132 | # artifact pin while the Git-only job imports no Muse package. |
| 133 | from tools.ci.supported_v1 import MUSE_SOURCE_SHA256 |
| 134 | assert lock['package_source_sha256'] == MUSE_SOURCE_SHA256 |
| 135 | assert lock['artifacts'][artifact.WHEEL_NAME] != lock['r3_reconstructed_wheel_sha256'] |