lhk229
705bb1e6b3
Project CI / AI game creator shell Rust lane 1/2 (push) Has been cancelled
Project CI / AI game creator shell Rust lane 2/2 (push) Has been cancelled
Project CI / AI game creator shell Rust smoke (push) Has been cancelled
Project CI / AI game creator shell Rust crates (push) Has been cancelled
Project CI / Backend tests (push) Has been cancelled
Project CI / Native shell tests (push) Has been cancelled
Project CI / Frontend tests (push) Has been cancelled
Project CI / Repository checks (push) Has been cancelled
Project CI / AI game creator shell web tests (push) Has been cancelled
Reviewed-on: https://git.genarrative.world/git/GenarrativeAI/Genarrative/pulls/462
241 lines
9.0 KiB
Python
241 lines
9.0 KiB
Python
#!/usr/bin/env python3
|
|
"""Tests for streamed Gitea sccache snapshot validation and merging."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import importlib.util
|
|
import io
|
|
import json
|
|
from pathlib import Path
|
|
import os
|
|
import sys
|
|
import tarfile
|
|
import tempfile
|
|
import unittest
|
|
import zipfile
|
|
|
|
|
|
SCRIPT = Path(__file__).with_name("gitea_cache_snapshot.py")
|
|
SPEC = importlib.util.spec_from_file_location("gitea_cache_snapshot", SCRIPT)
|
|
assert SPEC and SPEC.loader
|
|
snapshot = importlib.util.module_from_spec(SPEC)
|
|
sys.modules[SPEC.name] = snapshot
|
|
SPEC.loader.exec_module(snapshot)
|
|
|
|
REPOSITORY = "team/project"
|
|
SOURCE_SHA = "a" * 40
|
|
INHERITED_SOURCE_SHA = "c" * 40
|
|
BASE_IMAGE = "sha256:" + "b" * 64
|
|
RUSTC = "rustc 1.90.0\nbinary: rustc\n"
|
|
WORKSPACE = "/workspace/GenarrativeAI/Genarrative"
|
|
SCCACHE_VERSION = "0.18.0"
|
|
|
|
|
|
def object_path(character: str) -> str:
|
|
key = character * 64
|
|
return f"objects/{key[0]}/{key[1]}/{key}"
|
|
|
|
|
|
class SnapshotMergeTest(unittest.TestCase):
|
|
def setUp(self) -> None:
|
|
self.temporary_directory = tempfile.TemporaryDirectory(ignore_cleanup_errors=True)
|
|
self.root = Path(self.temporary_directory.name).resolve()
|
|
self.base = self.root / "base-objects"
|
|
self.base.mkdir()
|
|
|
|
def tearDown(self) -> None:
|
|
self.temporary_directory.cleanup()
|
|
|
|
def identity(self, job: str, run_id: int) -> object:
|
|
return snapshot.ArtifactIdentity(REPOSITORY, run_id, 1, job, SOURCE_SHA)
|
|
|
|
def archive(
|
|
self,
|
|
name: str,
|
|
job: str,
|
|
run_id: int,
|
|
objects: list[tuple[str, bytes, int]],
|
|
*,
|
|
prefix: bool = True,
|
|
manifest_update=None,
|
|
extra_member: tuple[str, bytes, str] | None = None,
|
|
touched: list[tuple[str, int]] | None = None,
|
|
) -> object:
|
|
rows = [
|
|
{
|
|
"path": path,
|
|
"size": len(contents),
|
|
"sha256": hashlib.sha256(contents).hexdigest(),
|
|
"mtime_ns": mtime_ns,
|
|
}
|
|
for path, contents, mtime_ns in objects
|
|
]
|
|
manifest = {
|
|
"schema": 1,
|
|
"repository": REPOSITORY,
|
|
"run_id": run_id,
|
|
"run_attempt": 1,
|
|
"job": job,
|
|
"source_sha": SOURCE_SHA,
|
|
"base_image": BASE_IMAGE,
|
|
"rustc": RUSTC,
|
|
"workspace": WORKSPACE,
|
|
"sccache_version": SCCACHE_VERSION,
|
|
"mode": "delta",
|
|
"inherited_source_sha": INHERITED_SOURCE_SHA,
|
|
"objects": rows,
|
|
"touched": [
|
|
{"path": path, "mtime_ns": mtime_ns}
|
|
for path, mtime_ns in (touched or [])
|
|
],
|
|
}
|
|
if manifest_update is not None:
|
|
manifest_update(manifest)
|
|
|
|
tar_bytes = io.BytesIO()
|
|
with tarfile.open(fileobj=tar_bytes, mode="w") as bundle:
|
|
for path, contents, _ in objects:
|
|
info = tarfile.TarInfo(path)
|
|
info.size = len(contents)
|
|
bundle.addfile(info, io.BytesIO(contents))
|
|
if extra_member is not None:
|
|
path, contents, kind = extra_member
|
|
info = tarfile.TarInfo(path)
|
|
if kind == "symlink":
|
|
info.type = tarfile.SYMTYPE
|
|
info.linkname = "manifest.json"
|
|
bundle.addfile(info)
|
|
else:
|
|
info.size = len(contents)
|
|
bundle.addfile(info, io.BytesIO(contents))
|
|
manifest_bytes = json.dumps(manifest, separators=(",", ":")).encode()
|
|
info = tarfile.TarInfo("manifest.json")
|
|
info.size = len(manifest_bytes)
|
|
bundle.addfile(info, io.BytesIO(manifest_bytes))
|
|
|
|
archive = self.root / f"{name}.zip"
|
|
member = f"{name}/snapshot.tar" if prefix else "snapshot.tar"
|
|
with zipfile.ZipFile(archive, "w", compression=zipfile.ZIP_STORED) as bundle:
|
|
bundle.writestr(member, tar_bytes.getvalue())
|
|
return snapshot.ArtifactInput(archive, self.identity(job, run_id))
|
|
|
|
def base_object(self, path: str, contents: bytes, mtime_ns: int) -> None:
|
|
target = self.base.joinpath(*Path(path).parts[1:])
|
|
target.parent.mkdir(parents=True, exist_ok=True)
|
|
target.write_bytes(contents)
|
|
os.utime(target, ns=(mtime_ns, mtime_ns))
|
|
|
|
def merge(self, inputs, output=None, maximum=1024 ** 3):
|
|
return snapshot.merge_snapshots(
|
|
inputs,
|
|
output or self.root / "merged",
|
|
base_objects=self.base,
|
|
expected_inherited_source_sha=INHERITED_SOURCE_SHA,
|
|
max_combined_bytes=maximum,
|
|
)
|
|
|
|
def test_merges_groups_deduplicates_newest_mtime_and_prunes_to_bound(self) -> None:
|
|
first_path = object_path("a")
|
|
second_path = object_path("b")
|
|
third_path = object_path("c")
|
|
inherited_path = object_path("d")
|
|
self.base_object(inherited_path, b"dd", 50)
|
|
inputs = [
|
|
self.archive("first", "Backend tests", 41, [
|
|
(first_path, b"aaaa", 100),
|
|
(second_path, b"bbbb", 200),
|
|
]),
|
|
self.archive("second", "Native shell tests", 41, [
|
|
(first_path, b"aaaa", 300),
|
|
(third_path, b"ccc", 250),
|
|
], prefix=False, touched=[(inherited_path, 400)]),
|
|
]
|
|
|
|
output = self.root / "merged"
|
|
result = self.merge(inputs, output, maximum=9)
|
|
|
|
self.assertEqual((result.object_count, result.total_bytes), (3, 9))
|
|
self.assertEqual((output / inherited_path).read_bytes(), b"dd")
|
|
self.assertEqual((output / first_path).read_bytes(), b"aaaa")
|
|
self.assertEqual((output / third_path).read_bytes(), b"ccc")
|
|
self.assertFalse((output / second_path).exists())
|
|
self.assertEqual((output / "rustc.txt").read_text(), RUSTC)
|
|
self.assertEqual((output / "workspace.txt").read_text(), WORKSPACE + "\n")
|
|
self.assertEqual((output / "source-commit.txt").read_text(), SOURCE_SHA + "\n")
|
|
self.assertEqual((output / "base-image.txt").read_text(), BASE_IMAGE + "\n")
|
|
self.assertFalse((output / "sccache").exists())
|
|
|
|
def test_rejects_traversal_member_without_creating_output(self) -> None:
|
|
item = self.archive(
|
|
"traversal", "Backend tests", 42, [],
|
|
extra_member=("../outside", b"bad", "file"),
|
|
)
|
|
output = self.root / "merged"
|
|
with self.assertRaisesRegex(snapshot.SnapshotError, "invalid sccache object path"):
|
|
self.merge([item], output)
|
|
self.assertFalse(output.exists())
|
|
self.assertFalse((self.root / "outside").exists())
|
|
|
|
def test_rejects_checksum_mismatch(self) -> None:
|
|
path = object_path("d")
|
|
|
|
def corrupt(manifest):
|
|
manifest["objects"][0]["sha256"] = "0" * 64
|
|
|
|
item = self.archive(
|
|
"checksum", "Backend tests", 43, [(path, b"content", 1)],
|
|
manifest_update=corrupt,
|
|
)
|
|
with self.assertRaisesRegex(snapshot.SnapshotError, "checksums do not match"):
|
|
self.merge([item])
|
|
|
|
def test_rejects_manifest_identity_mismatch(self) -> None:
|
|
item = self.archive(
|
|
"identity", "Backend tests", 44, [],
|
|
manifest_update=lambda manifest: manifest.update(run_id=999),
|
|
)
|
|
with self.assertRaisesRegex(snapshot.SnapshotError, "run_id does not match"):
|
|
self.merge([item])
|
|
|
|
def test_rejects_conflicting_duplicate_key(self) -> None:
|
|
path = object_path("e")
|
|
inputs = [
|
|
self.archive("conflict-one", "Backend tests", 45, [(path, b"one", 1)]),
|
|
self.archive("conflict-two", "Native shell tests", 45, [(path, b"two", 2)]),
|
|
]
|
|
with self.assertRaisesRegex(snapshot.SnapshotError, "conflicting content"):
|
|
self.merge(inputs)
|
|
|
|
def test_rejects_delta_that_conflicts_with_inherited_key(self) -> None:
|
|
path = object_path("f")
|
|
self.base_object(path, b"base", 1)
|
|
item = self.archive("base-conflict", "Backend tests", 46, [(path, b"evil", 2)])
|
|
with self.assertRaisesRegex(snapshot.SnapshotError, "conflicting content"):
|
|
self.merge([item])
|
|
|
|
def test_accepts_empty_delta(self) -> None:
|
|
item = self.archive("empty", "Backend tests", 47, [])
|
|
output = self.root / "merged"
|
|
result = self.merge([item], output)
|
|
self.assertEqual((result.object_count, result.total_bytes), (0, 0))
|
|
self.assertEqual(list((output / "objects").iterdir()), [])
|
|
|
|
def test_rejects_noncanonical_base_entry(self) -> None:
|
|
(self.base / "unexpected").write_bytes(b"not an object")
|
|
item = self.archive("base-path", "Backend tests", 48, [])
|
|
with self.assertRaisesRegex(snapshot.SnapshotError, "noncanonical entry"):
|
|
self.merge([item])
|
|
|
|
def test_rejects_tar_links(self) -> None:
|
|
item = self.archive(
|
|
"link", "Backend tests", 49, [],
|
|
extra_member=(object_path("9"), b"", "symlink"),
|
|
)
|
|
with self.assertRaisesRegex(snapshot.SnapshotError, "regular file"):
|
|
self.merge([item])
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|