feat(perception): stage TRAVEL qualification
This commit is contained in:
@@ -0,0 +1,61 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import hashlib
|
||||
import importlib.util
|
||||
import json
|
||||
import tarfile
|
||||
from pathlib import Path
|
||||
|
||||
REPOSITORY_ROOT = Path(__file__).resolve().parents[1]
|
||||
BUILDER_PATH = REPOSITORY_ROOT / "scripts/build_m49_t3_travel_worker_artifact.py"
|
||||
SPEC = importlib.util.spec_from_file_location("m49_t3_travel_builder", BUILDER_PATH)
|
||||
assert SPEC is not None and SPEC.loader is not None
|
||||
BUILDER = importlib.util.module_from_spec(SPEC)
|
||||
SPEC.loader.exec_module(BUILDER)
|
||||
|
||||
|
||||
def _sha256(value: bytes) -> str:
|
||||
return hashlib.sha256(value).hexdigest()
|
||||
|
||||
|
||||
def _regular_files(archive: tarfile.TarFile) -> dict[str, bytes]:
|
||||
result: dict[str, bytes] = {}
|
||||
for member in archive.getmembers():
|
||||
if not member.isfile():
|
||||
continue
|
||||
stream = archive.extractfile(member)
|
||||
assert stream is not None
|
||||
result[member.name] = stream.read()
|
||||
return result
|
||||
|
||||
|
||||
def test_m49_t3_worker_artifact_is_deterministic_and_bounded(tmp_path: Path) -> None:
|
||||
patch_id = "mission-core-m49-t3-travel-unit-001"
|
||||
revision = "a" * 40
|
||||
first = BUILDER.build(patch_id, tmp_path / "first", revision=revision)
|
||||
second = BUILDER.build(patch_id, tmp_path / "second", revision=revision)
|
||||
|
||||
first_bytes = Path(first["artifact"]).read_bytes()
|
||||
assert first_bytes == Path(second["artifact"]).read_bytes()
|
||||
assert first["sha256"] == _sha256(first_bytes)
|
||||
with tarfile.open(first["artifact"], "r:gz") as archive:
|
||||
members = archive.getmembers()
|
||||
regular = _regular_files(archive)
|
||||
assert all(not member.issym() and not member.islnk() for member in members)
|
||||
assert set(regular) == {
|
||||
"manifest.env",
|
||||
"files.txt",
|
||||
*(f"payload/{name}" for name in first["payload_files"]),
|
||||
}
|
||||
assert regular["files.txt"].decode().splitlines() == first["payload_files"]
|
||||
release = json.loads(regular["payload/release.json"])
|
||||
assert release["code_revision"] == revision
|
||||
assert release["candidate_id"] == "travel"
|
||||
assert release["license"] == "GPL-3.0-or-later"
|
||||
assert release["authority"] == {
|
||||
"navigation_or_actuation_allowed": False,
|
||||
"ravnoves00_quality_accepted": False,
|
||||
}
|
||||
serialized = json.dumps(release).lower()
|
||||
assert "password=" not in serialized
|
||||
assert "private key" not in serialized
|
||||
Reference in New Issue
Block a user