mirror of
https://github.com/rustfs/rustfs.git
synced 2026-09-19 10:03:13 +08:00
fix(ci): verify complete functional chain evidence (#7690)
* fix(ci): bind nightly lanes to one resolved source * fix(ci): verify complete functional chain evidence
This commit is contained in:
Chris
@@ -0,0 +1,151 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Bind functional-suite evidence to one candidate and one chain attempt."""
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import csv
|
||||
from datetime import datetime, timezone
|
||||
import hashlib
|
||||
import io
|
||||
import json
|
||||
import os
|
||||
from pathlib import Path
|
||||
import re
|
||||
import subprocess
|
||||
|
||||
from resolve_functional_candidate import ROOT, positive, require, sha, validate_manifest
|
||||
|
||||
SUITES = ("upgrade", "s3", "kms", "tier", "storage", "heal", "pool", "security", "replication", "performance")
|
||||
MAX_REPORT = 8 * 1024 * 1024
|
||||
|
||||
|
||||
def current_chain():
|
||||
chain = json.loads(os.environ["CHAIN_MANIFEST"])
|
||||
require(isinstance(chain, dict) and set(chain) == {"schema", "run_id", "attempt", "workflow_sha", "testing_sha", "candidate"}, "invalid chain envelope")
|
||||
require(type(chain["schema"]) is int and chain["schema"] == 1, "unsupported chain schema")
|
||||
require(positive(chain["run_id"]) and positive(chain["attempt"]), "invalid chain run identity")
|
||||
require(chain["run_id"] == int(os.environ["GITHUB_RUN_ID"]) and chain["attempt"] == int(os.environ["GITHUB_RUN_ATTEMPT"]), "chain belongs to another run attempt; rerun all jobs")
|
||||
require(sha(chain["workflow_sha"]) and chain["workflow_sha"] == os.environ["GITHUB_SHA"], "chain workflow source mismatch")
|
||||
head = subprocess.check_output(["git", "rev-parse", "HEAD"], cwd=ROOT, text=True).strip()
|
||||
require(head == chain["workflow_sha"], "lane checkout differs from chain workflow source")
|
||||
require(chain["testing_sha"] == (ROOT / ".config/functional-script-revision.txt").read_text().strip() and sha(chain["testing_sha"]), "private script pin differs from chain")
|
||||
candidate = chain["candidate"]
|
||||
require(isinstance(candidate, dict) and set(candidate) == {"manifest", "artifact_id", "artifact_digest", "workflow_sha", "workflow_ref", "build_started_at"}, "invalid candidate envelope")
|
||||
manifest = candidate["manifest"]
|
||||
require(positive(candidate["artifact_id"]) and isinstance(candidate["artifact_digest"], str) and bool(re.fullmatch(r"sha256:[0-9a-f]{64}", candidate["artifact_digest"])), "invalid candidate artifact identity")
|
||||
require(sha(candidate["workflow_sha"]) and candidate["workflow_ref"] == "main", "candidate workflow source is invalid")
|
||||
validate_manifest(manifest, {"id": manifest["build_run_id"], "run_attempt": manifest["build_run_attempt"], "head_sha": candidate["workflow_sha"]})
|
||||
return chain
|
||||
|
||||
|
||||
def consume(chain):
|
||||
manifest = chain["candidate"]["manifest"]
|
||||
with open(os.environ["GITHUB_ENV"], "a") as output:
|
||||
# Every pinned installer already verifies these hashes before dpkg.
|
||||
for key, value in (("RUSTFS_NIGHTLY_PACKAGE_URL", manifest["package_url"]),
|
||||
("PACKAGE_SHA256", manifest["package_sha256"]), ("TO_SHA256", manifest["package_sha256"])):
|
||||
output.write(key + "=" + value + "\n")
|
||||
with open(os.environ["GITHUB_OUTPUT"], "a") as output:
|
||||
output.write("testing_sha=" + chain["testing_sha"] + "\n")
|
||||
|
||||
|
||||
def report_counts(text, performance=False):
|
||||
counts = {"PASS": 0, "FAIL": 0, "SKIP": 0, "UNSUPPORTED": 0, "RUNNING": 0}
|
||||
if performance:
|
||||
rows = list(csv.DictReader(io.StringIO(text), delimiter="\t"))
|
||||
seen = set()
|
||||
for row in rows:
|
||||
key = (row.get("method"), row.get("size"))
|
||||
require(key[0] in ("get", "put", "mixed") and key[1] and key not in seen, "invalid or duplicate performance round")
|
||||
seen.add(key)
|
||||
fields = ("throughput", "obj_per_s", "req_avg", "req_p50")
|
||||
if key[0] != "mixed":
|
||||
fields += ("req_p90", "req_p99")
|
||||
require(all(isinstance(row.get(field), str) and row[field].strip() for field in fields), "missing benchmark metrics")
|
||||
counts["PASS"] = len(rows)
|
||||
return counts
|
||||
column = None
|
||||
for line in text.splitlines():
|
||||
if not line.startswith("|"):
|
||||
column = None
|
||||
continue
|
||||
cells = [cell.strip().strip("*") for cell in line.strip().strip("|").split("|")]
|
||||
for label in ("Status", "Result"):
|
||||
if cells[0] in ("ID", "Case", "Topology", "Step") and label in cells:
|
||||
column = cells.index(label)
|
||||
break
|
||||
else:
|
||||
if column is not None:
|
||||
require(len(cells) > column, "incomplete report row")
|
||||
status = cells[column]
|
||||
if re.fullmatch(r":?-+:?", status):
|
||||
continue
|
||||
require(status in counts, "unknown case result")
|
||||
counts[status] += 1
|
||||
return counts
|
||||
|
||||
|
||||
def record(chain, suite, report, output):
|
||||
require(suite in SUITES, "unknown suite")
|
||||
result = {"schema": 1, "suite": suite, "chain": chain, "valid": False, "counts": {}, "report_sha256": None}
|
||||
error = None
|
||||
try:
|
||||
private_head = subprocess.check_output(["git", "-C", "auto-testing", "rev-parse", "HEAD"], cwd=ROOT, text=True).strip()
|
||||
require(private_head == chain["testing_sha"], "suite used a different private script revision")
|
||||
require(report.is_file() and 0 < report.stat().st_size <= MAX_REPORT, "missing, empty or oversized report")
|
||||
data = report.read_bytes()
|
||||
result["report_sha256"] = hashlib.sha256(data).hexdigest()
|
||||
result["counts"] = report_counts(data.decode("utf-8"), suite == "performance")
|
||||
require(result["counts"]["PASS"] > 0 and not result["counts"]["FAIL"] and not result["counts"]["RUNNING"], "no passing executions or incomplete/failed cases")
|
||||
require(all(os.environ[key] == "success" for key in ("CHAIN_JOB_STATUS", "CHAIN_TEST_OUTCOME", "CHAIN_REPORT_OUTCOME")), "suite, report or job did not succeed")
|
||||
result["valid"] = True
|
||||
except (OSError, ValueError, subprocess.SubprocessError) as exc:
|
||||
error = exc
|
||||
output.parent.mkdir(parents=True, exist_ok=False)
|
||||
output.write_text(json.dumps(result, sort_keys=True) + "\n")
|
||||
if error:
|
||||
raise error
|
||||
|
||||
|
||||
def aggregate(chain, directory, needs):
|
||||
require(set(needs) == set(SUITES), "aggregate is missing a required lane")
|
||||
require(all(value.get("result") == "success" for value in needs.values()), "a required suite did not succeed")
|
||||
require({path.name for path in directory.iterdir()} == {suite + ".json" for suite in SUITES}, "missing or unexpected suite evidence")
|
||||
records = [json.loads((directory / (suite + ".json")).read_text()) for suite in SUITES]
|
||||
validate_records(chain, records)
|
||||
return {"schema": 1, "chain": chain, "suites": records, "complete": True, "completed_at": datetime.now(timezone.utc).isoformat()}
|
||||
|
||||
|
||||
def validate_records(chain, records):
|
||||
require(isinstance(records, list) and len(records) == len(SUITES), "missing suite evidence")
|
||||
require([record.get("suite") for record in records] == list(SUITES), "missing, duplicate or reordered suite evidence")
|
||||
for suite, result in zip(SUITES, records):
|
||||
require(type(result.get("schema")) is int and result["schema"] == 1 and result.get("suite") == suite and result.get("chain") == chain, "suite evidence identity mismatch")
|
||||
require(result.get("valid") is True and sha(result.get("report_sha256"), 64), "suite evidence is invalid")
|
||||
counts = result.get("counts", {})
|
||||
require(set(counts) == {"PASS", "FAIL", "SKIP", "UNSUPPORTED", "RUNNING"}, "missing suite counts")
|
||||
require(all(type(value) is int and value >= 0 for value in counts.values()) and counts["PASS"] > 0 and counts["FAIL"] == counts["RUNNING"] == 0, "suite has no complete passing evidence")
|
||||
|
||||
|
||||
def main():
|
||||
parser = argparse.ArgumentParser(description=__doc__)
|
||||
parser.add_argument("mode", choices=("consume", "record", "aggregate"))
|
||||
parser.add_argument("--suite", choices=SUITES)
|
||||
parser.add_argument("--report", type=Path)
|
||||
parser.add_argument("--output", type=Path)
|
||||
parser.add_argument("--directory", type=Path)
|
||||
args = parser.parse_args()
|
||||
chain = current_chain()
|
||||
if args.mode == "consume":
|
||||
consume(chain)
|
||||
elif args.mode == "record":
|
||||
record(chain, args.suite, args.report, args.output)
|
||||
else:
|
||||
needs = json.loads(os.environ["CHAIN_NEEDS"])
|
||||
needs.pop("prepare", None)
|
||||
result = aggregate(chain, args.directory, needs)
|
||||
args.output.write_text(json.dumps(result, sort_keys=True) + "\n")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
@@ -0,0 +1,147 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Report the latest chain attempt separately from verified complete successes."""
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import base64
|
||||
from datetime import datetime, timedelta, timezone
|
||||
import json
|
||||
from pathlib import Path
|
||||
import subprocess
|
||||
|
||||
from functional_chain_evidence import validate_records
|
||||
from resolve_functional_candidate import REPOSITORY, api, read_json_artifact, require, resolve, sha
|
||||
|
||||
WORKFLOW = "rustfs-functional-chain.yml"
|
||||
MAX_AGE = timedelta(hours=36)
|
||||
|
||||
|
||||
def timestamp(value):
|
||||
require(isinstance(value, str), "missing evidence timestamp")
|
||||
parsed = datetime.fromisoformat(value.replace("Z", "+00:00"))
|
||||
require(parsed.tzinfo is not None, "evidence timestamp has no timezone")
|
||||
return parsed
|
||||
|
||||
|
||||
def validate_summary(summary, run):
|
||||
require(isinstance(summary, dict) and type(summary.get("schema")) is int and summary["schema"] == 1 and summary.get("complete") is True, "unsupported complete-chain evidence")
|
||||
chain = summary["chain"]
|
||||
require(chain["run_id"] == run["id"] and chain["attempt"] == run["run_attempt"] and chain["workflow_sha"] == run["head_sha"], "complete evidence belongs to another run attempt")
|
||||
require(sha(chain["testing_sha"]), "missing test-script pin")
|
||||
validate_records(chain, summary["suites"])
|
||||
candidate = chain["candidate"]
|
||||
manifest = candidate["manifest"]
|
||||
require(resolve(manifest["build_run_id"], manifest["build_run_attempt"]) == candidate, "producer candidate identity changed")
|
||||
config = api(f"repos/{REPOSITORY}/contents/.config/functional-script-revision.txt?ref={run['head_sha']}")
|
||||
require(base64.b64decode(config["content"]).decode().strip() == chain["testing_sha"], "private pin differs from workflow source")
|
||||
completed = timestamp(summary["completed_at"])
|
||||
require(timestamp(run["run_started_at"]) <= completed <= datetime.now(timezone.utc) + timedelta(minutes=5), "invalid completion timestamp")
|
||||
source_ref = manifest.get("source_ref", candidate["workflow_ref"])
|
||||
return {"run_id": run["id"], "attempt": run["run_attempt"], "url": run["html_url"],
|
||||
"workflow_sha": chain["workflow_sha"], "testing_sha": chain["testing_sha"],
|
||||
"candidate": candidate, "source_ref": source_ref, "source_sha": manifest["source_sha"],
|
||||
"completed_at": completed.isoformat(), "expires_at": (timestamp(candidate["build_started_at"]) + MAX_AGE).isoformat(),
|
||||
"verified_at": datetime.now(timezone.utc).isoformat(), "evidence_schema": 1}
|
||||
|
||||
|
||||
def complete_success(run):
|
||||
require(run["path"] == ".github/workflows/" + WORKFLOW and run["head_branch"] == "main", "unexpected chain workflow source")
|
||||
require((run.get("head_repository") or {}).get("full_name") == REPOSITORY, "chain came from another repository")
|
||||
require(run.get("status") == "completed" and run.get("conclusion") == "success", "chain has not completed successfully")
|
||||
name = f"functional-chain-complete-{run['id']}-{run['run_attempt']}"
|
||||
payload = api(f"repos/{REPOSITORY}/actions/runs/{run['id']}/artifacts?per_page=100")
|
||||
require(payload["total_count"] <= 100, "chain artifact listing is incomplete")
|
||||
artifacts = [item for item in payload["artifacts"] if item.get("name") == name]
|
||||
require(len(artifacts) == 1, "missing or ambiguous complete-chain artifact")
|
||||
artifact = artifacts[0]
|
||||
require(type(artifact.get("size_in_bytes")) is int and 0 < artifact["size_in_bytes"] <= 1024 * 1024, "complete evidence size is invalid")
|
||||
archive = api(f"repos/{REPOSITORY}/actions/artifacts/{artifact['id']}/zip", binary=True)
|
||||
summary = read_json_artifact(archive, artifact, run, name, "chain-complete.json", max_json=128 * 1024)
|
||||
result = validate_summary(summary, run)
|
||||
result["artifact_id"] = artifact["id"]
|
||||
result["artifact_digest"] = artifact["digest"]
|
||||
return result
|
||||
|
||||
|
||||
def collect(limit=20):
|
||||
observed = datetime.now(timezone.utc)
|
||||
workflow = api(f"repos/{REPOSITORY}/actions/workflows/{WORKFLOW}")
|
||||
runs = api(f"repos/{REPOSITORY}/actions/workflows/{WORKFLOW}/runs?branch=main&per_page={limit}")["workflow_runs"]
|
||||
runs.sort(key=lambda run: timestamp(run["run_started_at"]), reverse=True)
|
||||
result = {"schema": 1, "observed_at": observed.isoformat(), "workflow_state": workflow["state"],
|
||||
"owner": "@overtrue", "scan_limit": limit, "inspection_complete": True,
|
||||
"latest_attempt": None, "last_complete_success": {}, "healthy": False}
|
||||
for index, listed in enumerate(runs):
|
||||
run = api(f"repos/{REPOSITORY}/actions/runs/{listed['id']}/attempts/{listed['run_attempt']}")
|
||||
if index == 0:
|
||||
result["latest_attempt"] = {"run_id": run["id"], "attempt": run["run_attempt"], "url": run["html_url"],
|
||||
"status": run["status"], "conclusion": run["conclusion"], "verification": "not_complete"}
|
||||
if run.get("conclusion") != "success":
|
||||
continue
|
||||
try:
|
||||
complete = complete_success(run)
|
||||
except (OSError, ValueError, KeyError, subprocess.SubprocessError) as error:
|
||||
if index == 0:
|
||||
result["latest_attempt"]["verification"] = "invalid"
|
||||
result["inspection_complete"] = False
|
||||
continue
|
||||
source = complete["source_ref"]
|
||||
if source.startswith("refs/heads/"):
|
||||
source = source[len("refs/heads/"):]
|
||||
if index == 0:
|
||||
result["latest_attempt"]["verification"] = "complete"
|
||||
result["latest_attempt"]["source_ref"] = source
|
||||
result["latest_attempt"]["source_sha"] = complete["source_sha"]
|
||||
previous = result["last_complete_success"].get(source)
|
||||
if previous is None or timestamp(complete["completed_at"]) > timestamp(previous["completed_at"]):
|
||||
result["last_complete_success"][source] = complete
|
||||
# Reject a snapshot if a new attempt started while artifacts were checked.
|
||||
refreshed = api(f"repos/{REPOSITORY}/actions/workflows/{WORKFLOW}/runs?branch=main&per_page={limit}")["workflow_runs"]
|
||||
fields = ("id", "run_attempt", "status", "conclusion", "head_sha", "run_started_at")
|
||||
snapshot = lambda values: sorted(tuple(item.get(key) for key in fields) for item in values)
|
||||
require(snapshot(runs) == snapshot(refreshed), "chain attempts changed during inspection; retry collection")
|
||||
for complete in result["last_complete_success"].values():
|
||||
complete["fresh"] = observed <= timestamp(complete["expires_at"])
|
||||
latest = result["latest_attempt"] or {}
|
||||
result["healthy"] = (result["workflow_state"] == "active" and latest.get("verification") == "complete"
|
||||
and latest.get("source_ref") == "main" and result["last_complete_success"].get("main", {}).get("fresh") is True)
|
||||
return result
|
||||
|
||||
|
||||
def merge_history(current, previous):
|
||||
require(isinstance(previous, dict) and type(previous.get("schema")) is int and previous["schema"] == 1, "unsupported existing dashboard state")
|
||||
require(timestamp(previous["observed_at"]) <= timestamp(current["observed_at"]), "refusing an older dashboard observation")
|
||||
for source, value in previous.get("last_complete_success", {}).items():
|
||||
if source not in current["last_complete_success"] or timestamp(value["completed_at"]) > timestamp(current["last_complete_success"][source]["completed_at"]):
|
||||
value = dict(value)
|
||||
value["fresh"] = timestamp(current["observed_at"]) <= timestamp(value["expires_at"])
|
||||
value["retained_history"] = True
|
||||
current["last_complete_success"][source] = value
|
||||
return current
|
||||
|
||||
|
||||
def publish(result):
|
||||
endpoint = "repos/rustfs/dashboard/contents/src/chain-health.json"
|
||||
existing = api(endpoint)
|
||||
previous = json.loads(base64.b64decode(existing["content"]))
|
||||
merge_history(result, previous)
|
||||
body = {"message": "chore(ci): update functional chain health", "sha": existing["sha"],
|
||||
"content": base64.b64encode((json.dumps(result, indent=2) + "\n").encode()).decode()}
|
||||
subprocess.run(["gh", "api", "--method", "PUT", endpoint, "--input", "-"], input=json.dumps(body), text=True, check=True, capture_output=True, timeout=60)
|
||||
|
||||
|
||||
def main():
|
||||
parser = argparse.ArgumentParser(description=__doc__)
|
||||
parser.add_argument("--output", type=Path, required=True)
|
||||
parser.add_argument("--publish", action="store_true")
|
||||
args = parser.parse_args()
|
||||
result = collect()
|
||||
if args.publish:
|
||||
publish(result)
|
||||
args.output.write_text(json.dumps(result, indent=2) + "\n")
|
||||
print(json.dumps({key: result[key] for key in ("workflow_state", "inspection_complete", "healthy", "latest_attempt")}))
|
||||
return 0 if result["healthy"] else 1
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
raise SystemExit(main())
|
||||
@@ -0,0 +1,124 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Resolve a nightly manifest from its exact GitHub build attempt and artifact."""
|
||||
from __future__ import annotations
|
||||
|
||||
import hashlib
|
||||
import io
|
||||
import json
|
||||
import os
|
||||
from pathlib import Path
|
||||
import re
|
||||
import subprocess
|
||||
import zipfile
|
||||
|
||||
REPOSITORY = "rustfs/rustfs"
|
||||
ROOT = Path(__file__).resolve().parents[1]
|
||||
MAX_ARCHIVE = 1024 * 1024
|
||||
|
||||
|
||||
def require(condition, message):
|
||||
if not condition:
|
||||
raise ValueError(message)
|
||||
|
||||
|
||||
def sha(value, length=40):
|
||||
return isinstance(value, str) and re.fullmatch(r"[0-9a-f]{%d}" % length, value) is not None
|
||||
|
||||
|
||||
def positive(value):
|
||||
return type(value) is int and value > 0
|
||||
|
||||
|
||||
def api(path, binary=False):
|
||||
result = subprocess.run(["gh", "api", path], check=True, capture_output=True, timeout=60)
|
||||
return result.stdout if binary else json.loads(result.stdout)
|
||||
|
||||
|
||||
def validate_manifest(manifest, run):
|
||||
require(isinstance(manifest, dict), "manifest must be an object")
|
||||
common = {"schema", "source_sha", "build_run_id", "build_run_attempt", "package_url", "package_sha256"}
|
||||
version = manifest.get("schema")
|
||||
require(type(version) is int and version in (1, 2), "unsupported candidate schema")
|
||||
require(set(manifest) == (common if version == 1 else common | {"workflow_sha", "source_ref"}), "unexpected candidate fields")
|
||||
require(positive(manifest["build_run_id"]) and positive(manifest["build_run_attempt"]), "invalid build identity")
|
||||
require((manifest["build_run_id"], manifest["build_run_attempt"]) == (run["id"], run["run_attempt"]), "candidate belongs to another build attempt")
|
||||
require(sha(manifest["source_sha"]) and sha(manifest["package_sha256"], 64), "invalid candidate hash")
|
||||
if version == 1:
|
||||
require(manifest["source_sha"] == run["head_sha"], "legacy manifest cannot identify a different build source")
|
||||
else:
|
||||
require(manifest["workflow_sha"] == run["head_sha"] and sha(manifest["workflow_sha"]), "candidate workflow SHA differs from artifact provenance")
|
||||
require(isinstance(manifest["source_ref"], str) and bool(re.fullmatch(r"[A-Za-z0-9_./-]{1,200}", manifest["source_ref"])), "invalid build source ref")
|
||||
expected = (f"https://dl.rustfs.com/artifacts/rustfs/packages/nightly/runs/{run['id']}/"
|
||||
f"{run['run_attempt']}/{manifest['package_sha256']}/rustfs.deb")
|
||||
require(manifest["package_url"] == expected, "package URL does not bind the run, attempt and checksum")
|
||||
return manifest
|
||||
|
||||
|
||||
def read_json_artifact(archive, artifact, run, expected_name, member, max_json=16384):
|
||||
require(artifact.get("expired") is False, "candidate artifact expired")
|
||||
require(positive(artifact.get("id")), "invalid artifact id")
|
||||
require(artifact.get("name") == expected_name, "candidate artifact belongs to another attempt")
|
||||
provenance = artifact.get("workflow_run") or {}
|
||||
require(provenance.get("id") == run["id"] and provenance.get("head_sha") == run["head_sha"], "candidate artifact belongs to another workflow run")
|
||||
require(0 < len(archive) <= MAX_ARCHIVE and artifact.get("size_in_bytes") == len(archive), "candidate artifact size mismatch")
|
||||
require(artifact.get("digest") == "sha256:" + hashlib.sha256(archive).hexdigest(), "candidate artifact checksum mismatch")
|
||||
with zipfile.ZipFile(io.BytesIO(archive)) as source:
|
||||
files = source.infolist()
|
||||
require(len(files) == 1 and files[0].filename == member, "unexpected candidate archive members")
|
||||
require(0 < files[0].file_size <= max_json and not files[0].is_dir(), "candidate manifest too large or empty")
|
||||
manifest = json.loads(source.read(files[0]))
|
||||
return manifest
|
||||
|
||||
|
||||
def read_manifest(archive, artifact, run):
|
||||
name = f"nightly-candidate-{run['id']}-{run['run_attempt']}"
|
||||
return validate_manifest(read_json_artifact(archive, artifact, run, name, name + ".json"), run)
|
||||
|
||||
|
||||
def resolve(run_id, attempt):
|
||||
require(positive(run_id) and positive(attempt), "build run and attempt are required positive integers")
|
||||
endpoint = f"repos/{REPOSITORY}/actions/runs/{run_id}"
|
||||
run = api(f"{endpoint}/attempts/{attempt}")
|
||||
require(run.get("id") == run_id and run.get("run_attempt") == attempt, "GitHub returned a different build attempt")
|
||||
require(run.get("path") == ".github/workflows/nightly-gnu.yml" and run.get("head_branch") == "main", "candidate must come from nightly-gnu on main")
|
||||
require((run.get("head_repository") or {}).get("full_name") == REPOSITORY, "candidate came from another repository")
|
||||
require(run.get("event") in ("schedule", "workflow_dispatch") and run.get("status") == "completed" and run.get("conclusion") == "success", "nightly attempt has not completed successfully")
|
||||
name = f"nightly-candidate-{run_id}-{attempt}"
|
||||
artifacts = []
|
||||
for page in range(1, 11):
|
||||
batch = api(f"{endpoint}/artifacts?per_page=100&page={page}")["artifacts"]
|
||||
artifacts.extend(item for item in batch if item.get("name") == name)
|
||||
if len(batch) < 100:
|
||||
break
|
||||
else:
|
||||
raise ValueError("too many build artifacts to resolve safely")
|
||||
require(len(artifacts) == 1, "missing or ambiguous candidate artifact")
|
||||
artifact = artifacts[0]
|
||||
require(type(artifact.get("size_in_bytes")) is int and 0 < artifact["size_in_bytes"] <= MAX_ARCHIVE, "candidate artifact size is invalid")
|
||||
archive = api(f"repos/{REPOSITORY}/actions/artifacts/{artifact['id']}/zip", binary=True)
|
||||
manifest = read_manifest(archive, artifact, run)
|
||||
return {"manifest": manifest, "artifact_id": artifact["id"], "artifact_digest": artifact["digest"],
|
||||
"workflow_sha": run["head_sha"], "workflow_ref": run["head_branch"], "build_started_at": run["run_started_at"]}
|
||||
|
||||
|
||||
def prepare():
|
||||
event = json.loads(Path(os.environ["GITHUB_EVENT_PATH"]).read_text())
|
||||
if os.environ["GITHUB_EVENT_NAME"] == "workflow_run":
|
||||
build = event["workflow_run"]
|
||||
require(build.get("event") == "schedule", "automatic chain requires a scheduled build")
|
||||
run_id, attempt = build["id"], build["run_attempt"]
|
||||
else:
|
||||
run_id, attempt = int(os.environ["BUILD_RUN_ID"]), int(os.environ["BUILD_RUN_ATTEMPT"])
|
||||
candidate = resolve(run_id, attempt)
|
||||
revision = (ROOT / ".config/functional-script-revision.txt").read_text().strip()
|
||||
require(sha(revision), "private test script revision must be pinned")
|
||||
chain = {"schema": 1, "run_id": int(os.environ["GITHUB_RUN_ID"]), "attempt": int(os.environ["GITHUB_RUN_ATTEMPT"]),
|
||||
"workflow_sha": os.environ["GITHUB_SHA"], "testing_sha": revision, "candidate": candidate}
|
||||
encoded = json.dumps(chain, sort_keys=True, separators=(",", ":"))
|
||||
with open(os.environ["GITHUB_OUTPUT"], "a") as output:
|
||||
output.write("manifest=" + encoded + "\n")
|
||||
Path(os.environ["CHAIN_OUTPUT"]).write_text(encoded + "\n")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
prepare()
|
||||
@@ -0,0 +1,204 @@
|
||||
#!/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 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(len(result["suites"]), 10)
|
||||
|
||||
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)
|
||||
|
||||
|
||||
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_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"
|
||||
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"])
|
||||
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_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_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)
|
||||
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)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -0,0 +1,99 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Historical successes must not hide incomplete evidence or newer attempts."""
|
||||
import base64
|
||||
import copy
|
||||
from datetime import datetime, timedelta, timezone
|
||||
import json
|
||||
import unittest
|
||||
from unittest import mock
|
||||
|
||||
import functional_chain_health as health
|
||||
from functional_chain_evidence import SUITES
|
||||
|
||||
|
||||
class HealthTests(unittest.TestCase):
|
||||
def setUp(self):
|
||||
now = datetime.now(timezone.utc)
|
||||
self.now = now.isoformat()
|
||||
self.run = {"id": 123, "run_attempt": 2, "head_sha": "a" * 40, "html_url": "https://github.com/rustfs/rustfs/actions/runs/123",
|
||||
"run_started_at": (now - timedelta(hours=1)).isoformat()}
|
||||
self.candidate = {"manifest": {"build_run_id": 456, "build_run_attempt": 3, "source_ref": "release", "source_sha": "b" * 40},
|
||||
"workflow_sha": "c" * 40, "workflow_ref": "main", "build_started_at": (now - timedelta(hours=2)).isoformat()}
|
||||
self.chain = {"run_id": 123, "attempt": 2, "workflow_sha": "a" * 40, "testing_sha": "d" * 40, "candidate": self.candidate}
|
||||
self.summary = {"schema": 1, "complete": True, "chain": self.chain, "completed_at": (now - timedelta(minutes=1)).isoformat(),
|
||||
"suites": [{"schema": 1, "suite": suite, "chain": self.chain, "valid": True, "report_sha256": "f" * 64,
|
||||
"counts": {"PASS": 1, "FAIL": 0, "SKIP": 0, "UNSUPPORTED": 0, "RUNNING": 0}} for suite in SUITES]}
|
||||
self.config = {"content": base64.b64encode(("d" * 40 + "\n").encode()).decode()}
|
||||
|
||||
def validate(self, summary=None, candidate=None, config=None):
|
||||
with mock.patch.object(health, "resolve", return_value=candidate or self.candidate), mock.patch.object(health, "api", return_value=config or self.config):
|
||||
return health.validate_summary(summary or self.summary, self.run)
|
||||
|
||||
def test_release_success_preserves_both_sources(self):
|
||||
result = self.validate()
|
||||
self.assertEqual(result["source_ref"], "release")
|
||||
self.assertEqual(result["source_sha"], "b" * 40)
|
||||
self.assertEqual(result["workflow_sha"], "a" * 40)
|
||||
self.assertEqual(result["candidate"]["workflow_sha"], "c" * 40)
|
||||
|
||||
def test_substituted_producer_pin_attempt_or_empty_suite_fails(self):
|
||||
with self.assertRaises(ValueError):
|
||||
self.validate(candidate={**self.candidate, "workflow_sha": "e" * 40})
|
||||
with self.assertRaises(ValueError):
|
||||
self.validate(config={"content": base64.b64encode(b"wrong pin").decode()})
|
||||
wrong = copy.deepcopy(self.summary)
|
||||
wrong["chain"]["attempt"] = 1
|
||||
with self.assertRaises(ValueError):
|
||||
self.validate(wrong)
|
||||
wrong = copy.deepcopy(self.summary)
|
||||
wrong["suites"][0]["counts"]["PASS"] = 0
|
||||
with self.assertRaises(ValueError):
|
||||
self.validate(wrong)
|
||||
|
||||
def test_new_failure_retains_last_complete_success_without_becoming_healthy(self):
|
||||
complete = self.validate()
|
||||
previous = {"schema": 1, "observed_at": self.now, "last_complete_success": {"release": complete}}
|
||||
current = {"schema": 1, "observed_at": self.now, "last_complete_success": {}, "healthy": False,
|
||||
"latest_attempt": {"conclusion": "failure"}}
|
||||
result = health.merge_history(current, previous)
|
||||
self.assertFalse(result["healthy"])
|
||||
self.assertEqual(result["latest_attempt"]["conclusion"], "failure")
|
||||
self.assertEqual(result["last_complete_success"]["release"]["source_sha"], "b" * 40)
|
||||
self.assertTrue(result["last_complete_success"]["release"]["retained_history"])
|
||||
|
||||
def test_expired_history_is_not_fresh_and_null_or_stale_state_cannot_publish(self):
|
||||
complete = self.validate()
|
||||
complete["expires_at"] = "2000-01-01T00:00:00Z"
|
||||
previous = {"schema": 1, "observed_at": self.now, "last_complete_success": {"release": complete}}
|
||||
current = {"schema": 1, "observed_at": self.now, "last_complete_success": {}, "healthy": False}
|
||||
self.assertFalse(health.merge_history(current, previous)["last_complete_success"]["release"]["fresh"])
|
||||
with self.assertRaises(ValueError):
|
||||
health.merge_history(current, None)
|
||||
with self.assertRaises(ValueError):
|
||||
health.merge_history({**current, "observed_at": "2000-01-01T00:00:00Z"}, previous)
|
||||
existing = {"sha": "old-blob", "content": base64.b64encode(b"null").decode()}
|
||||
with mock.patch.object(health, "api", return_value=existing), mock.patch.object(health.subprocess, "run") as write:
|
||||
with self.assertRaises(ValueError):
|
||||
health.publish(current)
|
||||
write.assert_not_called()
|
||||
|
||||
def test_collection_rejects_a_concurrent_rerun(self):
|
||||
run = {**self.run, "status": "completed", "conclusion": "failure"}
|
||||
responses = [{"state": "active"}, {"workflow_runs": [run]}, run,
|
||||
{"workflow_runs": [{**run, "run_attempt": 3, "status": "queued"}]}]
|
||||
with mock.patch.object(health, "api", side_effect=responses):
|
||||
with self.assertRaisesRegex(ValueError, "changed during inspection"):
|
||||
health.collect()
|
||||
|
||||
def test_publication_uses_the_read_blob_sha(self):
|
||||
current = {"schema": 1, "observed_at": self.now, "last_complete_success": {}, "healthy": False}
|
||||
existing = {"sha": "reviewed-blob", "content": base64.b64encode(json.dumps(current).encode()).decode()}
|
||||
with mock.patch.object(health, "api", return_value=existing), mock.patch.object(health.subprocess, "run") as write:
|
||||
health.publish(current)
|
||||
body = json.loads(write.call_args.kwargs["input"])
|
||||
self.assertEqual(body["sha"], "reviewed-blob")
|
||||
self.assertFalse(json.loads(base64.b64decode(body["content"]))["healthy"])
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -10,6 +10,7 @@ import tempfile
|
||||
import unittest
|
||||
|
||||
from check_test_wiring import yaml_block
|
||||
from resolve_functional_candidate import validate_manifest
|
||||
|
||||
|
||||
ROOT = Path(__file__).resolve().parents[1]
|
||||
@@ -172,6 +173,7 @@ SH
|
||||
self.assertEqual(self.manifest()["source_sha"], self.sha)
|
||||
self.assertEqual(self.manifest()["workflow_sha"], "f" * 40)
|
||||
self.assertEqual(self.manifest()["source_ref"], "release")
|
||||
validate_manifest(self.manifest(), {"id": 12345, "run_attempt": 1, "head_sha": "f" * 40})
|
||||
|
||||
def test_every_lane_uses_the_same_resolved_source(self):
|
||||
lines = WORKFLOW.read_text().splitlines()
|
||||
|
||||
@@ -133,7 +133,7 @@ class SecurityWorkflowTests(WorkflowSteps, unittest.TestCase):
|
||||
|
||||
def test_workflow_wiring(self) -> None:
|
||||
names = list(self.steps)
|
||||
self.assertLess(names.index("Checkout repository (for the OIDC live gate script)"), names.index("Checkout auto-testing scripts (with retry)"))
|
||||
self.assertLess(names.index("Checkout repository (for the OIDC live gate script)"), names.index("Checkout auto-testing scripts"))
|
||||
self.assertNotIn(" continue-on-error: true", self.job)
|
||||
self.assertIn(" continue-on-error: true", self.steps["Run security suite"])
|
||||
for name in ("Initialize security evidence", "Generate report"):
|
||||
@@ -291,7 +291,7 @@ class SecurityWorkflowTests(WorkflowSteps, unittest.TestCase):
|
||||
cleanup = named_steps(yaml_block(source, "jobs", 0))[cleanup_name]
|
||||
self.assertTrue(any(line.startswith(" if:") and "always()" in line for line in cleanup))
|
||||
|
||||
def test_root_dispatches_only_upgrade_and_replication_hands_off_after_failure(self) -> None:
|
||||
def test_legacy_replication_hands_off_after_failure(self) -> None:
|
||||
for failed_attempts, issue_exit, token in ((0, 0, "fixture"), (2, 0, "fixture"), (3, 0, "fixture"), (3, 7, "fixture"), (0, 0, "")):
|
||||
with self.subTest(failed_attempts=failed_attempts, issue_exit=issue_exit, token=bool(token)):
|
||||
self.setUp()
|
||||
@@ -337,16 +337,6 @@ fi
|
||||
RUSTFS_NIGHTLY_PACKAGE_URL="https://example.invalid/package.deb",
|
||||
)
|
||||
self.context.update({"secrets.PF_TESTING_GH_TOKEN": "fixture", "inputs.suite": "all"})
|
||||
driver = (ROOT / ".github/workflows/rustfs-functional-chain.yml").read_text()
|
||||
self.steps = named_steps(yaml_block(driver.splitlines(), "start-chain", 2))
|
||||
self.assertEqual(list(self.steps), ["Dispatch first suite (upgrade)"])
|
||||
started = self.run_step("Dispatch first suite (upgrade)")
|
||||
self.assertEqual(started.returncode, 0, started.stderr)
|
||||
self.assertEqual(dispatches.read_text().splitlines(), [
|
||||
"api --method POST repos/rustfs/rustfs/dispatches -f event_type=rustfs-chain-upgrade -F client_payload[from_suite]=nightly-build",
|
||||
])
|
||||
dispatches.unlink()
|
||||
|
||||
replication = (ROOT / ".github/workflows/rustfs-replication-test.yml").read_text()
|
||||
job = yaml_block(replication.splitlines(), "replication-test", 2)
|
||||
self.assertFalse(any(line.startswith(" continue-on-error:") for line in job))
|
||||
@@ -570,10 +560,15 @@ class FunctionalEvidenceTests(WorkflowSteps, unittest.TestCase):
|
||||
self.prepare(suite)
|
||||
self.assertNotIn("/tmp/rustfs-", self.source)
|
||||
names = list(self.steps)
|
||||
self.assertLess(names.index("Initialize functional evidence"), names.index("Checkout auto-testing scripts (with retry)"))
|
||||
self.assertLess(names.index("Initialize functional evidence"), names.index("Checkout auto-testing scripts"))
|
||||
if suite in FunctionalWorkflowTests.DIRECT_TESTS:
|
||||
self.assertLess(names.index("Checkout repository (for report parser)"), names.index("Checkout auto-testing scripts (with retry)"))
|
||||
self.assertLess(names.index("Checkout repository (for report parser)"), names.index("Checkout auto-testing scripts"))
|
||||
for name, lines in self.steps.items():
|
||||
if name == "Upload chain evidence":
|
||||
self.assertIn(" if: ${{ always() && steps.chain_record.outcome == 'success' }}", lines)
|
||||
self.assertIn(" if: ${{ always() && inputs.chain_manifest != '' && steps.evidence.outcome == 'success' }}", self.steps["Record chain evidence"])
|
||||
self.assertIn(" if-no-files-found: error", lines)
|
||||
continue
|
||||
if name in ("Generate report", "Upload functional report to dashboard") or any("uses: actions/upload-artifact@" in line for line in lines):
|
||||
self.assertIn(" if: ${{ always() && steps.evidence.outcome == 'success' }}", lines)
|
||||
if any("uses: actions/upload-artifact@" in line for line in lines):
|
||||
|
||||
Reference in New Issue
Block a user