gsadmin/rustfs
1
0
Fork 0
mirror of https://github.com/rustfs/rustfs.git synced 2026-09-06 12:09:12 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(test): bind scanner evidence to execution and build identity

Browse Source 
Co-Authored-By: heihutu <heihutu@gmail.com>
Co-Authored-By: zhi22915 <qiuzgang@gmail.com>
This commit is contained in:
houseme
2026-09-05 17:05:59 +08:00
parent d49434fbe1
commit ab570c8133
5 changed files with 256 additions and 30 deletions
Download Patch File Download Diff File Expand all files Collapse all files
scripts/check_test_wiring.py
+132 -19
View File
@@ -877,6 +877,11 @@ def check_core_listing(root: Path, listing: Path) -> list[str]:
return [f"cannot read core nextest listing: {error}"]
def evidence_integer(value: object, name: str, minimum: int, maximum: int) -> int:
require(type(value) is int and minimum <= value <= maximum, f"invalid integer {name}")
return value
def begin_scanner_heal_receipt(root: Path, directory: Path, binary: Path, test_binary: Path) -> None:
"""Record an existing build; this command never builds or runs a test."""
require(not directory.exists(), "scanner/heal run directory must be new")
@@ -894,13 +899,18 @@ def begin_scanner_heal_receipt(root: Path, directory: Path, binary: Path, test_b
embedded_status = re.search(r"^git status\s*:\s*(.*)\Z", version, re.MULTILINE | re.DOTALL)
require(embedded_revision is not None and embedded_revision[1] == revision, "server binary source revision mismatch")
require(embedded_status is not None and not embedded_status[1].strip(), "server binary was built from dirty/unknown source")
sources = root / "crates/e2e_test/src"
test_sources_sha256 = hashlib.sha256((sources / "heal_erasure_disk_rebuild_test.rs").read_bytes() +
(sources / "chaos.rs").read_bytes()).hexdigest()
lock_blob = subprocess.check_output(["git", "hash-object", "Cargo.lock"], cwd=root, text=True).strip()
require(re.fullmatch(r"[0-9a-f]{40}", lock_blob), "invalid Cargo.lock identity")
features = os.environ.get("RUSTFS_E2E_EXPECTED_FEATURES")
require(features is not None, "set RUSTFS_E2E_EXPECTED_FEATURES to the compiled e2e crate feature set")
features = ",".join(sorted(set(filter(None, (feature.strip() for feature in features.split(","))))))
require(all(re.fullmatch(r"[a-z0-9-]+", feature) for feature in features.split(",") if feature), "invalid expected features")
directory.mkdir(parents=True)
write_json(directory / "run.json", {"schema": 1, "run_id": uuid.uuid4().hex,
"source_revision": revision,
"binary_source_revision": embedded_revision[1], "test_sources_sha256": test_sources_sha256,
"binary_source_revision": embedded_revision[1],
"test_build": {"source_revision": revision, "dirty": False,
"lock_blob": lock_blob, "features": features},
"started_at": datetime.now(timezone.utc).timestamp(), **builds})
@@ -925,14 +935,20 @@ def check_scanner_heal_evidence(root: Path, directory: Path, case_id: str) -> li
"""Validate one actual case, or fail the release while required lanes are pending."""
try:
registry = read_json(root / ".config/scanner-heal-required-tests.json")
require(registry.get("schema") == 1 and registry.get("cases"), "invalid scanner/heal registry")
evidence_integer(registry.get("schema"), "registry schema", 1, 1)
require(registry.get("cases"), "invalid scanner/heal registry")
selected = registry["cases"] if case_id == "release" else {case_id: registry["cases"][case_id]}
run = read_json(directory / "run.json")
execution = read_json(directory / "execution.json")
require(run.get("schema") == 1 and re.fullmatch(r"[0-9a-f]{32}", run["run_id"]), "invalid run identity")
evidence_integer(run.get("schema"), "run schema", 1, 1)
require(re.fullmatch(r"[0-9a-f]{32}", run["run_id"]), "invalid run identity")
require(re.fullmatch(r"[0-9a-f]{40}", run["source_revision"]), "invalid source revision")
require(run.get("binary_source_revision") == run["source_revision"], "server source provenance missing")
require(sha(run.get("test_sources_sha256")), "test source provenance missing")
expected_build = run["test_build"]
require(expected_build["source_revision"] == run["source_revision"] and expected_build["dirty"] is False,
"test source provenance missing")
require(re.fullmatch(r"[0-9a-f]{40}", expected_build["lock_blob"]), "invalid test lockfile identity")
require(isinstance(expected_build["features"], str), "missing test features")
require(execution.get("run_id") == run["run_id"], "execution belongs to another run")
require(type(execution.get("exit_code")) is int and execution["exit_code"] == 0, "test command failed or did not run")
number(run["started_at"], "started_at", 1)
@@ -957,25 +973,49 @@ def check_scanner_heal_evidence(root: Path, directory: Path, case_id: str) -> li
errors = []
for name, requirement in selected.items():
suite, test = requirement["suite"], requirement["name"]
listed = suites.get(suite, {}).get("testcases", {}).get(test, {})
for key in ("nodes", "drives_per_node"):
evidence_integer(requirement["topology"][key], f"required {key}", 1, 16)
listing_suite = suites[suite]
require(listing_suite["binary-id"] == suite, "nextest suite binary identity mismatch")
listed_binary = Path(listing_suite["binary-path"]).resolve(strict=True)
require(listed_binary == Path(run["test_binary"]["path"]).resolve(strict=True) and
digest(listed_binary) == run["test_binary"]["sha256"], "nextest selected another test binary")
require(listing_suite["package-name"] == "e2e_test" and listing_suite["build-platform"] in ("host", "target"),
"unexpected nextest suite metadata")
listed = listing_suite.get("testcases", {}).get(test, {})
require(listed.get("ignored") is False and listed.get("filter-match", {}).get("status") == "matches",
f"required test not selected: {suite}::{test}")
matches = [case for case in cases if case.get("name") == test and case.get("classname") == suite]
require(len(matches) == 1, f"missing/duplicate JUnit case: {suite}::{test}")
started = datetime.fromisoformat(matches[0].attrib["timestamp"].replace("Z", "+00:00"))
require(started.tzinfo is not None, "JUnit timestamp must include timezone")
# quick-junit truncates timestamps to milliseconds.
require(run["started_at"] - 0.001 <= started.timestamp() <= execution["finished_at"],
"JUnit testcase executed outside this run")
path = directory / requirement["oracle"]
require(path.resolve().is_relative_to(directory.resolve()), "oracle path escapes run directory")
require(run["started_at"] <= path.stat().st_mtime <= execution["finished_at"], "oracle outside run window")
require(digest(path) == execution["artifacts"][requirement["oracle"]], "oracle hash mismatch")
oracle = read_json(path)
require(oracle.get("schema") == 1 and oracle.get("evidence") == "process-restart", "not real process-restart evidence")
evidence_integer(oracle.get("schema"), "oracle schema", 1, 1)
require(oracle.get("evidence") == "process-restart", "not real process-restart evidence")
require(oracle.get("case") == name and oracle.get("run_id") == run["run_id"], "oracle belongs to another case/run")
require(oracle.get("source_revision") == run["source_revision"], "oracle source mismatch")
require(oracle.get("test_sources_sha256") == run["test_sources_sha256"], "compiled test source mismatch")
built = oracle["test_build"]
for key in ("source_revision", "dirty", "lock_blob", "features"):
require(built[key] == expected_build[key], f"compiled test {key} mismatch")
require(built["dirty"] is False, "test binary was compiled from dirty source")
require(all(isinstance(built[key], str) and built[key] and built[key] != "unknown" for key in ("target", "profile")),
"missing compiled target/profile")
require(isinstance(built["rustflags_hex"], str) and re.fullmatch(r"(?:[0-9a-f]{2})*", built["rustflags_hex"]) is not None,
"invalid compiled rustflags")
for label in ("binary", "test_binary"):
require(oracle.get(f"{label}_sha256") == run[label]["sha256"], f"oracle {label} mismatch")
require(oracle.get("topology") == requirement["topology"], "oracle topology mismatch")
number(oracle.get("pid_before"), "pid_before", 1)
number(oracle.get("pid_after"), "pid_after", 1)
for key in ("nodes", "drives_per_node"):
evidence_integer(oracle["topology"][key], f"observed {key}", 1, 16)
evidence_integer(oracle.get("pid_before"), "pid_before", 1, 2**32 - 1)
evidence_integer(oracle.get("pid_after"), "pid_after", 1, 2**32 - 1)
require(oracle["pid_before"] != oracle["pid_after"], "no process restart witnessed")
objects = oracle["objects"]
require(isinstance(objects, list) and requirement["min_objects"] <= len(objects) <= requirement["max_objects"],
@@ -992,8 +1032,13 @@ def check_scanner_heal_evidence(root: Path, directory: Path, case_id: str) -> li
physical = obj["physical"]
if obj["expected_physical"] is not None:
require(physical == obj["expected_physical"], "target shard differs from pre-fault manifest")
for geometry in [physical] + ([obj["expected_physical"]] if obj["expected_physical"] is not None else []):
data = evidence_integer(geometry["data_blocks"], "EC data blocks", 1, 16)
parity = evidence_integer(geometry["parity_blocks"], "EC parity blocks", 1, 16)
require(data + parity == oracle["topology"]["nodes"] * oracle["topology"]["drives_per_node"],
"EC geometry differs from this case's single set")
evidence_integer(geometry["erasure_index"], "target erasure index", 1, data + parity)
require(physical["has_xl_meta"] is True and physical["version_id"] is None, "missing target metadata")
number(physical["erasure_index"], "target erasure index", 1)
parts = physical["expected_part_numbers"]
require(isinstance(parts, list) and 0 < len(parts) <= 10000, "no physical part coverage")
require(all(type(part) is int and part > 0 for part in parts) and len(set(parts)) == len(parts),
@@ -1171,16 +1216,21 @@ class SelfTests(unittest.TestCase):
binary.chmod(0o700)
build = {"path": str(binary), "sha256": digest(binary)}
write_json(run_dir / "run.json", {"schema": 1, "run_id": "a" * 32, "source_revision": "b" * 40,
"binary_source_revision": "b" * 40, "test_sources_sha256": "e" * 64,
"binary_source_revision": "b" * 40,
"test_build": {"source_revision": "b" * 40, "dirty": False,
"lock_blob": "c" * 40, "features": "default"},
"started_at": datetime.now(timezone.utc).timestamp() - 1,
"binary": build, "test_binary": build})
write_json(run_dir / "listing.json", {"rust-suites": {requirement["suite"]: {"testcases": {
write_json(run_dir / "listing.json", {"rust-suites": {requirement["suite"]: {
"binary-id": requirement["suite"], "binary-path": str(binary), "package-name": "e2e_test", "build-platform": "target",
"testcases": {
requirement["name"]: {"ignored": False, "filter-match": {"status": "matches"}}
}}}})
(run_dir / "junit.xml").write_text(
f'<testsuites><testsuite><testcase name="{requirement["name"]}" classname="{requirement["suite"]}"/></testsuite></testsuites>')
f'<testsuites><testsuite><testcase name="{requirement["name"]}" classname="{requirement["suite"]}" '
f'timestamp="{datetime.now(timezone.utc).isoformat(timespec="milliseconds")}"/></testsuite></testsuites>')
physical = {"has_xl_meta": True, "version_id": None, "data_dir": "data-generation",
"erasure_index": 1, "expected_part_numbers": [1],
"erasure_index": 1, "data_blocks": 2, "parity_blocks": 2, "expected_part_numbers": [1],
"present_part_fingerprints": {"1": {"size": 12, "sha256": "c" * 64}},
"inline_data_fingerprint": None}
obj = {"key": "object", "version_id": None, "expected_bytes": 16, "actual_bytes": 16,
@@ -1191,7 +1241,8 @@ class SelfTests(unittest.TestCase):
write_json(run_dir / "background-target-restart.json", {
"schema": 1, "evidence": "process-restart", "case": "background-target-restart",
"run_id": "a" * 32, "source_revision": "b" * 40,
"test_sources_sha256": "e" * 64,
"test_build": {"source_revision": "b" * 40, "dirty": False, "lock_blob": "c" * 40,
"features": "default", "target": "aarch64-apple-darwin", "profile": "debug", "rustflags_hex": ""},
"binary_sha256": build["sha256"], "test_binary_sha256": build["sha256"],
"topology": {"nodes": 4, "drives_per_node": 1}, "pid_before": 10, "pid_after": 11,
"objects": objects, "node_listings": [[item["key"] for item in objects]] * 4,
@@ -1286,7 +1337,8 @@ class SelfTests(unittest.TestCase):
version += "modified source\n"
if kind == "unknown":
version = "rustfs without build provenance"
with mock.patch("subprocess.check_output", side_effect=["", "b" * 40, version]):
with mock.patch("subprocess.check_output", side_effect=["", "b" * 40, version, "c" * 40]), \
mock.patch.dict(os.environ, {"RUSTFS_E2E_EXPECTED_FEATURES": "default"}):
directory = Path(tmp) / "fresh"
binary = Path(tmp) / "fake-binary"
if kind == "current":
@@ -1297,6 +1349,67 @@ class SelfTests(unittest.TestCase):
begin_scanner_heal_receipt(root, directory, binary, binary)
self.assertFalse(directory.exists())
def test_scanner_heal_rejects_copied_junit_and_wrong_suite_build(self) -> None:
for fault in ("old-junit", "missing-time", "wrong-binary", "common-source", "lockfile", "features", "dirty-build"):
with self.subTest(fault=fault), tempfile.TemporaryDirectory() as tmp:
root, run_dir = self.scanner_heal_fixture(Path(tmp))
if fault in ("old-junit", "missing-time"):
path = run_dir / "junit.xml"
xml = ET.fromstring(path.read_bytes())
testcase = next(xml.iter("testcase"))
if fault == "old-junit":
testcase.set("timestamp", "2000-01-01T00:00:00.000Z")
else:
del testcase.attrib["timestamp"]
# Rewriting/copying gives an old execution a fresh mtime.
path.write_bytes(ET.tostring(xml))
elif fault == "wrong-binary":
path = run_dir / "listing.json"
listing = read_json(path)
another = Path(tmp) / "another-binary"
another.write_bytes(Path(tmp, "fake-binary").read_bytes())
listing["rust-suites"]["e2e_test"]["binary-path"] = str(another)
write_json(path, listing)
else:
path = run_dir / "background-target-restart.json"
oracle = read_json(path)
key, value = {"common-source": ("source_revision", "f" * 40), "lockfile": ("lock_blob", "f" * 40),
"features": ("features", "default,sftp"), "dirty-build": ("dirty", True)}[fault]
oracle["test_build"][key] = value
write_json(path, oracle)
(run_dir / "execution.json").unlink()
finish_scanner_heal_receipt(run_dir, 0)
self.assertTrue(check_scanner_heal_evidence(root, run_dir, "background-target-restart"), fault)
def test_scanner_heal_rejects_boolean_fractional_and_out_of_geometry_integers(self) -> None:
valid = {"schema": 1, "nodes": 4, "drives_per_node": 1, "pid_before": 10, "pid_after": 11,
"erasure_index": 1, "data_blocks": 2, "parity_blocks": 2}
cases = [(field, value) for field, correct in valid.items() for value in (True, float(correct))]
cases += [("erasure_index", 5), ("erasure_index", 0), ("pid_after", -1)]
for field, value in cases:
with self.subTest(field=field, value=value), tempfile.TemporaryDirectory() as tmp:
root, run_dir = self.scanner_heal_fixture(Path(tmp))
path = run_dir / "background-target-restart.json"
oracle = read_json(path)
if field in ("nodes", "drives_per_node"):
oracle["topology"][field] = value
elif field in ("erasure_index", "data_blocks", "parity_blocks"):
oracle["objects"][-1]["physical"][field] = value
else:
oracle[field] = value
write_json(path, oracle)
(run_dir / "execution.json").unlink()
finish_scanner_heal_receipt(run_dir, 0)
self.assertTrue(check_scanner_heal_evidence(root, run_dir, "background-target-restart"))
for filename in ("run.json", ".config/scanner-heal-required-tests.json"):
with self.subTest(filename=filename), tempfile.TemporaryDirectory() as tmp:
root, run_dir = self.scanner_heal_fixture(Path(tmp))
path = (root if filename.startswith(".config") else run_dir) / filename
content = read_json(path)
content["schema"] = True
write_json(path, content)
self.assertTrue(check_scanner_heal_evidence(root, run_dir, "background-target-restart"))
def test_core_gate_rejects_missing_ignored_filtered_and_corrupt_inputs(self) -> None:
with tempfile.TemporaryDirectory() as tmp:
root = Path(tmp)