563 lines
24 KiB
Python
563 lines
24 KiB
Python
from __future__ import annotations
|
|
|
|
import copy
|
|
from fractions import Fraction
|
|
import hashlib
|
|
import json
|
|
from pathlib import Path
|
|
import struct
|
|
import unittest
|
|
|
|
|
|
ROOT = Path(__file__).resolve().parents[2]
|
|
FIXTURE = ROOT / "tests/fixtures/ignored-workspace-v1"
|
|
EXPECTED_GIT_COMPATIBILITY = {
|
|
"status-diff": "supported",
|
|
"ordinary-add-reset-commit": "supported-host-backed-paths",
|
|
"checkout-restore": "supported",
|
|
"branch-reference-update": "supported",
|
|
"merge-rebase-stash": "supported",
|
|
"linked-worktree": "supported",
|
|
"nested-repository": "supported",
|
|
"force-add-isolated-only": "EBUSY-before-index-publication",
|
|
"host-config-write": "EROFS",
|
|
"default-hook-host-access": "sandbox-transient-only",
|
|
}
|
|
EXPECTED_FORCE_ADD_CASE = {
|
|
"id": "guest-force-add-isolated-only",
|
|
"tags": ["ignored-to-tracked-transition", "git-add", "no-silent-loss"],
|
|
"path_hex": "7461726765742f617070",
|
|
"setup": ["isolated-object", "path-ignored", "host-path-absent"],
|
|
"actions": ["sandbox-git-add-force"],
|
|
"expect": [
|
|
"EBUSY-before-index-publication",
|
|
"prior-index-retained",
|
|
"isolated-object-visible",
|
|
"host-path-absent",
|
|
],
|
|
}
|
|
|
|
|
|
def load(name: str) -> dict:
|
|
with (FIXTURE / name).open("rb") as source:
|
|
return json.load(source)
|
|
|
|
|
|
def odd_median(values: list[int | Fraction]) -> int | Fraction:
|
|
if not values or len(values) % 2 == 0:
|
|
raise ValueError("median input must have odd length")
|
|
ordered = sorted(values)
|
|
return ordered[len(ordered) // 2]
|
|
|
|
|
|
def relative_mad(values: list[int | Fraction]) -> Fraction:
|
|
center = odd_median(values)
|
|
if center == 0:
|
|
raise ValueError("relative MAD center must be positive")
|
|
return Fraction(odd_median([abs(value - center) for value in values]), center)
|
|
|
|
|
|
def median_spread(run_medians: list[int]) -> Fraction:
|
|
center = odd_median(run_medians)
|
|
if center == 0:
|
|
raise ValueError("median spread center must be positive")
|
|
return Fraction(max(run_medians) - min(run_medians), center)
|
|
|
|
|
|
def fixture_identity(records: list[tuple[bytes, str, int, bytes]]) -> bytes:
|
|
encoding = load("benchmark.json")["identity_encoding"]["fixture"]
|
|
kinds = encoding["kind_tags"]
|
|
ordered = sorted(records, key=lambda record: record[0])
|
|
if len({record[0] for record in ordered}) != len(ordered):
|
|
raise ValueError("duplicate fixture path")
|
|
stream = bytearray.fromhex(encoding["context_hex"])
|
|
stream.extend(struct.pack(">I", len(ordered)))
|
|
for path, kind, mode, payload in ordered:
|
|
components = path.split(b"/")
|
|
if (
|
|
not path
|
|
or path.startswith(b"/")
|
|
or path.endswith(b"/")
|
|
or b"\x00" in path
|
|
or any(component in (b"", b".", b"..") for component in components)
|
|
):
|
|
raise ValueError("invalid fixture path")
|
|
if kind not in kinds:
|
|
raise ValueError("unsupported fixture kind")
|
|
if kind == "directory" and payload:
|
|
raise ValueError("directory fixture payload must be empty")
|
|
size = 0 if kind == "directory" else len(payload)
|
|
stream.extend(struct.pack(">I", len(path)))
|
|
stream.extend(path)
|
|
stream.extend(struct.pack(">B", kinds[kind]))
|
|
stream.extend(struct.pack(">I", mode & 0o7777))
|
|
stream.extend(struct.pack(">Q", size))
|
|
stream.extend(hashlib.sha256(payload).digest())
|
|
return hashlib.sha256(stream).digest()
|
|
|
|
|
|
def result_identity(
|
|
workload: str,
|
|
initial_fixture: bytes,
|
|
commands: list[tuple[int, bytes, bytes]],
|
|
post_fixture: bytes,
|
|
) -> bytes:
|
|
encoding = load("benchmark.json")["identity_encoding"]["result"]
|
|
workload_bytes = workload.encode("utf-8")
|
|
if not workload_bytes or len(initial_fixture) != 32 or len(post_fixture) != 32:
|
|
raise ValueError("invalid result identity input")
|
|
stream = bytearray.fromhex(encoding["context_hex"])
|
|
stream.extend(struct.pack(">I", len(workload_bytes)))
|
|
stream.extend(workload_bytes)
|
|
stream.extend(initial_fixture)
|
|
stream.extend(struct.pack(">I", len(commands)))
|
|
for exit_status, stdout, stderr in commands:
|
|
stream.extend(struct.pack(">iQ", exit_status, len(stdout)))
|
|
stream.extend(stdout)
|
|
stream.extend(struct.pack(">Q", len(stderr)))
|
|
stream.extend(stderr)
|
|
stream.extend(post_fixture)
|
|
return hashlib.sha256(stream).digest()
|
|
|
|
|
|
def validate_semantic_consistency(contract: dict, corpus: dict) -> None:
|
|
compatibility = {
|
|
operation["operation"]: operation["support"]
|
|
for operation in contract["git_compatibility"]
|
|
}
|
|
if compatibility != EXPECTED_GIT_COMPATIBILITY:
|
|
raise ValueError("Git compatibility differs from the selected contract")
|
|
transitions = {
|
|
transition["id"]: transition
|
|
for transition in contract["transition_contracts"]
|
|
}
|
|
guest_publication = transitions["isolated-to-tracked-guest-publication"]
|
|
if guest_publication != {
|
|
"id": "isolated-to-tracked-guest-publication",
|
|
"trigger": "guest index publication tracks an isolated-only path",
|
|
"result": "EBUSY",
|
|
"index_mutation": "none",
|
|
"isolated_state": "retained-visible",
|
|
"copy": "none",
|
|
}:
|
|
raise ValueError("guest force-add transition differs from the selected contract")
|
|
cases = {case["id"]: case for case in corpus["cases"]}
|
|
if cases["guest-force-add-isolated-only"] != EXPECTED_FORCE_ADD_CASE:
|
|
raise ValueError("guest force-add corpus differs from the selected contract")
|
|
|
|
|
|
class IgnoredWorkspaceContractTests(unittest.TestCase):
|
|
def test_contract_dispatch_and_selected_mechanism_are_exact(self) -> None:
|
|
contract = load("contract.json")
|
|
self.assertEqual(contract["schema"], "lazaret-ignored-workspace-contract-v1")
|
|
self.assertEqual(contract["mechanism"], "rootless-host-fuse-router")
|
|
self.assertEqual(contract["route_classes"], ["host", "isolated"])
|
|
|
|
def test_cross_route_namespace_operations_are_directionally_bounded(self) -> None:
|
|
transitions = {
|
|
transition["id"]: transition
|
|
for transition in load("contract.json")["transition_contracts"]
|
|
}
|
|
for transition_id in (
|
|
"cross-route-rename",
|
|
"rename-whiteout",
|
|
"cross-route-hard-link",
|
|
):
|
|
with self.subTest(transition_id=transition_id):
|
|
self.assertEqual(transitions[transition_id]["mutation"], "none")
|
|
self.assertEqual(transitions["cross-route-rename"]["result"], "EXDEV")
|
|
self.assertEqual(
|
|
transitions["qualified-host-to-isolated-rename"],
|
|
{
|
|
"id": "qualified-host-to-isolated-rename",
|
|
"operations": ["rename", "rename-noreplace"],
|
|
"result": "success",
|
|
"mutation": "durable-host-to-isolated-migration",
|
|
"copy": "host-to-isolated-only",
|
|
},
|
|
)
|
|
self.assertEqual(transitions["cross-route-hard-link"]["result"], "EXDEV")
|
|
self.assertEqual(transitions["rename-whiteout"]["result"], "EOPNOTSUPP")
|
|
|
|
def test_transitions_copy_only_toward_isolated_state_between_generations(self) -> None:
|
|
transitions = load("contract.json")["transition_contracts"]
|
|
copying = [
|
|
transition["id"]
|
|
for transition in transitions
|
|
if "copy" in transition and transition["copy"] != "none"
|
|
]
|
|
self.assertEqual(
|
|
copying,
|
|
["qualified-host-to-isolated-rename", "host-to-ignored-policy"],
|
|
)
|
|
handle_transitions = {
|
|
transition["id"]: transition["open_handles"]
|
|
for transition in transitions
|
|
if "open_handles" in transition
|
|
}
|
|
self.assertEqual(
|
|
handle_transitions,
|
|
{},
|
|
)
|
|
|
|
def test_isolated_only_force_add_is_explicitly_refused(self) -> None:
|
|
contract = load("contract.json")
|
|
transitions = {
|
|
transition["id"]: transition
|
|
for transition in contract["transition_contracts"]
|
|
}
|
|
transition = transitions["isolated-to-tracked-guest-publication"]
|
|
self.assertEqual(transition["result"], "EBUSY")
|
|
self.assertEqual(transition["index_mutation"], "none")
|
|
compatibility = {
|
|
operation["operation"]: operation["support"]
|
|
for operation in contract["git_compatibility"]
|
|
}
|
|
self.assertEqual(
|
|
compatibility["force-add-isolated-only"],
|
|
"EBUSY-before-index-publication",
|
|
)
|
|
|
|
def test_transition_matrix_is_exactly_the_selected_contract(self) -> None:
|
|
transitions = load("contract.json")["transition_contracts"]
|
|
self.assertEqual(
|
|
transitions,
|
|
[
|
|
{
|
|
"id": "qualified-host-to-isolated-rename",
|
|
"operations": ["rename", "rename-noreplace"],
|
|
"result": "success",
|
|
"mutation": "durable-host-to-isolated-migration",
|
|
"copy": "host-to-isolated-only",
|
|
},
|
|
{
|
|
"id": "cross-route-rename",
|
|
"operations": ["rename", "rename-noreplace", "rename-exchange"],
|
|
"result": "EXDEV",
|
|
"mutation": "none",
|
|
},
|
|
{
|
|
"id": "rename-whiteout",
|
|
"operations": ["rename-whiteout"],
|
|
"result": "EOPNOTSUPP",
|
|
"mutation": "none",
|
|
},
|
|
{
|
|
"id": "cross-route-hard-link",
|
|
"operations": ["link"],
|
|
"result": "EXDEV",
|
|
"mutation": "none",
|
|
},
|
|
{
|
|
"id": "ignored-to-host-policy",
|
|
"trigger": "next Domain initialization classifies an isolated path as host-backed",
|
|
"new_lookups": "host-or-absent",
|
|
"isolated_state": "deleted-before-readiness",
|
|
"copy": "none",
|
|
},
|
|
{
|
|
"id": "host-to-ignored-policy",
|
|
"trigger": "next Domain initialization classifies a host-backed path as ignored",
|
|
"new_lookups": "host-clone-or-absent",
|
|
"host_state": "retained-unchanged",
|
|
"isolated_state": "host-clone-replaces-stale-state",
|
|
"copy": "host-to-isolated-only",
|
|
},
|
|
{
|
|
"id": "isolated-to-tracked-guest-publication",
|
|
"trigger": "guest index publication tracks an isolated-only path",
|
|
"result": "EBUSY",
|
|
"index_mutation": "none",
|
|
"isolated_state": "retained-visible",
|
|
"copy": "none",
|
|
},
|
|
],
|
|
)
|
|
|
|
def test_git_compatibility_is_exactly_the_selected_contract(self) -> None:
|
|
compatibility = {
|
|
operation["operation"]: operation["support"]
|
|
for operation in load("contract.json")["git_compatibility"]
|
|
}
|
|
self.assertEqual(
|
|
compatibility,
|
|
EXPECTED_GIT_COMPATIBILITY,
|
|
)
|
|
|
|
def test_operation_matrix_covers_required_surface(self) -> None:
|
|
operations = {
|
|
operation["operation"]: operation
|
|
for operation in load("contract.json")["filesystem_operations"]
|
|
}
|
|
required = {
|
|
"lookup-getattr-readdir",
|
|
"create-mkdir-open-read-write-truncate-fallocate-fsync",
|
|
"rename",
|
|
"rename-noreplace",
|
|
"rename-exchange",
|
|
"rename-whiteout",
|
|
"hard-link",
|
|
"symlink-readlink",
|
|
"user-xattrs",
|
|
"posix-acls",
|
|
"sparse-seek-hole-data-punch-hole",
|
|
"copy-file-range",
|
|
"utimens-nanoseconds",
|
|
"mmap-shared-private",
|
|
"flock-posix-record-lock",
|
|
"fifo-unix-socket",
|
|
"character-block-device-create",
|
|
}
|
|
self.assertTrue(required.issubset(operations))
|
|
self.assertEqual(
|
|
operations["rename"]["cross_route"],
|
|
"qualified-host-to-isolated-or-EXDEV",
|
|
)
|
|
self.assertEqual(operations["hard-link"]["cross_route"], "EXDEV")
|
|
self.assertEqual(operations["copy-file-range"]["cross_route"], "EXDEV")
|
|
self.assertEqual(operations["character-block-device-create"]["host"], "EPERM")
|
|
|
|
def test_corpus_ids_and_byte_paths_are_canonical(self) -> None:
|
|
cases = load("corpus.json")["cases"]
|
|
ids = [case["id"] for case in cases]
|
|
self.assertEqual(len(ids), len(set(ids)))
|
|
for case in cases:
|
|
with self.subTest(case=case["id"]):
|
|
path_hex = case["path_hex"]
|
|
self.assertEqual(path_hex, path_hex.lower())
|
|
self.assertEqual(bytes.fromhex(path_hex).hex(), path_hex)
|
|
self.assertNotIn(b"\x00", bytes.fromhex(path_hex))
|
|
non_utf8 = next(case for case in cases if case["id"] == "non-utf8-ignore-match")
|
|
with self.assertRaises(UnicodeDecodeError):
|
|
bytes.fromhex(non_utf8["path_hex"]).decode("utf-8")
|
|
|
|
def test_corpus_covers_every_required_behavior(self) -> None:
|
|
cases = load("corpus.json")["cases"]
|
|
tags = {tag for case in cases for tag in case["tags"]}
|
|
required = {
|
|
"non-utf8-names",
|
|
"ignore-precedence",
|
|
"core-ignore-case",
|
|
"worktrees",
|
|
"nested-repositories",
|
|
"mmap",
|
|
"locks",
|
|
"xattrs",
|
|
"acls",
|
|
"sparse",
|
|
"copy-range",
|
|
"rename",
|
|
"link",
|
|
"symlink",
|
|
"timestamps",
|
|
"special-files",
|
|
"concurrent-mutation",
|
|
"cross-policy-rename",
|
|
"ignored-to-tracked-transition",
|
|
"startup-sealed-policy",
|
|
"startup-policy-transition",
|
|
"host-to-isolated-copy",
|
|
"no-silent-loss",
|
|
"no-copy-back",
|
|
}
|
|
self.assertTrue(required.issubset(tags), sorted(required - tags))
|
|
|
|
def test_corpus_asserts_backing_state_for_refusal_and_loss_cases(self) -> None:
|
|
cases = {case["id"]: case for case in load("corpus.json")["cases"]}
|
|
self.assertEqual(
|
|
cases["cross-route-file-rename"]["expect"],
|
|
["EXDEV", "source-unchanged", "destination-absent"],
|
|
)
|
|
self.assertIn("isolated-object-deleted", cases["unmediated-inconsistent-index"]["expect"])
|
|
self.assertIn("access-error-not-host-fallback", cases["router-loss-fails-closed"]["expect"])
|
|
self.assertIn("host-sentinel-unchanged", cases["router-loss-fails-closed"]["expect"])
|
|
|
|
def test_force_add_corpus_is_exactly_fail_closed_without_copy_back(self) -> None:
|
|
cases = {case["id"]: case for case in load("corpus.json")["cases"]}
|
|
self.assertEqual(
|
|
cases["guest-force-add-isolated-only"],
|
|
EXPECTED_FORCE_ADD_CASE,
|
|
)
|
|
|
|
def test_reproduced_semantic_mutations_are_refused(self) -> None:
|
|
contract = load("contract.json")
|
|
corpus = load("corpus.json")
|
|
validate_semantic_consistency(contract, corpus)
|
|
|
|
unsupported_merge = copy.deepcopy(contract)
|
|
merge = next(
|
|
operation
|
|
for operation in unsupported_merge["git_compatibility"]
|
|
if operation["operation"] == "merge-rebase-stash"
|
|
)
|
|
merge["support"] = "unsupported"
|
|
with self.assertRaisesRegex(ValueError, "Git compatibility"):
|
|
validate_semantic_consistency(unsupported_merge, corpus)
|
|
|
|
copied_force_add = copy.deepcopy(corpus)
|
|
force_add = next(
|
|
case
|
|
for case in copied_force_add["cases"]
|
|
if case["id"] == "guest-force-add-isolated-only"
|
|
)
|
|
force_add["actions"] = [
|
|
"sandbox-git-add-force",
|
|
"copy-isolated-bytes-to-host",
|
|
"publish-index",
|
|
]
|
|
force_add["expect"] = ["success", "host-copy", "index-published"]
|
|
with self.assertRaisesRegex(ValueError, "force-add corpus"):
|
|
validate_semantic_consistency(contract, copied_force_add)
|
|
|
|
def test_benchmark_reference_conditions_are_complete(self) -> None:
|
|
qualification = load("benchmark.json")["qualification_environment"]
|
|
required = {
|
|
"architecture",
|
|
"kernel",
|
|
"cpu",
|
|
"memory",
|
|
"storage",
|
|
"filesystem",
|
|
"qualification_result_fields",
|
|
"qualification_result_policy",
|
|
}
|
|
self.assertEqual(set(qualification), required)
|
|
self.assertEqual(qualification["architecture"], "x86_64")
|
|
self.assertEqual(
|
|
set(qualification["cpu"]),
|
|
{"physical_cores_minimum", "allocation", "governor"},
|
|
)
|
|
self.assertGreaterEqual(qualification["cpu"]["physical_cores_minimum"], 8)
|
|
self.assertGreaterEqual(qualification["memory"]["bytes_minimum"], 16 * 1024**3)
|
|
self.assertEqual(
|
|
qualification["storage"],
|
|
{
|
|
"kind": "local-block-device",
|
|
"network_backed": False,
|
|
"copy_on_write": False,
|
|
},
|
|
)
|
|
self.assertEqual(qualification["filesystem"]["type"], "XFS")
|
|
self.assertEqual(qualification["filesystem"]["block_bytes"], 4096)
|
|
self.assertIn("cpu_model", qualification["qualification_result_fields"])
|
|
self.assertIn("storage_model", qualification["qualification_result_fields"])
|
|
|
|
def test_sampling_protocol_fixes_statistics_and_variance(self) -> None:
|
|
sampling = load("benchmark.json")["sampling"]
|
|
self.assertEqual(sampling["independent_runs"], 3)
|
|
self.assertEqual(sampling["warmup_samples_per_cell"], 2)
|
|
self.assertEqual(sampling["measured_samples_per_cell"], 15)
|
|
self.assertEqual(sampling["statistic"], "median")
|
|
self.assertEqual(
|
|
sampling["relative_statistic"],
|
|
"median of paired routed/direct elapsed-time ratios",
|
|
)
|
|
self.assertEqual(
|
|
sampling["direct_calibration_formula"],
|
|
"direct placement median <= absolute_median_ms / relative_median_ratio in every independent run and preparation state",
|
|
)
|
|
self.assertEqual(
|
|
sampling["within_run_relative_mad_formula"],
|
|
"median(abs(sample - median(samples))) / median(samples), evaluated separately for direct elapsed times, routed elapsed times, and paired routed/direct ratios in each workload, preparation state, and independent run",
|
|
)
|
|
self.assertEqual(
|
|
sampling["between_run_median_spread_formula"],
|
|
"(max(run_medians) - min(run_medians)) / median(run_medians), evaluated across the three independent-run medians separately for direct elapsed time, routed elapsed time, and paired routed/direct ratio in each workload and preparation state",
|
|
)
|
|
self.assertEqual(sampling["outlier_policy"], "retain every completed measured sample")
|
|
self.assertLessEqual(sampling["within_run_relative_mad_limit"], 0.10)
|
|
self.assertLessEqual(sampling["between_run_median_spread_limit"], 0.10)
|
|
self.assertEqual(sampling["budget_tolerance"], 0.0)
|
|
|
|
def test_variance_formulas_distinguish_the_reproduced_values(self) -> None:
|
|
values = [100, 110, 111]
|
|
self.assertEqual(relative_mad(values), Fraction(1, 110))
|
|
self.assertEqual(median_spread(values), Fraction(1, 10))
|
|
self.assertNotEqual(relative_mad(values), median_spread(values))
|
|
|
|
def test_identity_encoding_is_exactly_framed_and_ordered(self) -> None:
|
|
encoding = load("benchmark.json")["identity_encoding"]
|
|
self.assertEqual(
|
|
encoding,
|
|
{
|
|
"integer_encoding": "unsigned integers are fixed-width big-endian; the command exit status is signed two's-complement i32 big-endian",
|
|
"fixture": {
|
|
"context_hex": "6c617a617265742d69676e6f7265642d776f726b73706163652d666978747572652d763100",
|
|
"root_entry": "omitted",
|
|
"path": "raw relative pathname bytes with slash separators, no leading or trailing slash, and no NUL",
|
|
"ordering": "ascending lexicographic order of complete raw path byte strings",
|
|
"kind_tags": {"directory": 1, "regular": 2, "symlink": 3},
|
|
"mode": "u32 of st_mode masked with octal 7777",
|
|
"size": "u64 payload length for regular files and symlinks; zero for directories",
|
|
"payload": "regular file bytes, raw symlink target bytes, or empty bytes for a directory",
|
|
"record": "u32(path length) || path || u8(kind tag) || u32(mode) || u64(size) || SHA-256(payload)",
|
|
"stream": "context bytes || u32(record count) || records in declared ordering",
|
|
"identity": "SHA-256(stream)",
|
|
"unsupported_entry": "reject identity construction for any other file kind",
|
|
},
|
|
"result": {
|
|
"context_hex": "6c617a617265742d69676e6f7265642d776f726b73706163652d726573756c742d763100",
|
|
"workload": "u32(UTF-8 workload ID length) || UTF-8 workload ID",
|
|
"command_record": "i32(exit status) || u64(stdout length) || exact stdout bytes || u64(stderr length) || exact stderr bytes",
|
|
"ordering": "command records remain in command_sequence order",
|
|
"stream": "context bytes || framed workload || initial fixture identity || u32(command record count) || ordered command records || post-workload fixture identity",
|
|
"identity": "SHA-256(stream)",
|
|
"successful_sample": "every command exit status is zero and both fixture identities and every output byte are retained",
|
|
},
|
|
},
|
|
)
|
|
|
|
def test_fixture_identity_has_one_exact_canonical_encoding(self) -> None:
|
|
records = [
|
|
(b"link", "symlink", 0o777, b"a/\xff"),
|
|
(b"a/\xff", "regular", 0o640, b"x"),
|
|
(b"a", "directory", 0o755, b""),
|
|
]
|
|
self.assertEqual(
|
|
fixture_identity(records).hex(),
|
|
"72f539c740d0048772a2380969c2b22fa970f725958d17b7a534069e92d1994d",
|
|
)
|
|
self.assertEqual(fixture_identity(records), fixture_identity(list(reversed(records))))
|
|
with self.assertRaisesRegex(ValueError, "unsupported fixture kind"):
|
|
fixture_identity([(b"socket", "socket", 0o600, b"")])
|
|
|
|
def test_result_identity_frames_commands_outputs_and_fixture_states(self) -> None:
|
|
initial = bytes.fromhex("11" * 32)
|
|
post = bytes.fromhex("22" * 32)
|
|
commands = [(0, b"a", b""), (1, b"", b"b")]
|
|
self.assertEqual(
|
|
result_identity("git", initial, commands, post).hex(),
|
|
"0fe91ab5411628dc22cb2fb3161486535794c8044d74453c7ab7eafd1c93b4e7",
|
|
)
|
|
self.assertNotEqual(
|
|
result_identity("git", initial, list(reversed(commands)), post),
|
|
result_identity("git", initial, commands, post),
|
|
)
|
|
|
|
def test_each_workload_has_absolute_and_relative_budgets(self) -> None:
|
|
benchmark = load("benchmark.json")
|
|
workloads = benchmark["workloads"]
|
|
self.assertEqual(
|
|
{workload["category"] for workload in workloads},
|
|
{"metadata", "git", "package-manager", "build"},
|
|
)
|
|
for workload in workloads:
|
|
with self.subTest(workload=workload["id"]):
|
|
self.assertGreater(workload["absolute_median_ms"], 0)
|
|
self.assertGreater(workload["relative_median_ratio"], 1.0)
|
|
self.assertEqual(workload["preparation_states"], ["unprimed", "primed"])
|
|
self.assertEqual(len(workload["fixture_seed"]), 64)
|
|
self.assertTrue(workload["fixture_algorithm"])
|
|
self.assertTrue(workload["command_sequence"])
|
|
self.assertTrue(all(command for command in workload["command_sequence"]))
|
|
self.assertTrue(workload["result_oracle"])
|
|
|
|
def test_specification_names_the_committed_contract(self) -> None:
|
|
specification = (ROOT / "docs/specification.md").read_text()
|
|
self.assertIn("#### Ignored Workspace State routing contract", specification)
|
|
self.assertIn("`lazaret-ignored-workspace-contract-v1`", specification)
|
|
self.assertIn("rootless host FUSE filesystem", specification)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|