File size: 5,656 Bytes
2569bec | 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 148 149 150 151 152 153 | import hashlib
import sys
import tempfile
import unittest
from collections.abc import Iterator
from contextlib import contextmanager
from pathlib import Path
from unittest.mock import MagicMock, patch
sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
from codex_package import v8
from codex_package.targets import TARGET_SPECS, TargetSpec
class FetchCodexV8ArtifactsTest(unittest.TestCase):
version = "150.4.0"
def setUp(self) -> None:
temporary = tempfile.TemporaryDirectory()
self.addCleanup(temporary.cleanup)
self.root = Path(temporary.name)
@contextmanager
def release(
self,
target: str,
*,
line_ending: bytes = b"\n",
trusted_digest: str | None = None,
trusted_name: str | None = None,
create_pins: bool = True,
) -> Iterator[tuple[TargetSpec, MagicMock, str]]:
spec = TARGET_SPECS[target]
profile = v8.V8_ARTIFACT_PROFILE
archive_name = (
f"rusty_v8_{profile}_{target}.lib.gz"
if spec.is_windows
else f"librusty_v8_{profile}_{target}.a.gz"
)
binding_name = f"src_binding_{profile}_{target}.rs"
manifest_name = f"rusty_v8_{profile}_{target}.sha256"
archive = b"trusted V8 archive"
binding = b"trusted V8 binding"
manifest = (
line_ending.join(
(
f"{hashlib.sha256(archive).hexdigest()} {archive_name}".encode(),
f"{hashlib.sha256(binding).hexdigest()} {binding_name}".encode(),
)
)
+ line_ending
)
payloads = {
manifest_name: manifest,
archive_name: archive,
binding_name: binding,
}
if create_pins:
pins = (
self.root / "third_party/v8/rusty_v8_150_4_0_release_manifests.sha256"
)
pins.parent.mkdir(parents=True)
digest = trusted_digest or hashlib.sha256(manifest).hexdigest()
name = trusted_name or manifest_name
pins.write_bytes(f"{digest} {name}".encode() + line_ending)
def download(_url: str, destination: Path) -> None:
destination.parent.mkdir(parents=True, exist_ok=True)
destination.write_bytes(payloads[destination.name])
with (
patch.object(v8, "REPO_ROOT", self.root),
patch.object(v8, "download_file", side_effect=download) as download_file,
):
yield spec, download_file, manifest_name
def test_fetches_artifacts_after_authenticating_manifest(self) -> None:
with self.release("x86_64-unknown-linux-gnu") as (
spec,
download,
manifest_name,
):
artifacts = v8.fetch_codex_v8_artifacts(
spec, version=self.version, cache_root=self.root / "cache"
)
self.assertEqual(artifacts.archive.read_bytes(), b"trusted V8 archive")
self.assertEqual(artifacts.binding.read_bytes(), b"trusted V8 binding")
self.assertEqual(download.call_args_list[0].args[1].name, manifest_name)
self.assertEqual(download.call_count, 3)
def test_authenticates_windows_manifest_with_crlf(self) -> None:
with self.release("x86_64-pc-windows-msvc", line_ending=b"\r\n") as (
spec,
download,
_manifest_name,
):
artifacts = v8.fetch_codex_v8_artifacts(
spec, version=self.version, cache_root=self.root / "cache"
)
self.assertEqual(artifacts.archive.read_bytes(), b"trusted V8 archive")
self.assertEqual(artifacts.binding.read_bytes(), b"trusted V8 binding")
self.assertEqual(download.call_count, 3)
def test_rejects_tampered_manifest_before_downloading_artifacts(self) -> None:
with self.release("x86_64-unknown-linux-gnu", trusted_digest="0" * 64) as (
spec,
download,
manifest_name,
):
with self.assertRaisesRegex(
RuntimeError, "does not match its trusted SHA-256"
):
v8.fetch_codex_v8_artifacts(
spec, version=self.version, cache_root=self.root / "cache"
)
download.assert_called_once()
self.assertEqual(download.call_args.args[1].name, manifest_name)
def test_rejects_missing_manifest_pin_before_downloading_artifacts(self) -> None:
with self.release(
"x86_64-unknown-linux-gnu", trusted_name="another-target.sha256"
) as (spec, download, manifest_name):
with self.assertRaisesRegex(RuntimeError, "has no trusted SHA-256"):
v8.fetch_codex_v8_artifacts(
spec, version=self.version, cache_root=self.root / "cache"
)
download.assert_called_once()
self.assertEqual(download.call_args.args[1].name, manifest_name)
def test_rejects_missing_pin_file_before_downloading_artifacts(self) -> None:
with self.release("x86_64-unknown-linux-gnu", create_pins=False) as (
spec,
download,
manifest_name,
):
with self.assertRaises(FileNotFoundError):
v8.fetch_codex_v8_artifacts(
spec, version=self.version, cache_root=self.root / "cache"
)
download.assert_called_once()
self.assertEqual(download.call_args.args[1].name, manifest_name)
if __name__ == "__main__":
unittest.main()
|