mirror of
https://github.com/rustfs/rustfs.git
synced 2026-10-06 13:13:03 +00:00
hector
e22414fe68
timeout-minutes requires a numeric value, but inputs.pool_timeout_minutes
is a string, so '${{ inputs.pool_timeout_minutes || 240 }}' evaluated to
the string '240' and the runner rejected the template: 'An error occurred
when attempting to determine the step timeout ... Unexpected value 240'
(chain run 36505625402 job at 02:25:52, standalone run 36393453114 at
07:46:37). Both pool runs since #8155 merged have therefore executed
without the intended step budget. Wrap the expression in fromJSON and
update the policy test that pinned the broken literal.
479 lines
30 KiB
Python
479 lines
30 KiB
Python
#!/usr/bin/env python3
|
|
"""Exercise candidate substitution and complete-chain acceptance boundaries."""
|
|
import copy
|
|
import hashlib
|
|
import io
|
|
import json
|
|
from pathlib import Path
|
|
import subprocess
|
|
import tempfile
|
|
import unittest
|
|
from unittest import mock
|
|
import zipfile
|
|
|
|
import functional_chain_evidence as evidence
|
|
import resolve_functional_candidate as candidate
|
|
|
|
|
|
class CandidateTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.run = {"id": 123, "run_attempt": 2, "head_sha": "a" * 40}
|
|
self.manifest = {"schema": 2, "workflow_sha": "a" * 40, "source_sha": "b" * 40, "source_ref": "release",
|
|
"build_run_id": 123, "build_run_attempt": 2, "package_sha256": "c" * 64,
|
|
"package_url": "https://dl.rustfs.com/artifacts/rustfs/packages/nightly/runs/123/2/" + "c" * 64 + "/rustfs.deb"}
|
|
|
|
def archive(self, names=None):
|
|
output = io.BytesIO()
|
|
with zipfile.ZipFile(output, "w") as archive:
|
|
for name in names or ["nightly-candidate-123-2.json"]:
|
|
archive.writestr(name, json.dumps(self.manifest))
|
|
payload = output.getvalue()
|
|
artifact = {"id": 789, "name": "nightly-candidate-123-2", "expired": False, "size_in_bytes": len(payload),
|
|
"digest": "sha256:" + hashlib.sha256(payload).hexdigest(), "workflow_run": {"id": 123, "head_sha": "a" * 40}}
|
|
return payload, artifact
|
|
|
|
def test_distinct_build_source_preserves_both_identities(self):
|
|
payload, artifact = self.archive()
|
|
result = candidate.read_manifest(payload, artifact, self.run)
|
|
self.assertEqual(result["source_sha"], "b" * 40)
|
|
self.assertEqual(result["workflow_sha"], "a" * 40)
|
|
self.assertEqual(result["source_ref"], "release")
|
|
|
|
def test_legacy_requires_the_build_and_workflow_sha_to_agree(self):
|
|
self.manifest["schema"] = 1
|
|
del self.manifest["workflow_sha"], self.manifest["source_ref"]
|
|
with self.assertRaisesRegex(ValueError, "legacy"):
|
|
candidate.validate_manifest(self.manifest, self.run)
|
|
self.manifest["source_sha"] = self.run["head_sha"]
|
|
candidate.validate_manifest(self.manifest, self.run)
|
|
|
|
def test_manifest_substitutions_fail(self):
|
|
for key, value in (("workflow_sha", "d" * 40), ("build_run_id", 124), ("build_run_attempt", 1),
|
|
("package_sha256", "d" * 64), ("package_url", "https://example.com/package.deb"),
|
|
("schema", True), ("source_ref", "release\nFORGED=value")):
|
|
with self.subTest(key=key), self.assertRaises(ValueError):
|
|
candidate.validate_manifest({**self.manifest, key: value}, self.run)
|
|
|
|
def test_artifact_substitutions_and_archive_members_fail(self):
|
|
payload, artifact = self.archive()
|
|
for key, value in (("expired", True), ("name", "nightly-candidate-123-1"), ("size_in_bytes", 1),
|
|
("digest", "sha256:" + "d" * 64), ("workflow_run", {"id": 124, "head_sha": "a" * 40})):
|
|
with self.subTest(key=key), self.assertRaises(ValueError):
|
|
candidate.read_manifest(payload, {**artifact, key: value}, self.run)
|
|
for names in (["../nightly-candidate-123-2.json"], ["nightly-candidate-123-2.json", "extra.json"]):
|
|
payload, artifact = self.archive(names)
|
|
with self.assertRaises(ValueError):
|
|
candidate.read_manifest(payload, artifact, self.run)
|
|
|
|
def test_resolver_uses_attempt_metadata_and_does_not_resolve_moving_branch(self):
|
|
payload, artifact = self.archive()
|
|
run = {**self.run, "path": ".github/workflows/nightly-gnu.yml", "head_branch": "main", "head_repository": {"full_name": "rustfs/rustfs"},
|
|
"event": "schedule", "status": "completed", "conclusion": "success", "run_started_at": "2026-09-12T00:00:00Z"}
|
|
with mock.patch.object(candidate, "api", side_effect=[run, {"artifacts": [artifact]}, payload]) as api:
|
|
result = candidate.resolve(123, 2)
|
|
self.assertTrue(api.call_args_list[0].args[0].endswith("/attempts/2"))
|
|
self.assertFalse(any("branches/" in call.args[0] for call in api.call_args_list))
|
|
self.assertEqual(result["artifact_id"], 789)
|
|
self.assertEqual(result["manifest"]["source_sha"], "b" * 40)
|
|
|
|
|
|
class EvidenceTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.temp = tempfile.TemporaryDirectory()
|
|
self.addCleanup(self.temp.cleanup)
|
|
self.directory = Path(self.temp.name)
|
|
self.chain = {"run_id": 456, "attempt": 1, "candidate": {"source_sha": "a" * 40, "workflow_sha": "b" * 40}}
|
|
self.needs = {suite: {"result": "success"} for suite in evidence.SUITES}
|
|
for suite in evidence.SUITES:
|
|
value = {"schema": 1, "suite": suite, "chain": self.chain, "valid": True, "report_sha256": "c" * 64,
|
|
"counts": {"PASS": 1, "FAIL": 0, "SKIP": 0, "UNSUPPORTED": 0, "RUNNING": 0}}
|
|
(self.directory / (suite + ".json")).write_text(json.dumps(value))
|
|
|
|
def test_complete_chain_retains_source_identity(self):
|
|
result = evidence.aggregate(self.chain, self.directory, self.needs)
|
|
self.assertTrue(result["complete"])
|
|
self.assertEqual(result["chain"], self.chain)
|
|
self.assertEqual([suite["suite"] for suite in result["suites"]], [
|
|
"upgrade", "s3", "kms", "tier", "storage", "heal", "pool", "security", "replication",
|
|
"fault-tolerance", "table", "performance"])
|
|
|
|
def test_release_suites_cannot_be_missing_failed_or_have_missing_artifacts(self):
|
|
for suite in ("fault-tolerance", "table"):
|
|
with self.subTest(suite=suite):
|
|
missing = {key: value for key, value in self.needs.items() if key != suite}
|
|
with self.assertRaises(ValueError):
|
|
evidence.aggregate(self.chain, self.directory, missing)
|
|
with self.assertRaises(ValueError):
|
|
evidence.aggregate(self.chain, self.directory, {**self.needs, suite: {"result": "failure"}})
|
|
path = self.directory / (suite + ".json")
|
|
original = path.read_text()
|
|
path.unlink()
|
|
with self.assertRaises(ValueError):
|
|
evidence.aggregate(self.chain, self.directory, self.needs)
|
|
path.write_text(original)
|
|
|
|
def test_failure_report_preserves_counts_without_weakening_success_gate(self):
|
|
path = self.directory / "s3.json"
|
|
record = json.loads(path.read_text())
|
|
record.update(valid=False, counts={**record["counts"], "FAIL": 2}, error="failed cases")
|
|
path.write_text(json.dumps(record))
|
|
needs = {**self.needs, "s3": {"result": "failure"}, "performance": {"result": "cancelled"}}
|
|
(self.directory / "performance.json").unlink()
|
|
result = evidence.summarize(self.chain, self.directory, needs)
|
|
self.assertFalse(result["complete"])
|
|
self.assertEqual(result["lanes"][1]["evidence"]["counts"]["FAIL"], 2)
|
|
self.assertEqual(result["lanes"][-1]["result"], "cancelled")
|
|
self.assertIsNone(result["lanes"][-1]["evidence"])
|
|
with self.assertRaises(ValueError):
|
|
evidence.aggregate(self.chain, self.directory, needs)
|
|
|
|
def test_preparation_failure_still_has_all_required_lanes(self):
|
|
result = evidence.summarize(None, self.directory / "absent", {"prepare": {"result": "failure"}})
|
|
self.assertFalse(result["complete"])
|
|
self.assertEqual([lane["suite"] for lane in result["lanes"]], list(evidence.SUITES))
|
|
self.assertIn("| performance | missing | missing |", evidence.render_summary(result))
|
|
self.assertIn("Preparation: failure", evidence.render_summary(result))
|
|
|
|
def test_malformed_or_cross_attempt_record_is_not_reused(self):
|
|
for payload in ("broken json", "[]", json.dumps({"suite": "s3", "chain": {"attempt": 9}})):
|
|
(self.directory / "s3.json").write_text(payload)
|
|
result = evidence.summarize(self.chain, self.directory, self.needs)
|
|
self.assertFalse(result["complete"])
|
|
self.assertIsNone(result["lanes"][1]["evidence"])
|
|
|
|
def test_missing_failed_cancelled_or_skipped_lane_never_passes(self):
|
|
for state in ("failure", "cancelled", "skipped", "pending"):
|
|
with self.subTest(state=state), self.assertRaises(ValueError):
|
|
evidence.aggregate(self.chain, self.directory, {**self.needs, "s3": {"result": state}})
|
|
del self.needs["s3"]
|
|
with self.assertRaises(ValueError):
|
|
evidence.aggregate(self.chain, self.directory, self.needs)
|
|
|
|
def test_partial_rerun_missing_artifact_and_zero_test_fail(self):
|
|
path = self.directory / "s3.json"
|
|
original = json.loads(path.read_text())
|
|
values = [{**original, "chain": {**self.chain, "attempt": 2}},
|
|
{**original, "counts": {**original["counts"], "PASS": 0}},
|
|
{**original, "counts": {**original["counts"], "FAIL": 1}},
|
|
{**original, "valid": False}, {**original, "report_sha256": ""}]
|
|
for value in values:
|
|
path.write_text(json.dumps(value))
|
|
with self.assertRaises(ValueError):
|
|
evidence.aggregate(self.chain, self.directory, self.needs)
|
|
path.unlink()
|
|
with self.assertRaises(ValueError):
|
|
evidence.aggregate(self.chain, self.directory, self.needs)
|
|
|
|
def test_reports_count_case_status_not_cleanup_status(self):
|
|
text = "| Topology | Case | Name | Status | Cleanup |\n| --- | --- | --- | --- | --- |\n| sns | TIER-1 | test | UNSUPPORTED | PASS |\n| sns | TIER-2 | test | FAIL | PASS |\n"
|
|
counts = evidence.report_counts(text)
|
|
self.assertEqual(counts["PASS"], 0)
|
|
self.assertEqual(counts["FAIL"], 1)
|
|
self.assertEqual(counts["UNSUPPORTED"], 1)
|
|
self.assertEqual(evidence.report_counts("")["PASS"], 0)
|
|
|
|
def test_performance_needs_real_complete_metrics(self):
|
|
header = "method\tsize\tthroughput\tobj_per_s\treq_avg\treq_p50\treq_p90\treq_p99\n"
|
|
row = "put\t1MiB\t100MiB/s\t100\t1ms\t1ms\t2ms\t3ms\n"
|
|
self.assertEqual(evidence.report_counts(header + row, True)["PASS"], 1)
|
|
self.assertEqual(evidence.report_counts(header, True)["PASS"], 0)
|
|
with self.assertRaises(ValueError):
|
|
evidence.report_counts(header + row + row, True)
|
|
with self.assertRaises(ValueError):
|
|
evidence.report_counts(header + "put\t1MiB\t\t\t\t\t\t\n", True)
|
|
|
|
def test_fault_tolerance_requires_complete_known_verdicts(self):
|
|
passing = "FT-CASE: A-read verdict=pass observed expected result\n"
|
|
divergent = "FT-CASE: C2-read verdict=known-divergence lock majority unavailable\n"
|
|
summary = "FT-SUMMARY: unexpected=0 known-divergence=1 strict=0\n"
|
|
counts = evidence.fault_tolerance_counts(passing + divergent + summary)
|
|
self.assertEqual(counts["PASS"], 1)
|
|
self.assertEqual(counts["UNSUPPORTED"], 1)
|
|
for invalid in (passing, passing + passing + summary, passing + divergent + summary.replace("strict=0", "strict=1"),
|
|
passing + divergent + summary.replace("known-divergence=1", "known-divergence=0"),
|
|
passing.replace("verdict=pass", "verdict=UNKNOWN") + summary):
|
|
with self.subTest(invalid=invalid), self.assertRaises(ValueError):
|
|
evidence.fault_tolerance_counts(invalid)
|
|
|
|
|
|
class EnvelopeTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.temp = tempfile.TemporaryDirectory()
|
|
self.addCleanup(self.temp.cleanup)
|
|
self.root = Path(self.temp.name)
|
|
(self.root / ".config").mkdir()
|
|
(self.root / ".config/functional-script-revision.txt").write_text("d" * 40)
|
|
manifest = {"schema": 2, "workflow_sha": "a" * 40, "source_sha": "b" * 40, "source_ref": "release",
|
|
"build_run_id": 123, "build_run_attempt": 2, "package_sha256": "c" * 64,
|
|
"package_url": "https://dl.rustfs.com/artifacts/rustfs/packages/nightly/runs/123/2/" + "c" * 64 + "/rustfs.deb"}
|
|
self.chain = {"schema": 1, "run_id": 456, "attempt": 3, "workflow_sha": "e" * 40, "testing_sha": "d" * 40,
|
|
"candidate": {"manifest": manifest, "artifact_id": 789, "artifact_digest": "sha256:" + "f" * 64,
|
|
"workflow_sha": "a" * 40, "workflow_ref": "main", "build_started_at": "2026-09-12T00:00:00Z"}}
|
|
self.env = {"CHAIN_MANIFEST": json.dumps(self.chain), "GITHUB_RUN_ID": "456", "GITHUB_RUN_ATTEMPT": "3", "GITHUB_SHA": "e" * 40,
|
|
"GITHUB_ENV": str(self.root / "env"), "GITHUB_OUTPUT": str(self.root / "output"),
|
|
"CHAIN_JOB_STATUS": "success", "CHAIN_TEST_OUTCOME": "success", "CHAIN_REPORT_OUTCOME": "success"}
|
|
|
|
def test_consumer_exports_the_same_package_and_checksum_to_installers(self):
|
|
with mock.patch.object(evidence, "ROOT", self.root), mock.patch.dict(evidence.os.environ, self.env), mock.patch.object(evidence.subprocess, "check_output", return_value="e" * 40):
|
|
evidence.consume(evidence.current_chain())
|
|
variables = dict(line.split("=", 1) for line in (self.root / "env").read_text().splitlines())
|
|
self.assertEqual(variables["PACKAGE_SHA256"], "c" * 64)
|
|
self.assertEqual(variables["TO_SHA256"], "c" * 64)
|
|
self.assertEqual(variables["RUSTFS_NIGHTLY_PACKAGE_URL"], self.chain["candidate"]["manifest"]["package_url"])
|
|
|
|
def test_partial_rerun_and_wrong_lane_checkout_fail_before_install(self):
|
|
for changes, head in (({"GITHUB_RUN_ATTEMPT": "4"}, "e" * 40), ({}, "f" * 40), ({"GITHUB_RUN_ID": "457"}, "e" * 40)):
|
|
with mock.patch.object(evidence, "ROOT", self.root), mock.patch.dict(evidence.os.environ, {**self.env, **changes}), mock.patch.object(evidence.subprocess, "check_output", return_value=head), self.assertRaises(ValueError):
|
|
evidence.current_chain()
|
|
self.assertFalse((self.root / "env").exists())
|
|
|
|
def test_testing_sha_fallback_is_accepted_while_garbage_is_rejected(self):
|
|
# prepare's >24h staleness fallback legitimately sets testing_sha to
|
|
# auto-testing main HEAD, which differs from the committed pin; only
|
|
# the sha format is an invariant now.
|
|
for testing_sha, ok in (("d" * 40, True), ("1" * 40, True), ("xyz", False), ("", False)):
|
|
chain = dict(self.chain, testing_sha=testing_sha)
|
|
env = dict(self.env, CHAIN_MANIFEST=json.dumps(chain))
|
|
if ok:
|
|
with mock.patch.object(evidence, "ROOT", self.root), mock.patch.dict(evidence.os.environ, env), mock.patch.object(evidence.subprocess, "check_output", return_value="e" * 40):
|
|
evidence.consume(evidence.current_chain())
|
|
else:
|
|
with mock.patch.object(evidence, "ROOT", self.root), mock.patch.dict(evidence.os.environ, env), mock.patch.object(evidence.subprocess, "check_output", return_value="e" * 40), self.assertRaises(ValueError):
|
|
evidence.current_chain()
|
|
|
|
def test_report_or_swallowed_test_failure_cannot_produce_valid_evidence(self):
|
|
report = self.root / "cases.md"
|
|
report.write_text("| Case | Name | Status |\n| --- | --- | --- |\n| KMS-1 | fixture | PASS |\n")
|
|
for index, (key, status) in enumerate((("CHAIN_TEST_OUTCOME", "failure"), ("CHAIN_REPORT_OUTCOME", "failure"), ("CHAIN_JOB_STATUS", "cancelled"))):
|
|
output = self.root / str(index) / "kms.json"
|
|
Path(self.env["GITHUB_OUTPUT"]).unlink(missing_ok=True)
|
|
with mock.patch.dict(evidence.os.environ, {**self.env, key: status}), mock.patch.object(evidence.subprocess, "check_output", return_value="d" * 40), self.assertRaises(ValueError):
|
|
evidence.record(self.chain, "kms", report, output)
|
|
self.assertFalse(json.loads(output.read_text())["valid"])
|
|
self.assertIn("error", json.loads(output.read_text()))
|
|
self.assertEqual(Path(self.env["GITHUB_OUTPUT"]).read_text(), "written=true\n")
|
|
output = self.root / "success" / "kms.json"
|
|
with mock.patch.dict(evidence.os.environ, self.env), mock.patch.object(evidence.subprocess, "check_output", return_value="d" * 40):
|
|
evidence.record(self.chain, "kms", report, output)
|
|
self.assertTrue(json.loads(output.read_text())["valid"])
|
|
|
|
def test_collision_or_failed_write_never_authorizes_evidence_upload(self):
|
|
report = self.root / "cases.md"
|
|
report.write_text("| Case | Status |\n| --- | --- |\n| KMS-1 | PASS |\n")
|
|
output = self.root / "stale" / "kms.json"
|
|
output.parent.mkdir()
|
|
output.write_text("OLD RUN EVIDENCE")
|
|
with mock.patch.dict(evidence.os.environ, self.env), mock.patch.object(evidence.subprocess, "check_output", return_value="d" * 40):
|
|
with self.assertRaises(FileExistsError):
|
|
evidence.record(self.chain, "kms", report, output)
|
|
self.assertEqual(output.read_text(), "OLD RUN EVIDENCE")
|
|
self.assertFalse(Path(self.env["GITHUB_OUTPUT"]).exists())
|
|
with mock.patch.object(Path, "write_text", side_effect=OSError("disk full")), self.assertRaises(OSError):
|
|
evidence.record(self.chain, "kms", report, self.root / "new" / "kms.json")
|
|
self.assertFalse(Path(self.env["GITHUB_OUTPUT"]).exists())
|
|
|
|
def test_record_uses_workspace_testing_checkout_for_nested_lane(self):
|
|
lane_checkout = self.root / "rustfs-repo"
|
|
lane_checkout.mkdir()
|
|
private_checkout = self.root / "auto-testing"
|
|
private_checkout.mkdir()
|
|
subprocess.run(["git", "init", "--quiet", str(private_checkout)], check=True)
|
|
subprocess.run(["git", "-c", "user.name=Fixture", "-c", "user.email=fixture@example.invalid",
|
|
"-c", "commit.gpgSign=false", "commit", "--allow-empty", "-qm", "fixture"],
|
|
cwd=private_checkout, check=True)
|
|
private_head = subprocess.check_output(["git", "rev-parse", "HEAD"], cwd=private_checkout, text=True).strip()
|
|
chain = {**self.chain, "testing_sha": private_head}
|
|
report = self.root / "security.md"
|
|
report.write_text("| Case | Status |\n| --- | --- |\n| SEC-1 | PASS |\n")
|
|
env = {**self.env, "GITHUB_WORKSPACE": str(self.root)}
|
|
output = self.root / "nested-success" / "security.json"
|
|
with mock.patch.object(evidence, "ROOT", lane_checkout), mock.patch.dict(evidence.os.environ, env):
|
|
evidence.record(chain, "security", report, output)
|
|
self.assertTrue(json.loads(output.read_text())["valid"])
|
|
wrong_pin_output = self.root / "nested-wrong-pin" / "security.json"
|
|
with self.assertRaisesRegex(ValueError, "suite used a different private script revision"):
|
|
evidence.record({**chain, "testing_sha": "f" * 40}, "security", report, wrong_pin_output)
|
|
self.assertFalse(json.loads(wrong_pin_output.read_text())["valid"])
|
|
fallback_output = self.root / "local-success" / "security.json"
|
|
with mock.patch.object(evidence, "ROOT", self.root), mock.patch.dict(evidence.os.environ, self.env):
|
|
evidence.os.environ.pop("GITHUB_WORKSPACE", None)
|
|
evidence.record(chain, "security", report, fallback_output)
|
|
self.assertTrue(json.loads(fallback_output.read_text())["valid"])
|
|
|
|
def test_unknown_status_cannot_hide_among_passing_cases(self):
|
|
text = "| Case | Name | Status |\n| --- | --- | --- |\n| KMS-1 | fixture | PASS |\n| KMS-2 | fixture | NOT RUN |\n"
|
|
with self.assertRaises(ValueError):
|
|
evidence.report_counts(text)
|
|
|
|
def test_release_suite_records_require_successful_execution_and_complete_reports(self):
|
|
reports = {
|
|
"table": "| Case | Name | Status |\n| --- | --- | --- |\n| TBL-101 | Iceberg smoke | PASS |\n",
|
|
"fault-tolerance": "FT-CASE: A-read verdict=pass read succeeded\nFT-SUMMARY: unexpected=0 known-divergence=0 strict=0\n",
|
|
}
|
|
for suite, text in reports.items():
|
|
report = self.root / (suite + ".txt")
|
|
report.write_text(text)
|
|
for status in ("success", "failure"):
|
|
with self.subTest(suite=suite, status=status):
|
|
output = self.root / (suite + "-" + status) / (suite + ".json")
|
|
with mock.patch.dict(evidence.os.environ, {**self.env, "CHAIN_TEST_OUTCOME": status}), \
|
|
mock.patch.object(evidence.subprocess, "check_output", return_value="d" * 40):
|
|
if status == "success":
|
|
evidence.record(self.chain, suite, report, output)
|
|
else:
|
|
with self.assertRaises(ValueError):
|
|
evidence.record(self.chain, suite, report, output)
|
|
self.assertEqual(json.loads(output.read_text())["valid"], status == "success")
|
|
|
|
def test_driver_passes_one_manifest_and_runs_every_lane_after_failure(self):
|
|
from check_test_wiring import yaml_block
|
|
lines = (candidate.ROOT / ".github/workflows/rustfs-functional-chain.yml").read_text().splitlines()
|
|
previous = None
|
|
for suite in evidence.SUITES:
|
|
job = "\n".join(yaml_block(lines, suite, 2))
|
|
self.assertIn("needs: [prepare" + (", " + previous if previous else "") + "]", job)
|
|
if suite == "performance":
|
|
# gated on the preflight probe: runs only when its fleet is online
|
|
self.assertIn("if: ${{ always() && needs.prepare.result == 'success' && needs.prepare.outputs.performance_ready == 'online' }}", job)
|
|
else:
|
|
self.assertIn("if: ${{ always() && needs.prepare.result == 'success' }}", job)
|
|
self.assertIn("chain_manifest: ${{ needs.prepare.outputs.manifest }}", job)
|
|
previous = suite
|
|
complete = "\n".join(yaml_block(lines, "complete-chain", 2))
|
|
self.assertIn("needs: [prepare, " + ", ".join(evidence.SUITES) + "]", complete)
|
|
self.assertIn("functional_chain_evidence.py aggregate", complete)
|
|
self.assertIn("functional_chain_evidence.py summarize", complete)
|
|
self.assertIn("functional-chain-report-", complete)
|
|
self.assertIn("needs.prepare.result != 'skipped'", complete)
|
|
prepare = "\n".join(yaml_block(lines, "prepare", 2))
|
|
self.assertIn("Check shared functional fleet runner before scheduling suites", prepare)
|
|
self.assertIn("check_functional_runners.py smoke-testing", prepare)
|
|
self.assertIn("Probe performance fleet runner", prepare)
|
|
self.assertIn("check_functional_runners.py pf-testing", prepare)
|
|
self.assertIn("performance_ready: ${{ steps.perf_probe.outputs.performance_ready }}", prepare)
|
|
|
|
def test_every_lane_retains_failed_evidence_and_identifies_its_own_attempt(self):
|
|
paths = list((candidate.ROOT / ".github/workflows").glob("rustfs-*-test.yml"))
|
|
lanes = [path for path in paths if "name: Upload chain evidence" in path.read_text()]
|
|
self.assertEqual(len(lanes), len(evidence.SUITES))
|
|
for path in lanes:
|
|
with self.subTest(path=path.name):
|
|
text = path.read_text()
|
|
self.assertIn("if: ${{ always() && steps.chain_record.outputs.written == 'true' }}", text)
|
|
self.assertIn("python3 auto-testing/scripts/issue_manager.py handle", text)
|
|
self.assertIn('--run-id "${GITHUB_RUN_ID}"', text)
|
|
self.assertIn('--attempt "${GITHUB_RUN_ATTEMPT}"', text)
|
|
run_url = next(line for line in text.splitlines() if "--run-url" in line)
|
|
self.assertIn('/attempts/${GITHUB_RUN_ATTEMPT}#summary"', run_url)
|
|
|
|
def test_table_issue_manager_receives_outcome_and_case_evidence(self):
|
|
text = (candidate.ROOT / ".github/workflows/rustfs-table-test.yml").read_text()
|
|
self.assertIn("--outcome '${{ steps.test.outcome }}'", text)
|
|
self.assertIn('--report "${FUNCTIONAL_ARTIFACTS_DIR}/cases.md"', text)
|
|
|
|
def test_pool_topology_is_checked_before_cleanup_with_four_node_defaults(self):
|
|
text = (candidate.ROOT / ".github/workflows/rustfs-pool-expand-test.yml").read_text()
|
|
self.assertIn("http://rustfs-node4:9000", text)
|
|
self.assertIn("vars.RUSTFS_NODES || 'vm000 vm001 vm002'", text)
|
|
self.assertLess(text.index("name: Validate pool topology"), text.index("name: Cleanup environment (before)"))
|
|
self.assertIn("steps.topology.outcome == 'success' && inputs.cleanup_after", text)
|
|
|
|
|
|
class RunnerTests(unittest.TestCase):
|
|
def test_offline_or_missing_runner_fails_before_dispatch(self):
|
|
import check_functional_runners as runners
|
|
for inventory in ([], [{"status": "offline", "labels": [{"name": "pf-testing"}]}]):
|
|
with mock.patch.object(runners, "api", return_value={"runners": inventory}), self.assertRaisesRegex(ValueError, "pf-testing"):
|
|
runners.check(["pf-testing"])
|
|
|
|
def test_busy_online_runner_is_available_and_inventory_is_paginated(self):
|
|
import check_functional_runners as runners
|
|
pages = [{"runners": [{"status": "online", "labels": []}] * 100},
|
|
{"runners": [{"status": "online", "busy": True, "labels": [{"name": "pf-testing"}]}]}]
|
|
with mock.patch.object(runners, "api", side_effect=pages) as api:
|
|
runners.check(["pf-testing"])
|
|
self.assertIn("page=2", api.call_args.args[0])
|
|
|
|
def test_unavailable_inventory_does_not_assume_online(self):
|
|
import check_functional_runners as runners
|
|
with mock.patch.object(runners, "api", side_effect=OSError("forbidden")), self.assertRaises(OSError):
|
|
runners.check(["pf-testing"])
|
|
|
|
def test_release_lanes_use_pinned_scripts_and_verified_package_before_recording(self):
|
|
for suite, report in (("fault-tolerance", "suite.log"), ("table", "cases.md")):
|
|
with self.subTest(suite=suite):
|
|
workflow = (candidate.ROOT / f".github/workflows/rustfs-{suite}-test.yml").read_text()
|
|
self.assertIn(" workflow_call:", workflow)
|
|
self.assertIn(" workflow_dispatch:", workflow)
|
|
self.assertIn("group: rustfs-shared-functional-tests-v2", workflow)
|
|
expected_ref = (
|
|
"ref: ${{ steps.chain.outputs.testing_sha || inputs.auto_testing_ref || 'main' }}"
|
|
if suite == "fault-tolerance"
|
|
else "ref: ${{ steps.chain.outputs.testing_sha || 'main' }}"
|
|
)
|
|
self.assertIn(expected_ref, workflow)
|
|
self.assertIn("steps.chain_package.outputs.package_url || inputs.package_url", workflow)
|
|
self.assertLess(workflow.index("prepare_functional_package.py prepare"), workflow.index("id: test"))
|
|
self.assertIn("prepare_functional_package.py cleanup", workflow)
|
|
self.assertIn(f"functional_chain_evidence.py record --suite {suite}", workflow)
|
|
self.assertIn('--report "${FUNCTIONAL_ARTIFACTS_DIR}/' + report + '"', workflow)
|
|
self.assertIn("CHAIN_TEST_OUTCOME: ${{ steps.test.outcome }}", workflow)
|
|
self.assertIn("CHAIN_REPORT_OUTCOME: ${{ steps.chain_report.outcome }}", workflow)
|
|
|
|
|
|
class WorkflowTimeoutTests(unittest.TestCase):
|
|
def test_non_performance_suites_have_hard_and_step_deadlines(self):
|
|
from check_test_wiring import yaml_block
|
|
from test_security_workflow import FunctionalWorkflowTests, named_steps
|
|
jobs = {**FunctionalWorkflowTests.JOBS, "security": "security-test"}
|
|
jobs.pop("performance")
|
|
self.assertEqual(len(jobs), 11)
|
|
for suite, job_id in jobs.items():
|
|
with self.subTest(suite=suite):
|
|
source = (candidate.ROOT / f".github/workflows/rustfs-{suite}-test.yml").read_text()
|
|
job = yaml_block(source.splitlines(), job_id, 2)
|
|
job_timeout = 360 if suite == "pool-expand" else 60
|
|
self.assertIn(f" timeout-minutes: {job_timeout}", job)
|
|
steps = named_steps(job)
|
|
primary = [step for step in steps.values() if any(
|
|
line in (" id: test", " id: pool_test") for line in step)]
|
|
self.assertEqual(len(primary), 1)
|
|
if suite == "pool-expand":
|
|
self.assertIn(" timeout-minutes: ${{ fromJSON(inputs.pool_timeout_minutes || '240') }}", primary[0])
|
|
for event in ("workflow_call", "workflow_dispatch"):
|
|
event_block = yaml_block(source.splitlines(), event, 2)
|
|
timeout_input = yaml_block(event_block, "pool_timeout_minutes", 6)
|
|
self.assertIsNotNone(timeout_input, event)
|
|
self.assertIn(" default: '240'", timeout_input)
|
|
self.assertIn(" required: false", timeout_input)
|
|
else:
|
|
self.assertIn(" timeout-minutes: 45", primary[0])
|
|
cleanup = "Cleanup environment"
|
|
for phase in ("before", "after"):
|
|
self.assertIn(" timeout-minutes: 5", steps[f"{cleanup} ({phase})"])
|
|
self.assertTrue(any("always()" in line for line in steps[f"{cleanup} (after)"]))
|
|
for name, step in steps.items():
|
|
if name.startswith(("Generate report", "Upload functional report", "File failure issue",
|
|
"Upload report", "Upload test logs", "Manage backlog issues")):
|
|
self.assertIn(" timeout-minutes: 2", step, name)
|
|
if suite in ("heal", "pool-expand"):
|
|
install = next(step for name, step in steps.items() if name.startswith("Install RustFS package"))
|
|
self.assertIn(" timeout-minutes: 5", install)
|
|
self.assertIn(" timeout-minutes: 5", steps["Preflight checks"])
|
|
|
|
def test_performance_is_exempt_from_functional_timeouts(self):
|
|
from check_test_wiring import yaml_block
|
|
source = (candidate.ROOT / ".github/workflows/rustfs-performance-test.yml").read_text()
|
|
job = "\n".join(yaml_block(source.splitlines(), "performance-test", 2))
|
|
self.assertIn(" timeout-minutes: 900", job)
|
|
self.assertNotIn(" timeout-minutes:", job)
|
|
self.assertIn("RUSTFS_WARP_DURATION: ${{ inputs.warp_duration || '5m' }}", job)
|
|
self.assertNotIn("RUSTFS_WARP_SLEEP:", job)
|
|
self.assertIn("RUSTFS_WARP_METHODS: ${{ inputs.test_method }}", job)
|
|
self.assertIn("RUSTFS_WARP_SIZES: ${{ inputs.object_size }}", job)
|
|
self.assertIn("default: '5m'", source)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|