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

Merge branch 'main' into houseme/test/scanner-heal-v2-w20

Browse Source
This commit is contained in:
houseme
2026-09-06 01:33:36 +08:00
committed by GitHub
parent 1a49f19574 8fc1c9281e
commit 20414b3cba
71 changed files with 3739 additions and 1401 deletions
Download Patch File Download Diff File Expand all files Collapse all files
scripts/check_test_wiring.py
+9 -5
View File
@@ -567,7 +567,7 @@ def check_quick_checks(root: Path) -> list[str]:
errors.append(f"{relative}: missing composite action")
return errors
steps = yaml_block(runs, "steps", 2) or []
for command in ("shellcheck --version && actionlint", "./scripts/check_error_other_format_ratchet.sh"):
for command in ("shellcheck --version && actionlint", "./scripts/check_error_other_format_ratchet.sh", "make script-tests"):
step = workflow_step_block(steps, command, key="run", indent=4)
if step is None:
errors.append(f"{relative}: missing direct execution of {command}")
@@ -906,6 +906,7 @@ class SelfTests(unittest.TestCase):
" - name: Lint workflows\n shell: bash\n run: shellcheck --version && actionlint\n"
" - name: Error format ratchet\n shell: bash\n"
" run: ./scripts/check_error_other_format_ratchet.sh\n"
" - name: Script tests\n shell: bash\n run: make script-tests\n"
)
sources = {
".github/workflows/ci.yml": caller.replace(
@@ -964,6 +965,8 @@ class SelfTests(unittest.TestCase):
"only installed actionlint": action.replace("run: shellcheck --version && actionlint", "run: echo actionlint"),
"missing shellcheck preflight": action.replace("shellcheck --version && ", ""),
"missing ratchet": action.replace("run: ./scripts/check_error_other_format_ratchet.sh", "run: echo skipped"),
"missing script tests": action.replace("run: make script-tests", "run: echo skipped"),
"swallowed script failure": action.replace("run: make script-tests", "run: make script-tests || true"),
"swallowed lint failure": action.replace("&& actionlint", "&& actionlint || true"),
"swallowed ratchet failure": action.replace("ratchet.sh", "ratchet.sh || true"),
"conditional lint": action.replace("run: shellcheck", "if: false\n run: shellcheck"),
@@ -973,7 +976,7 @@ class SelfTests(unittest.TestCase):
"name: Lint workflows", "name: |\n run: shellcheck --version && actionlint"
).replace("\n run: shellcheck --version && actionlint\n", "\n run: shellcheck --version && actionlint\n || true\n"),
}
for command in ("shellcheck --version && actionlint", "./scripts/check_error_other_format_ratchet.sh"):
for command in ("shellcheck --version && actionlint", "./scripts/check_error_other_format_ratchet.sh", "make script-tests"):
for key in ("'if' : false", '"if": false', "'continue-on-error': true", '"continue-on-error" : true'):
mutations[f"quoted {command} {key}"] = action.replace(f"run: {command}", f"{key}\n run: {command}")
for separator in ("", "\n", " # continued command\n"):
@@ -994,9 +997,10 @@ class SelfTests(unittest.TestCase):
root = Path(tmp)
(root / "scripts").mkdir()
commands = ("shellcheck", "actionlint", "./scripts/check_error_other_format_ratchet.sh")
for failing in commands:
(root / "Makefile").write_text(".PHONY: script-tests\nscript-tests:\n\texit 17\n")
for failing in (*commands, "make script-tests"):
with self.subTest(command=failing):
run = "shellcheck --version && actionlint" if failing != commands[-1] else failing
run = "shellcheck --version && actionlint" if failing in ("shellcheck", "actionlint") else failing
step = workflow_step_block(steps, run, key="run", indent=4)
self.assertIsNotNone(step)
run_index = next(index for index, line in enumerate(step[1]) if line.startswith(" run:"))
@@ -1011,7 +1015,7 @@ class SelfTests(unittest.TestCase):
cwd=root, env=dict(os.environ, PATH=f"{root}{os.pathsep}{os.environ['PATH']}"),
capture_output=True, text=True,
)
self.assertEqual(result.returncode, 17, result.stderr)
self.assertEqual(result.returncode, 2 if failing == "make script-tests" else 17, result.stderr)
def test_validate_includes_quick_checks(self) -> None:
error = "Quick Checks wiring regression"
scripts/functional_case_report.py
+77
View File
@@ -0,0 +1,77 @@
#!/usr/bin/env python3
"""Preserve every functional case execution and its suite context in reports."""
from __future__ import annotations
import argparse
from pathlib import Path
import re
def generate_report(log_file: Path, case_file: Path, matrix_file: Path | None = None) -> bool:
ansi = re.compile(r"\x1b\[[0-9;]*m")
start_re = re.compile(r"^---\s+([A-Z][A-Z0-9]*-[0-9]+)\s+(.+?)\s+---$")
done_re = re.compile(r"^\[(PASS|FAIL|UNSUPPORTED)\]\s+([A-Z][A-Z0-9]*-[0-9]+)\b")
context_re = re.compile(r"^(?:\[INFO\]\s+)?==\s+((?:topology|suite):.+?)\s+==$")
topo_re = re.compile(r"^\[UPG-TOPO\]\s+(\S+)\s+(\S+)\s+(\S+)\s+(\S+)\s+PASS=(\d+)\s+FAIL=(\d+)\s*$")
rows = []
pending = {}
topo_rows = []
context = "context not recorded"
complete = True
try:
with log_file.open(encoding="utf-8", errors="replace") as log:
for raw in log:
line = ansi.sub("", raw).strip()
if match := context_re.match(line):
context = match[1]
pending.clear()
elif match := topo_re.match(line):
topo_rows.append(match.groups())
elif match := start_re.match(line):
case_id, name = match.groups()
pending[case_id] = len(rows)
rows.append([case_id, f"{name} ({context})", "RUNNING"])
elif match := done_re.match(line):
status, case_id = match.groups()
index = pending.pop(case_id, None)
if index is None:
complete = False
rows.append([case_id, f"{case_id} ({context}; start not recorded)", status])
else:
rows[index][2] = status
except FileNotFoundError:
pass
counts = {status: sum(row[2] == status for row in rows) for status in ("PASS", "FAIL", "UNSUPPORTED", "RUNNING")}
with case_file.open("w", encoding="utf-8") as out:
out.write(f"## Case Summary\n\n- Total: {len(rows)}\n")
for status, count in counts.items():
out.write(f"- {status}: {count}\n")
out.write("\n| Case | Name | Status |\n| --- | --- | --- |\n")
for row in rows:
out.write("| " + " | ".join(value.replace("|", "|") for value in row) + " |\n")
if not rows:
out.write("\nNo case execution was recorded; the log is missing, empty, or stopped before the cases.\n")
valid = complete and bool(rows) and not counts["FAIL"] and not counts["RUNNING"]
if matrix_file is not None:
with matrix_file.open("w", encoding="utf-8") as out:
out.write("## Upgrade Matrix\n\n| Topology | KMS Backend | From Version | To Version | Result |\n")
out.write("| --- | --- | --- | --- | --- |\n")
for topo, backend, old_v, new_v, npass, nfail in topo_rows:
result = "PASS" if nfail == "0" else "FAIL"
out.write(f"| {topo} | {backend} | {old_v} | {new_v} | {result} (PASS={npass} FAIL={nfail}) |\n")
if not topo_rows:
out.write("| - | - | - | - | NOT RUN (suite failed before upgrade) |\n")
valid = valid and bool(topo_rows) and all(row[-1] == "0" for row in topo_rows)
return valid
if __name__ == "__main__":
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("log_file", type=Path)
parser.add_argument("case_file", type=Path)
parser.add_argument("matrix_file", type=Path, nargs="?")
args = parser.parse_args()
raise SystemExit(0 if generate_report(args.log_file, args.case_file, args.matrix_file) else 1)
scripts/test/oidc_keycloak_live.sh
+1 -3
View File
@@ -187,7 +187,7 @@ values = {}
for element in root.iter():
values[element.tag.rsplit("}", 1)[-1]] = element.text or ""
for field in ("AccessKeyId", "SecretAccessKey", "SessionToken", "Expiration", "SubjectFromWebIdentityToken"):
assert values.get(field), values
assert values.get(field), f"missing required STS field: {field}"
print("\t".join(values[field] for field in ("AccessKeyId", "SecretAccessKey", "SessionToken")))
PY
)
@@ -218,7 +218,6 @@ TAMPERED_STATUS="$(curl --noproxy '*' -sS \
--data-urlencode DurationSeconds=900 \
--data-urlencode "WebIdentityToken=${TAMPERED_TOKEN}")"
[[ "${TAMPERED_STATUS}" == 403 ]] || {
cat "${WORK_DIR}/sts-tampered.xml" >&2
echo "expected tampered token to return HTTP 403, got ${TAMPERED_STATUS}" >&2
exit 1
}
@@ -235,7 +234,6 @@ BAD_STATUS="$(curl --noproxy '*' -sS \
--data-urlencode DurationSeconds=900 \
--data-urlencode "WebIdentityToken=${BAD_TOKEN}")"
[[ "${BAD_STATUS}" == 403 ]] || {
cat "${WORK_DIR}/sts-bad.xml" >&2
echo "expected wrong-audience token to return HTTP 403, got ${BAD_STATUS}" >&2
exit 1
}
scripts/test_nightly_candidate.py
+208
View File
@@ -0,0 +1,208 @@
#!/usr/bin/env python3
"""Exercise the nightly publication step without AWS, network or package builds."""
import hashlib
import json
import os
from pathlib import Path
import subprocess
import tempfile
import unittest
from check_test_wiring import yaml_block
ROOT = Path(__file__).resolve().parents[1]
WORKFLOW = ROOT / ".github/workflows/nightly-gnu.yml"
class NightlyCandidateTests(unittest.TestCase):
def setUp(self):
self.temp = tempfile.TemporaryDirectory()
self.addCleanup(self.temp.cleanup)
self.root = Path(self.temp.name)
self.package = self.root / "rustfs-nightly-2026-09-06.deb"
self.package.write_bytes(b"built package bytes\x00\xff")
for command in (["git", "init", "-q"], ["git", "add", self.package.name],
["git", "-c", "user.name=Test", "-c", "user.email=test@example.invalid", "commit", "-qm", "fixture"]):
subprocess.run(command, cwd=self.root, check=True, capture_output=True)
self.sha = subprocess.check_output(["git", "rev-parse", "HEAD"], cwd=self.root, text=True).strip()
self.digest = hashlib.sha256(self.package.read_bytes()).hexdigest()
self.output = self.root / "github-output"
self.store = self.root / "store"
self.shims = self.root / "fake-tools.sh"
self.shims.write_text(r'''aws() {
printf '%s\n' "$*" >> "$FAKE_AWS_LOG"
if [[ "$1" == --version ]]; then printf 'aws-cli/1.44.79 fixture\n'; return; fi
if [[ "$*" == *--generate-cli-skeleton* ]]; then
if [[ "$FAKE_MODE" == broken-install || "$FAKE_MODE" =~ ^(old-cli|bootstrap-failure|install-failure)$ && ! -e "$FAKE_INSTALLED" ]]; then
printf '{}\n'
else
printf '{"IfNoneMatch":""}\n'
fi
return
fi
if [[ "$1 $2" == 's3api put-object' ]]; then
[[ "$FAKE_MODE" != upload-failure ]] || return 42
shift 2
local key="" body="" condition=""
while [[ $# -gt 0 ]]; do
case "$1" in
--key) key="$2";;
--body) body="$2";;
--if-none-match) condition="$2";;
esac
shift 2
done
[[ -z "$condition" || "$condition" == '*' ]] || return 43
if [[ "$condition" == '*' && -e "$FAKE_STORE/$key" ]]; then return 44; fi
mkdir -p "$(dirname "$FAKE_STORE/$key")"
cp "$body" "$FAKE_STORE/$key"
elif [[ "$1 $2" == 's3 cp' ]]; then
[[ "$FAKE_MODE" != alias-failure ]] || return 45
local destination="${4#s3://test-bucket/}"
[[ "$destination" != */ ]] || destination+="$(basename "$3")"
mkdir -p "$(dirname "$FAKE_STORE/$destination")"
cp "$3" "$FAKE_STORE/$destination"
else
return 46
fi
}
curl() {
local url="${!#}"
printf '%s\n' "$url" >> "$FAKE_CURL_LOG"
[[ "$url" == https://dl.rustfs.com/artifacts/rustfs/packages/nightly/runs/* ]] || return 22
[[ "$FAKE_MODE" != missing-public-url ]] || return 22
if [[ "$FAKE_MODE" == wrong-public-bytes ]]; then printf 'different package'; return; fi
cat "$FAKE_STORE/${url#https://dl.rustfs.com/}" || return 22
[[ "$FAKE_MODE" != incomplete-download ]] || return 47
}
sudo() {
[[ "$*" == 'apt-get update' || "$*" == 'apt-get install -y -qq python3-venv' ]] || return 49
[[ "$FAKE_MODE" != bootstrap-failure ]] || return 48
}
python3() {
[[ "$1 $2" == '-m venv' ]] || return 50
mkdir -p "$3/bin"
cat > "$3/bin/python" <<'SH'
#!/usr/bin/env bash
[[ "$*" == '-m pip install --disable-pip-version-check awscli==1.44.79' ]] || exit 51
[[ "$FAKE_MODE" != install-failure ]] || exit 52
: > "$FAKE_INSTALLED"
SH
printf '#!/usr/bin/env bash\naws "$@"\n' > "$3/bin/aws"
chmod +x "$3/bin/python" "$3/bin/aws"
}
''')
self.env = dict(os.environ, BASH_ENV=str(self.shims), DEB_FILE=self.package.name,
R2_ACCESS_KEY_ID="fake-access", R2_SECRET_ACCESS_KEY="fake-secret", R2_ENDPOINT="https://r2.example.invalid", R2_BUCKET="test-bucket",
RUNNER_TEMP=str(self.root), GITHUB_SHA=self.sha, GITHUB_RUN_ID="12345", GITHUB_RUN_ATTEMPT="1", GITHUB_OUTPUT=str(self.output),
FAKE_STORE=str(self.store), FAKE_AWS_LOG=str(self.root / "aws.log"), FAKE_CURL_LOG=str(self.root / "curl.log"), FAKE_INSTALLED=str(self.root / "installed"), FAKE_MODE="success")
source = WORKFLOW.read_text()
job = yaml_block(source.splitlines(), "build", 2)
starts = [i for i, line in enumerate(job) if line.startswith(" - name: ")]
self.steps = {
job[start].split(": ", 1)[1]: job[start:end]
for start, end in zip(starts, starts[1:] + [len(job)])
}
self.publish = self.steps["Upload DEB to Cloudflare R2"]
start = self.publish.index(" run: |") + 1
self.shell = "\n".join(line[10:] for line in self.publish[start:] if not line.strip() or line.startswith(" "))
def run_publish(self, **overrides):
self.output.unlink(missing_ok=True)
return subprocess.run(["bash", "--noprofile", "--norc", "-e", "-o", "pipefail", "-c", self.shell],
cwd=self.root, env=dict(self.env, **overrides), capture_output=True, text=True)
def manifest(self):
output = self.output.read_text().strip()
self.assertTrue(output.startswith("candidate_file="), output)
return json.loads(Path(output.split("=", 1)[1]).read_text())
def test_success_binds_actual_package_checkout_and_attempt(self):
result = self.run_publish()
self.assertEqual(result.returncode, 0, result.stderr)
manifest = self.manifest()
self.assertEqual(manifest, {"schema": 1, "source_sha": self.sha, "build_run_id": 12345, "build_run_attempt": 1,
"package_sha256": self.digest, "package_url": f"https://dl.rustfs.com/artifacts/rustfs/packages/nightly/runs/12345/1/{self.digest}/rustfs.deb"})
for path in (f"runs/12345/1/{self.digest}/rustfs.deb", self.package.name, "rustfs-nightly-latest.deb"):
self.assertEqual((self.store / "artifacts/rustfs/packages/nightly" / path).read_bytes(), self.package.read_bytes())
self.assertEqual((self.root / "curl.log").read_text().strip(), manifest["package_url"])
def test_missing_credentials_remain_artifact_only(self):
for key in ("R2_ACCESS_KEY_ID", "R2_SECRET_ACCESS_KEY", "R2_ENDPOINT", "R2_BUCKET"):
with self.subTest(missing=key):
result = self.run_publish(**{key: ""})
self.assertEqual(result.returncode, 0, result.stderr)
self.assertFalse(self.output.exists())
self.assertFalse((self.root / "aws.log").exists())
self.assertEqual(list(self.root.glob("nightly-candidate-*.json")), [])
def test_publication_failures_never_emit_a_candidate(self):
for index, mode in enumerate(("upload-failure", "missing-public-url", "wrong-public-bytes", "incomplete-download", "alias-failure")):
with self.subTest(mode=mode):
result = self.run_publish(FAKE_MODE=mode, GITHUB_RUN_ID=str(20000 + index))
self.assertNotEqual(result.returncode, 0, result.stdout)
self.assertFalse(self.output.exists())
self.assertEqual(list(self.root.glob("nightly-candidate-*.json")), [])
def test_old_cli_is_upgraded_in_an_isolated_temporary_environment(self):
result = self.run_publish(FAKE_MODE="old-cli")
self.assertEqual(result.returncode, 0, result.stderr)
self.assertTrue((self.root / "installed").exists())
self.assertEqual(self.manifest()["package_sha256"], self.digest)
self.assertEqual(list(self.root.glob("nightly-awscli.*")), [])
def test_failed_cli_bootstrap_cannot_publish(self):
for mode in ("bootstrap-failure", "install-failure", "broken-install"):
with self.subTest(mode=mode):
(self.root / "installed").unlink(missing_ok=True)
result = self.run_publish(FAKE_MODE=mode)
self.assertNotEqual(result.returncode, 0)
self.assertFalse(self.output.exists())
self.assertFalse(self.store.exists())
self.assertEqual(list(self.root.glob("nightly-awscli.*")), [])
def test_checkout_sha_mismatch_fails_before_upload(self):
result = self.run_publish(GITHUB_SHA="f" * 40)
self.assertNotEqual(result.returncode, 0)
self.assertIn("Checkout SHA", result.stderr)
self.assertFalse(self.output.exists())
self.assertFalse((self.root / "aws.log").exists())
def test_same_date_builds_and_reruns_keep_distinct_candidates(self):
urls = []
for run, attempt in (("12345", "1"), ("54321", "1"), ("12345", "2")):
result = self.run_publish(GITHUB_RUN_ID=run, GITHUB_RUN_ATTEMPT=attempt)
self.assertEqual(result.returncode, 0, result.stderr)
urls.append(self.manifest()["package_url"])
self.assertEqual(len(set(urls)), 3)
self.assertEqual(len(list(self.root.glob("nightly-candidate-*.json"))), 3)
def test_duplicate_key_is_not_overwritten_or_recertified(self):
result = self.run_publish()
self.assertEqual(result.returncode, 0, result.stderr)
key = self.manifest()["package_url"].removeprefix("https://dl.rustfs.com/")
stored = self.store / key
stored.write_bytes(b"preexisting conflicting object")
(self.root / "nightly-candidate-12345-1.json").unlink()
result = self.run_publish()
self.assertNotEqual(result.returncode, 0)
self.assertEqual(stored.read_bytes(), b"preexisting conflicting object")
self.assertFalse(self.output.exists())
self.assertEqual(list(self.root.glob("nightly-candidate-*.json")), [])
def test_manifest_upload_requires_publication_output(self):
upload = self.steps["Upload nightly candidate manifest"]
self.assertIn(" id: publish", self.publish)
self.assertIn(" DEB_FILE: ${{ steps.deb.outputs.deb_file }}", self.publish)
self.assertIn(" if: ${{ steps.publish.outputs.candidate_file != '' }}", upload)
self.assertIn(" name: nightly-candidate-${{ github.run_id }}-${{ github.run_attempt }}", upload)
self.assertIn(" path: ${{ steps.publish.outputs.candidate_file }}", upload)
self.assertIn(" if-no-files-found: error", upload)
self.assertNotIn(" continue-on-error: true", self.publish)
self.assertNotIn(" overwrite: true", upload)
if __name__ == "__main__":
unittest.main()
scripts/test_python_bin.sh
+8 -14
View File
@@ -62,20 +62,14 @@ exit 1
STUB
chmod +x "$TMP_ROOT/bin/python3"
SANDBOX_PATH="$TMP_ROOT/bin:/usr/bin:/bin"
if PATH="$SANDBOX_PATH" command -v uv >/dev/null 2>&1; then
# uv is reachable even from the sandbox PATH, so the resolver would
# legitimately fall back to it instead of failing. Skip this case.
echo "ℹ️ uv is on the sandbox PATH; skipping the no-interpreter case"
else
if PATH="$SANDBOX_PATH" "$RESOLVER" -c 'pass' \
>"$TMP_ROOT/none.out" 2>"$TMP_ROOT/none.err"; then
fail "resolver succeeded with no usable interpreter on PATH"
fi
grep -q 'No Python 3.11+ interpreter found' "$TMP_ROOT/none.err" \
|| fail "missing-interpreter failure did not name the requirement"
grep -q 'RUSTFS_PYTHON=' "$TMP_ROOT/none.err" \
|| fail "missing-interpreter failure did not point at the override"
ln -s "$(command -v bash)" "$TMP_ROOT/bin/bash"
if PATH="$TMP_ROOT/bin" RUSTFS_PYTHON="" "$RESOLVER" -c 'pass' \
>"$TMP_ROOT/none.out" 2>"$TMP_ROOT/none.err"; then
fail "resolver succeeded with no usable interpreter on PATH"
fi
grep -q 'No Python 3.11+ interpreter found' "$TMP_ROOT/none.err" \
|| fail "missing-interpreter failure did not name the requirement"
grep -q 'RUSTFS_PYTHON=' "$TMP_ROOT/none.err" \
|| fail "missing-interpreter failure did not point at the override"
echo "✅ scripts/python_bin.sh resolver checks passed"
scripts/test_security_workflow.py
+615 -64
View File
@@ -1,16 +1,20 @@
#!/usr/bin/env python3
"""Exercise functional workflow failures and security evidence without remote VMs."""
"""Exercise functional failures, chain dispatch, and security evidence without remote VMs."""
from __future__ import annotations
import glob
import json
import os
import re
import subprocess
import sys
import tempfile
import unittest
from pathlib import Path
from check_test_wiring import yaml_block
from functional_case_report import generate_report
ROOT = Path(__file__).resolve().parents[1]
@@ -38,7 +42,46 @@ def shell_body(lines: list[str]) -> str:
return "\n".join(shell_lines)
class SecurityWorkflowTests(unittest.TestCase):
class WorkflowSteps:
def uploaded_files(self) -> set[Path]:
upload = next(lines for lines in self.steps.values() if any("uses: actions/upload-artifact@" in line for line in lines))
start = upload.index(" path: |") + 1
paths = []
for line in upload[start:]:
if not line.startswith(" "):
break
paths.extend(Path(path) for path in glob.glob(self.render(line.strip())))
return {file for path in paths for file in (path.rglob("*") if path.is_dir() else [path]) if file.is_file()}
def render(self, value: str) -> str:
return re.sub(r"\$\{\{\s*(.*?)\s*\}\}", lambda match: self.context[match[1]], value)
def step_env(self, lines: list[str], indent: int = 8) -> dict[str, str]:
result = {}
for line in yaml_block(lines, "env", indent) or []:
if line.strip() and not line.lstrip().startswith("#"):
key, value = line.strip().split(": ", 1)
result[key] = self.render(value.strip("'\""))
return result
def run_step(self, name: str) -> subprocess.CompletedProcess[str]:
lines = self.steps[name]
result = subprocess.run(
["bash", "--noprofile", "--norc", "-e", "-o", "pipefail", "-c", self.render(shell_body(lines))],
cwd=self.directory, env={**self.env, **self.step_env(lines)}, capture_output=True, text=True,
)
for line in lines:
if line.startswith(" id: "):
self.context[f"steps.{line.split(': ', 1)[1]}.outcome"] = "failure" if result.returncode else "success"
if Path(self.env["GITHUB_ENV"]).exists():
for line in Path(self.env["GITHUB_ENV"]).read_text().splitlines():
key, value = line.split("=", 1)
self.env[key] = value
self.context[f"env.{key}"] = value
return result
class SecurityWorkflowTests(WorkflowSteps, unittest.TestCase):
def setUp(self) -> None:
self.source = WORKFLOW.read_text()
self.job = yaml_block(self.source.splitlines(), "security-test", 2)
@@ -78,6 +121,7 @@ class SecurityWorkflowTests(unittest.TestCase):
'#!/usr/bin/env bash\nset -euo pipefail\n'
'log_dir=$(mktemp -d "$TMPDIR/rustfs-security.XXXXXX")\n'
'echo "CURRENT SUITE LOG" > "$log_dir/suite.log"\n'
'echo "CURRENT SUITE STDOUT"; echo "CURRENT SUITE STDERR" >&2\n'
'case "$FAKE_REPORT" in\n'
f' present) printf "%s\\n" "CURRENT SUITE DIAGNOSTIC" "{CASE_ROW}" > "$REPORT_FILE" ;;\n'
' empty) : > "$REPORT_FILE" ;;\n'
@@ -86,33 +130,6 @@ class SecurityWorkflowTests(unittest.TestCase):
'exit "$FAKE_EXIT"\n'
)
def render(self, value: str) -> str:
return re.sub(r"\$\{\{\s*(.*?)\s*\}\}", lambda match: self.context[match[1]], value)
def step_env(self, lines: list[str], indent: int = 8) -> dict[str, str]:
result = {}
for line in yaml_block(lines, "env", indent) or []:
if line.strip() and not line.lstrip().startswith("#"):
key, value = line.strip().split(": ", 1)
result[key] = self.render(value.strip("'\""))
return result
def run_step(self, name: str) -> subprocess.CompletedProcess[str]:
lines = self.steps[name]
result = subprocess.run(
["bash", "--noprofile", "--norc", "-e", "-o", "pipefail", "-c", self.render(shell_body(lines))],
cwd=self.directory, env={**self.env, **self.step_env(lines)}, capture_output=True, text=True,
)
for line in lines:
if line.startswith(" id: "):
self.context[f"steps.{line.split(': ', 1)[1]}.outcome"] = "failure" if result.returncode else "success"
if Path(self.env["GITHUB_ENV"]).exists():
for line in Path(self.env["GITHUB_ENV"]).read_text().splitlines():
key, value = line.split("=", 1)
self.env[key] = value
self.context[f"env.{key}"] = value
return result
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)"))
@@ -128,7 +145,7 @@ class SecurityWorkflowTests(unittest.TestCase):
for name in ("Upload functional report to dashboard", "Upload report and logs"):
self.assertIn(" if: ${{ always() && steps.evidence.outcome == 'success' }}", self.steps[name])
artifact_settings = yaml_block(self.steps["Upload report and logs"], "with", 8)
self.assertIn(" path: ${{ env.SECURITY_ARTIFACTS_DIR }}/", artifact_settings)
self.assertIn(" path: |", artifact_settings)
self.assertIn(" if-no-files-found: error", artifact_settings)
def test_suite_report_and_result_matrix(self) -> None:
@@ -147,7 +164,7 @@ class SecurityWorkflowTests(unittest.TestCase):
if outcome != "skipped" or mode == "present":
suite = self.run_step("Run security suite")
self.assertEqual(suite.returncode, exit_code, suite.stderr)
logs = list(self.artifacts.glob("rustfs-security.*/suite.log"))
logs = list(Path(str(self.artifacts) + "-scratch").glob("rustfs-security.*/suite.log"))
self.assertEqual(len(logs), 1)
self.assertEqual(logs[0].read_text(), "CURRENT SUITE LOG\n")
self.context["steps.test.outcome"] = outcome
@@ -169,37 +186,197 @@ class SecurityWorkflowTests(unittest.TestCase):
summary = Path(self.env["GITHUB_STEP_SUMMARY"]).read_text()
self.assertEqual(summary, contents)
self.assertNotIn("UNWRAPPED SUITE SUMMARY", summary)
expected = {self.artifacts / "report.md"}
if outcome != "skipped" or mode == "present":
expected.add(self.artifacts / "suite.log")
self.assertEqual((self.artifacts / "suite.log").read_text(), "CURRENT SUITE STDOUT\nCURRENT SUITE STDERR\n")
if mode in ("present", "empty"):
expected.add(self.artifacts / "suite-report.md")
(self.artifacts / "unexpected-token.json").write_text("FAKE-SECRET-CANARY")
scratch = Path(str(self.artifacts) + "-scratch")
(scratch / "case.out").write_text("FAKE-SECRET-CANARY")
self.assertEqual(self.uploaded_files(), expected)
def test_oidc_negative_responses_never_print_issued_credentials(self):
source = (ROOT / "scripts/test/oidc_keycloak_live.sh").read_text()
for variable, filename in (("TAMPERED_STATUS", "sts-tampered.xml"), ("BAD_STATUS", "sts-bad.xml")):
start = source.index('[[ "${' + variable + '}" == 403 ]]')
end = source.index("\n", source.index("grep -q '<Code>AccessDenied</Code>'", start))
guard = source[start:end]
credential_xml = "<Credentials><AccessKeyId>FAKE-ACCESS-CANARY</AccessKeyId><SecretAccessKey>FAKE-SECRET-CANARY</SecretAccessKey><SessionToken>FAKE-SESSION-CANARY</SessionToken></Credentials>"
for status, body, expected in (("200", credential_xml, 1), ("403", credential_xml, 1),
("403", "<Code>AccessDenied</Code>", 0)):
with self.subTest(variable=variable, status=status, expected=expected):
(self.directory / filename).write_text(body)
result = subprocess.run(["bash", "-e", "-o", "pipefail", "-c", guard],
env={**self.env, variable: status, "WORK_DIR": str(self.directory)},
capture_output=True, text=True)
self.assertEqual(result.returncode, expected, result.stderr)
for canary in ("FAKE-ACCESS-CANARY", "FAKE-SECRET-CANARY", "FAKE-SESSION-CANARY"):
self.assertNotIn(canary, result.stdout + result.stderr)
if status == "200":
self.assertIn("HTTP 403, got 200", result.stderr)
def test_oidc_incomplete_credentials_report_only_the_missing_field(self):
source = (ROOT / "scripts/test/oidc_keycloak_live.sh").read_text()
start = source.index("IFS=$'\\t' read -r STS_ACCESS_KEY")
end = source.index("\n)\n", start) + 3
extract = source[start:end]
values = {"AccessKeyId": "FAKE-ACCESS-CANARY", "SecretAccessKey": "FAKE-SECRET-CANARY",
"SessionToken": "FAKE-SESSION-CANARY", "Expiration": "2099-01-01T00:00:00Z",
"SubjectFromWebIdentityToken": "alice"}
for missing in (None, "Expiration", "SubjectFromWebIdentityToken"):
with self.subTest(missing=missing):
xml = "<Credentials>" + "".join(f"<{key}>{value}</{key}>" for key, value in values.items() if key != missing) + "</Credentials>"
(self.directory / "sts-good.xml").write_text(xml)
result = subprocess.run(["bash", "-e", "-o", "pipefail", "-c", extract],
env={**self.env, "WORK_DIR": str(self.directory)}, capture_output=True, text=True)
self.assertEqual(result.returncode, 1 if missing else 0, result.stderr)
for canary in ("FAKE-ACCESS-CANARY", "FAKE-SECRET-CANARY", "FAKE-SESSION-CANARY"):
self.assertNotIn(canary, result.stdout + result.stderr)
if missing:
self.assertIn(f"missing required STS field: {missing}", result.stderr)
def test_existing_evidence_directory_is_rejected(self) -> None:
self.artifacts.mkdir()
stale = self.artifacts / "suite-report.md"
stale.write_text("OLD RUN REPORT")
self.assertNotEqual(self.run_step("Initialize security evidence").returncode, 0)
self.assertEqual(stale.read_text(), "OLD RUN REPORT")
self.assertFalse(Path(self.env["GITHUB_ENV"]).exists())
(self.artifacts / "report.md").write_text("OLD RUN REPORT")
self.context.update({
"env.SECURITY_ARTIFACTS_DIR": str(self.artifacts), "secrets.PF_TESTING_GH_TOKEN": "fake-local-token",
})
fake_bin = self.directory / "bin"
fake_bin.mkdir()
gh = fake_bin / "gh"
gh.write_text(
'#!/usr/bin/env bash\nset -euo pipefail\n'
'if [ "$1 $2" = "issue create" ]; then\n'
' while [ "$#" -gt 0 ]; do\n'
' if [ "$1" = "--body-file" ]; then cat "$2" > "$CAPTURE_BODY"; fi\n'
' shift\n'
' done\n'
'fi\n'
)
gh.chmod(0o755)
body = self.directory / "issue-body.md"
self.env.update(PATH=f"{fake_bin}{os.pathsep}{os.environ['PATH']}", CAPTURE_BODY=str(body))
result = self.run_step("File failure issue in rustfs/backlog")
self.assertEqual(result.returncode, 0, result.stderr)
self.assertNotIn("OLD RUN REPORT", body.read_text())
self.assertIn("https://github.com/rustfs/rustfs/actions/runs/314159", body.read_text())
for suffix in ("", "-scratch"):
self.setUp()
existing = Path(str(self.artifacts) + suffix)
existing.mkdir()
stale = existing / "suite-report.md"
stale.write_text("OLD RUN REPORT")
self.assertNotEqual(self.run_step("Initialize security evidence").returncode, 0)
self.assertEqual(stale.read_text(), "OLD RUN REPORT")
self.assertFalse(Path(self.env["GITHUB_ENV"]).exists())
(self.artifacts / "report.md").write_text("OLD RUN REPORT")
self.context.update({
"env.SECURITY_ARTIFACTS_DIR": str(self.artifacts), "secrets.PF_TESTING_GH_TOKEN": "fake-local-token",
})
fake_bin = self.directory / "bin"
fake_bin.mkdir()
gh = fake_bin / "gh"
gh.write_text(
'#!/usr/bin/env bash\nset -euo pipefail\n'
'if [ "$1 $2" = "issue create" ]; then\n'
' while [ "$#" -gt 0 ]; do\n'
' if [ "$1" = "--body-file" ]; then cat "$2" > "$CAPTURE_BODY"; fi\n'
' shift\n'
' done\n'
'fi\n'
)
gh.chmod(0o755)
body = self.directory / "issue-body.md"
self.env.update(PATH=f"{fake_bin}{os.pathsep}{os.environ['PATH']}", CAPTURE_BODY=str(body))
result = self.run_step("File failure issue in rustfs/backlog")
self.assertEqual(result.returncode, 0, result.stderr)
self.assertNotIn("OLD RUN REPORT", body.read_text())
self.assertIn("https://github.com/rustfs/rustfs/actions/runs/314159", body.read_text())
def test_all_ten_suites_hold_the_shared_lock_for_manual_and_chain_runs(self) -> None:
for suite in ("upgrade", "s3-compat", "kms", "tier", "storage", "heal", "pool-expand", "security", "replication", "performance"):
with self.subTest(suite=suite):
source = (ROOT / f".github/workflows/rustfs-{suite}-test.yml").read_text().splitlines()
# Workflow-level concurrency covers every job, including cleanup,
# regardless of trigger or the runner hosting the job.
self.assertEqual([
line.strip() for line in yaml_block(source, "concurrency", 0)
if line.strip() and not line.lstrip().startswith("#")
], [
"group: rustfs-shared-functional-tests", "cancel-in-progress: false",
])
self.assertIsNotNone(yaml_block(source, "workflow_dispatch", 2))
self.assertIsNotNone(yaml_block(source, "repository_dispatch", 2))
cleanup_name = "Reset test environment (after)" if suite == "performance" else "Cleanup environment (after)"
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:
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()
fake_bin = self.directory / "bin"
fake_bin.mkdir()
commands = {
"gh": '''#!/usr/bin/env bash
set -euo pipefail
if [ "$1" = api ]; then
printf '%s\\n' "$*" >> "$DISPATCHES"
attempt=$(wc -l < "$DISPATCHES")
[ "$attempt" -gt "$FAILED_ATTEMPTS" ]
elif [ "$1 $2" = 'issue create' ]; then
printf 'issue\\n' >> "$EXECUTED"
while [ "$#" -gt 0 ]; do
if [ "$1" = --body-file ]; then
cat "$2" > "$CAPTURE_BODY"
printf '%s\\n' "$2" > "$CAPTURE_BODY_PATH"
fi
shift
done
exit "$ISSUE_EXIT"
else
exit 99
fi
''',
"sleep": '#!/bin/sh\nprintf "sleep %s\\n" "$1" >> "$EXECUTED"\n',
"ssh": '#!/bin/sh\nprintf "cleanup\\n" >> "$EXECUTED"\n',
}
for name, contents in commands.items():
command = fake_bin / name
command.write_text(contents)
command.chmod(0o755)
dispatches = self.directory / "dispatches"
executed = self.directory / "executed"
body = self.directory / "issue-body.md"
body_path = self.directory / "issue-body-path"
self.env.update(
PATH=f"{fake_bin}{os.pathsep}{os.environ['PATH']}", DISPATCHES=str(dispatches),
EXECUTED=str(executed), CAPTURE_BODY=str(body), CAPTURE_BODY_PATH=str(body_path),
FAILED_ATTEMPTS="0", ISSUE_EXIT=str(issue_exit),
RUSTFS_NODES="fixture-node", RUSTFS_SSH_USER="fixture-user",
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))
self.steps = named_steps(job)
handoff = "Continue functional chain (next: Performance)"
self.assertIn(" if: ${{ always() && github.event_name == 'repository_dispatch' }}", self.steps[handoff])
self.assertFalse(any(line.strip().startswith("continue-on-error:") for line in self.steps[handoff]))
self.assertIn(" if: always()", self.steps["Cleanup environment (after)"])
self.assertLess(list(self.steps).index("Cleanup environment (after)"), list(self.steps).index(handoff))
initialized = self.run_step("Initialize functional evidence")
self.assertEqual(initialized.returncode, 0, initialized.stderr)
suite = self.directory / "auto-testing/rustfs-replication-test.sh"
suite.write_text('#!/bin/sh\nprintf "suite failed\\n" >> "$EXECUTED"\nexit 17\n')
failed = self.run_step("Run replication suite")
self.assertEqual(failed.returncode, 17, failed.stderr)
cleaned = self.run_step("Cleanup environment (after)")
self.assertEqual(cleaned.returncode, 0, cleaned.stderr)
self.assertEqual(executed.read_text().splitlines(), ["suite failed", "cleanup"])
self.env["FAILED_ATTEMPTS"] = str(failed_attempts)
self.context["secrets.PF_TESTING_GH_TOKEN"] = token
forwarded = self.run_step(handoff)
self.assertEqual(forwarded.returncode == 0, bool(token) and failed_attempts < 3, forwarded.stderr)
calls = dispatches.read_text().splitlines() if dispatches.exists() else []
self.assertEqual(calls, [
"api --method POST repos/rustfs/rustfs/dispatches -f event_type=rustfs-chain-performance -F client_payload[from_suite]=replication",
] * (min(failed_attempts + 1, 3) if token else 0))
if failed_attempts == 3:
self.assertIn("could not hand off from **replication** to **Performance**", body.read_text())
self.assertIn("rustfs-chain-performance", body.read_text())
self.assertEqual(executed.read_text().splitlines().count("issue"), 2 if issue_exit else 1)
self.assertFalse(Path(body_path.read_text().strip()).exists())
class FunctionalWorkflowTests(unittest.TestCase):
@@ -225,7 +402,7 @@ class FunctionalWorkflowTests(unittest.TestCase):
if suite in self.DIRECT_TESTS:
test = steps[self.DIRECT_TESTS[suite]]
self.assertNotRegex("\n".join(test), r'''(?m)^ ["']?continue-on-error["']?\s*:''')
self.assertIn(" if: always()", steps["Generate report"])
self.assertIn(" if: ${{ always() && steps.evidence.outcome == 'success' }}", steps["Generate report"])
cleanup = steps["Reset test environment (after)" if suite == "performance" else "Cleanup environment (after)"]
condition = next(line.strip() for line in cleanup if line.startswith(" if:"))
self.assertIn(condition, (
@@ -234,7 +411,7 @@ class FunctionalWorkflowTests(unittest.TestCase):
"if: ${{ always() && (inputs.cleanup_after != 'false' || github.event_name != 'workflow_dispatch') }}",
))
if suite != "performance":
handoff = steps["Chain complete"] if suite == "replication" else next(
handoff = next(
value for name, value in steps.items() if name.startswith("Continue functional chain")
)
self.assertIn(" if: ${{ always() && github.event_name == 'repository_dispatch' }}", handoff)
@@ -277,13 +454,387 @@ class FunctionalWorkflowTests(unittest.TestCase):
self.assertIn("partial suite diagnostics", failed.stdout)
cleanup = execute("Cleanup environment (after)")
self.assertEqual(cleanup.returncode, 0, cleanup.stderr)
handoff_name = "Chain complete" if suite == "replication" else next(
handoff_name = next(
name for name in steps if name.startswith("Continue functional chain")
)
handoff = execute(handoff_name)
self.assertEqual(handoff.returncode, 0, handoff.stderr)
markers = (root / "executed").read_text().splitlines()
self.assertEqual(markers, ["cleanup"] if suite == "replication" else ["cleanup", "dispatch"])
self.assertEqual(markers, ["cleanup", "dispatch"])
class FunctionalCaseReportTests(unittest.TestCase):
def report(self, text: str | None, matrix: bool = False) -> tuple[bool, str, str]:
with tempfile.TemporaryDirectory() as directory:
root = Path(directory)
log = root / "suite.log"
if text is not None:
log.write_text(text)
valid = generate_report(log, root / "cases.md", root / "matrix.md" if matrix else None)
return valid, (root / "cases.md").read_text(), (root / "matrix.md").read_text() if matrix else ""
def test_repeated_case_executions_preserve_failure_and_context(self):
# log() from rustfs/auto-testing@6120aa0a76de, rustfs-kms-test.sh:131.
log = subprocess.check_output(["bash", "-c", r'''
log() { printf '\033[1;36m[INFO]\033[0m %s\n' "$*"; }
log '== topology: single-single kms-backend: local =='
printf '\033[32m--- KMS-101 roundtrip ---\033[0m\n[FAIL] KMS-101\n'
log '== topology: single-multi kms-backend: vault-kv2 =='
printf '%s\n' '--- KMS-101 roundtrip ---' '[PASS] KMS-101'
printf '%s\n' '--- KMS-101 roundtrip ---' '[UNSUPPORTED] KMS-101'
'''], text=True)
valid, cases, _ = self.report(log)
self.assertFalse(valid)
self.assertEqual(cases.count("| KMS-101 |"), 3)
self.assertIn("- Total: 3\n- PASS: 1\n- FAIL: 1\n- UNSUPPORTED: 1\n- RUNNING: 0\n", cases)
self.assertIn("roundtrip (topology: single-single kms-backend: local) | FAIL |", cases)
self.assertIn("roundtrip (topology: single-multi kms-backend: vault-kv2) | PASS |", cases)
self.assertNotIn("\\n", cases)
def test_missing_empty_unfinished_and_orphan_results_are_not_success(self):
for text in (None, "", "setup failed\n", "--- KMS-101 roundtrip ---\n", "[PASS] KMS-101\n",
"--- KMS-101 first ---\n--- KMS-101 second ---\n[PASS] KMS-101\n",
"--- KMS-101 first ---\n[FAIL] KMS-101\n[PASS] KMS-101\n"):
with self.subTest(log=text):
valid, cases, _ = self.report(text)
self.assertFalse(valid)
self.assertIn("## Case Summary", cases)
valid, cases, _ = self.report("[INFO] == suite: bucket replication (REP-*) ==\n--- REP-101 unsupported ---\n[UNSUPPORTED] REP-101\n")
self.assertTrue(valid)
self.assertIn("suite: bucket replication", cases)
self.assertIn("- UNSUPPORTED: 1\n", cases)
def test_upgrade_matrix_is_preserved_and_required_for_complete_report(self):
case = "--- UPG-101 upgrade ---\n[PASS] UPG-101\n"
for suffix, expected in (("", False), ("[UPG-TOPO] single-single local v1 v2 PASS=1 FAIL=0\n", True),
("[UPG-TOPO] single-single local v1 v2 PASS=1 FAIL=1\n", False)):
with self.subTest(matrix=suffix):
valid, _, matrix = self.report(case + suffix, matrix=True)
self.assertEqual(valid, expected)
self.assertIn("| Topology | KMS Backend | From Version | To Version | Result |", matrix)
self.assertIn("| single-single | local | v1 | v2 |" if suffix else "NOT RUN", matrix)
def test_s3_case_identifiers_include_digits(self):
valid, cases, _ = self.report("--- S3C-101 CreateBucket ---\n[PASS] S3C-101\n")
self.assertTrue(valid)
self.assertIn("| S3C-101 | CreateBucket (context not recorded) | PASS |", cases)
class FunctionalEvidenceTests(WorkflowSteps, unittest.TestCase):
SUITES = (*FunctionalWorkflowTests.DIRECT_TESTS, "heal", "performance")
def prepare(self, suite: str) -> None:
self.temp = tempfile.TemporaryDirectory()
self.addCleanup(self.temp.cleanup)
self.directory = Path(self.temp.name)
self.source = (ROOT / f".github/workflows/rustfs-{suite}-test.yml").read_text()
self.steps = named_steps(yaml_block(self.source.splitlines(), FunctionalWorkflowTests.JOBS[suite], 2))
self.context = {expression: "" for expression in re.findall(r"\$\{\{\s*(.*?)\s*\}\}", self.source)}
self.context.update({
"github.server_url": "https://github.com", "github.repository": "rustfs/rustfs",
"github.run_id": "314159", "github.run_attempt": "2", "github.sha": "0123456789abcdef0123456789abcdef01234567",
"github.event_name": "repository_dispatch", "steps.test.outcome": "success",
"secrets.PF_TESTING_GH_TOKEN": "local-fixture", "env.PF_TESTING_GH_TOKEN": "local-fixture",
})
self.artifacts = self.directory / f"rustfs-{suite}-314159-2"
self.env = {
**os.environ, "GITHUB_ENV": str(self.directory / "github-env"), "RUNNER_TEMP": self.temp.name,
"GITHUB_STEP_SUMMARY": str(self.directory / "summary.md"), "RUSTFS_NODES": "fixture-node",
"RUSTFS_NIGHTLY_PACKAGE_URL": "https://example.invalid/package.deb", "CAPTURE_BODY": str(self.directory / "issue.md"),
}
for key in ("server_url", "repository", "run_id", "run_attempt", "sha", "event_name"):
self.env[f"GITHUB_{key.upper()}"] = self.context[f"github.{key}"]
(self.directory / "scripts").mkdir()
(self.directory / "scripts/functional_case_report.py").symlink_to(ROOT / "scripts/functional_case_report.py")
fake_bin = self.directory / "bin"
fake_bin.mkdir()
(fake_bin / "python3").symlink_to(sys.executable)
for command, body in (
("ssh", 'printf "fixture-version\\n"\n'),
("gh", 'if [ "$1 $2" = "issue create" ]; then\n'
' while [ "$#" -gt 0 ]; do\n'
' if [ "$1" = "--body-file" ]; then cat "$2" > "$CAPTURE_BODY"; fi\n'
' shift\n'
' done\n'
'elif [ "$1 $2" = "api --method" ]; then cat >/dev/null; fi\n'),
):
script = fake_bin / command
script.write_text("#!/bin/sh\n" + body)
script.chmod(0o755)
self.env["PATH"] = f"{fake_bin}{os.pathsep}{os.environ['PATH']}"
def test_evidence_wiring_and_failed_initialization_cannot_publish_stale_files(self):
for suite, suffix in ((suite, suffix) for suite in self.SUITES for suffix in ("", "-scratch")):
with self.subTest(suite=suite, collision=suffix or "artifact"):
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)"))
if suite in FunctionalWorkflowTests.DIRECT_TESTS:
self.assertLess(names.index("Checkout repository (for report parser)"), names.index("Checkout auto-testing scripts (with retry)"))
for name, lines in self.steps.items():
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):
self.assertIn(" path: |", lines)
self.assertIn(" if-no-files-found: error", lines)
existing = Path(str(self.artifacts) + suffix)
existing.mkdir()
for filename in ("report.md", "suite.log"):
(existing / filename).write_text("OLD RUN EVIDENCE")
self.env.update(REPORT_FILE=str(existing / "report.md"), LOG_FILE=str(existing / "suite.log"))
initialized = self.run_step("Initialize functional evidence")
self.assertNotEqual(initialized.returncode, 0)
self.assertFalse(Path(self.env["GITHUB_ENV"]).exists())
issue = self.run_step("File failure issue in rustfs/backlog")
self.assertEqual(issue.returncode, 0, issue.stderr)
body = Path(self.env["CAPTURE_BODY"]).read_text()
self.assertNotIn("OLD RUN EVIDENCE", body)
self.assertIn("no report or log file was produced", body)
self.assertEqual((existing / "report.md").read_text(), "OLD RUN EVIDENCE")
def test_reports_use_only_current_complete_suite_evidence(self):
for suite in self.SUITES[:-1]:
good = "--- KMS-101 roundtrip ---\n[PASS] KMS-101\n"
partial = "--- KMS-101 roundtrip ---\n[PASS] KMS-101\n--- KMS-102 unfinished ---\n"
if suite == "s3-compat":
good, partial = good.replace("KMS-", "S3C-"), partial.replace("KMS-", "S3C-")
if suite == "upgrade":
good += "[UPG-TOPO] single-single local v1 v2 PASS=1 FAIL=0\n"
if suite == "heal":
good = "".join(f"[HEAL-STEP] {step} fixture PASS\n" for step in range(1, 8))
partial = "[HEAL-STEP] 1 fixture PASS\n"
for outcome, log in (("success", good), ("failure", good), ("success", partial), ("success", ""),
("success", None), ("skipped", None), ("cancelled", good)):
with self.subTest(suite=suite, outcome=outcome, log=log):
self.prepare(suite)
stale = self.directory / "old-suite.log"
stale.write_text("OLD RUN EVIDENCE\n" + good)
self.env.update(LOG_FILE=str(stale), REPORT_FILE=str(stale))
self.assertEqual(self.run_step("Initialize functional evidence").returncode, 0)
self.assertEqual(self.env["LOG_FILE"], str(self.artifacts / "suite.log"))
self.assertEqual(self.env["TMPDIR"], str(self.artifacts) + "-scratch")
if log is not None:
Path(self.env["LOG_FILE"]).write_text(log)
self.context["steps.test.outcome"] = outcome
report = self.run_step("Generate report")
success = outcome == "success" and log == good
self.assertEqual(report.returncode == 0, success, report.stderr)
contents = Path(self.env["REPORT_FILE"]).read_text()
self.assertNotIn("OLD RUN EVIDENCE", contents)
self.assertEqual("| PASS |" in contents, success)
for value in ("actions/runs/314159", "Attempt: 2", "Workflow Commit: " + self.context["github.sha"],
f"Test Step Outcome: {'success' if success else 'failure'}", f"Suite Step Outcome: {outcome}"):
self.assertIn(value, contents)
self.assertEqual(Path(self.env["GITHUB_STEP_SUMMARY"]).read_text(), contents)
evidence = (self.artifacts / ("steps.md" if suite == "heal" else "cases.md")).read_text()
if log in (good, partial):
self.assertIn("| PASS |", evidence)
self.assertNotIn("OLD RUN EVIDENCE", evidence)
def test_actual_suite_commands_pass_the_current_log_and_scratch_paths(self):
for suite in self.SUITES:
with self.subTest(suite=suite):
self.prepare(suite)
self.assertEqual(self.run_step("Initialize functional evidence").returncode, 0)
if suite == "heal":
self.assertEqual(self.env["RUSTFS_WARP_LOG_FILE"], str(self.artifacts / "warp.log"))
scripts = self.directory / "auto-testing"
scripts.mkdir()
filename = f"rustfs_{suite}_test.sh" if suite in ("heal", "performance") else f"rustfs-{suite}-test.sh"
script = scripts / filename
script.write_text(
'#!/bin/bash\nset -euo pipefail\nlog=""\n'
'while [ "$#" -gt 0 ]; do\n'
' if [ "$1" = "--log-file" ]; then log="$2"; shift; fi\n'
' shift\n'
'done\n'
'[ "$log" = "$LOG_FILE" ] || exit 31\n'
'printf "CURRENT SUITE LOG\\n" > "$log"\n'
'scratch=$(mktemp -d "$TMPDIR/fixture.XXXXXX")\n'
'printf "CURRENT SCRATCH\\n" > "$scratch/trace.log"\n'
'if [ -n "${RUSTFS_RESULT_DIR:-}" ]; then\n'
' mkdir -p "$RUSTFS_RESULT_DIR"\n'
' printf "CURRENT RESULTS\\n" > "$RUSTFS_RESULT_DIR/summary.md"\n'
'fi\n'
)
script.chmod(0o755)
name = FunctionalWorkflowTests.DIRECT_TESTS.get(suite) or (
"Run benchmark (GET/PUT/MIXED)" if suite == "performance" else "Run heal test (write -> outage -> heal -> verify)"
)
result = self.run_step(name)
self.assertEqual(result.returncode, 0, result.stderr)
self.assertEqual((self.artifacts / "suite.log").read_text(), "CURRENT SUITE LOG\n")
self.assertEqual(len(list(Path(self.env["TMPDIR"]).glob("fixture.*/trace.log"))), 1)
self.assertEqual(list(self.artifacts.glob("fixture.*")), [])
if suite == "performance":
self.assertEqual((self.artifacts / "results/summary.md").read_text(), "CURRENT RESULTS\n")
def test_upload_allowlist_preserves_diagnostics_without_scratch(self):
extra = {
"kms": ["cases.md"], "storage": ["cases.md"], "s3-compat": ["cases.md"],
"upgrade": ["cases.md", "matrix.md"], "replication": ["cases.md"], "heal": ["steps.md", "warp.log"],
"performance": ["version.txt", "results/master.log", "results/summary.md", "results/summary.tsv",
"results/get_1KiB.txt", "results/put_1MiB.txt", "results/mixed_4MiB.txt"],
}
for suite in self.SUITES:
with self.subTest(suite=suite):
self.prepare(suite)
self.assertEqual(self.run_step("Initialize functional evidence").returncode, 0)
expected = {self.artifacts / name for name in ["report.md", "suite.log", *extra[suite]]}
for path in expected:
path.parent.mkdir(parents=True, exist_ok=True)
path.write_text("PARTIAL FAILURE DIAGNOSTIC")
for directory in (self.artifacts, Path(self.env["TMPDIR"]), self.artifacts / "results"):
directory.mkdir(exist_ok=True)
(directory / "init.json").write_text('{"root_token":"FAKE-SECRET-CANARY"}')
self.assertEqual(self.uploaded_files(), expected)
self.assertTrue(all("FAKE-SECRET-CANARY" not in path.read_text() for path in self.uploaded_files()))
def test_kms_failure_after_vault_init_keeps_only_failure_evidence(self):
self.prepare("kms")
self.assertEqual(self.run_step("Initialize functional evidence").returncode, 0)
scripts = self.directory / "auto-testing"
scripts.mkdir()
# Vault file writes and EXIT cleanup from auto-testing@06cd3c097350:23-24,57,479-487.
# The Docker boundary returns synthetic credentials; setup fails before vault_stop.
(scripts / "rustfs-kms-test.sh").write_text(r'''#!/bin/bash
set -Eeuo pipefail
TEST_TMP="$(mktemp -d "${TMPDIR:-/tmp}/rustfs-test.XXXXXX")"
trap 'rm -rf "${TEST_TMP}"' EXIT
VAULT_DATA_DIR="${RUSTFS_VAULT_DATA_DIR:-${TMPDIR:-/tmp}/rustfs-vault-data}"
VAULT_CONTAINER="rustfs-vault"
heal_run() { "$@"; }
while [ "$#" -gt 0 ]; do
if [ "$1" = "--log-file" ]; then LOG_FILE="$2"; shift; fi
shift
done
mkdir -p "${VAULT_DATA_DIR}"
tmp_init="${TEST_TMP}/vault-init.json"
tmp_err="${TEST_TMP}/vault-init.stderr"
heal_run docker exec -e "VAULT_ADDR=http://127.0.0.1:8200" "${VAULT_CONTAINER}" \
vault operator init -key-shares=1 -key-threshold=1 -format=json \
> "${tmp_init}" 2>"${tmp_err}"
cat "${tmp_init}" | heal_run tee "${VAULT_DATA_DIR}/init.json" >/dev/null
printf '%s\n' 'vault initialized; root token acquired' 'fixture setup failed after init' > "${LOG_FILE}"
exit 42
''')
docker = self.directory / "bin/docker"
docker.write_text("""#!/bin/sh
[ "$1" = exec ] || exit 99
printf '%s\\n' '{"root_token":"FAKE-ROOT-CANARY","unseal_keys_b64":["FAKE-UNSEAL-CANARY"]}'
""")
docker.chmod(0o755)
result = self.run_step("Run KMS suite")
self.assertEqual(result.returncode, 42, result.stderr)
vault_init = Path(self.env["TMPDIR"]) / "rustfs-vault-data/init.json"
self.assertEqual(json.loads(vault_init.read_text()), {"root_token": "FAKE-ROOT-CANARY", "unseal_keys_b64": ["FAKE-UNSEAL-CANARY"]})
self.assertNotIn(self.artifacts, vault_init.parents)
self.assertEqual(list(Path(self.env["TMPDIR"]).glob("rustfs-test.*")), [])
self.assertNotEqual(self.run_step("Generate report").returncode, 0)
self.assertEqual(self.uploaded_files(), {self.artifacts / name for name in ("suite.log", "cases.md", "report.md")})
self.assertIn("fixture setup failed after init", (self.artifacts / "suite.log").read_text())
for path in self.uploaded_files():
self.assertNotIn("FAKE-ROOT-CANARY", path.read_text())
self.assertNotIn("FAKE-UNSEAL-CANARY", path.read_text())
def test_heal_accumulates_actual_staged_steps_without_overwriting_failures(self):
self.prepare("heal")
self.assertEqual(self.run_step("Initialize functional evidence").returncode, 0)
script = self.directory / "auto-testing/rustfs_heal_test.sh"
script.parent.mkdir()
# Result printf and full-run condition from auto-testing@6120aa0a76de:143,1163-1168.
script.write_text(r'''#!/bin/bash
set -euo pipefail
SELECTED_STEPS=()
PREFLIGHT=0
while [ "$#" -gt 0 ]; do
case "$1" in
--steps) IFS=',' read -ra SELECTED_STEPS <<< "$2"; shift ;;
--log-file) LOG_FILE="$2"; shift ;;
--preflight) PREFLIGHT=1 ;;
esac
shift
done
if [ "$PREFLIGHT" -eq 1 ]; then
printf '\n' >> "$INVOKED_STEPS"
exit 0
fi
printf '%s\n' "${SELECTED_STEPS[*]}" >> "$INVOKED_STEPS"
emit_step_result() {
local n="$1" desc="$2" status="$3"
printf '[HEAL-STEP] %s %s %s\n' "${n}" "${desc}" "${status}"
}
{
for step in "${SELECTED_STEPS[@]}"; do
emit_step_result "$step" "fixture step $step" PASS
done
want_all=1
for s in 1 2 3 4 5 6 7; do
[[ " ${SELECTED_STEPS[*]} " == *" ${s} "* ]] || want_all=0
done
if [ "${want_all}" -eq 1 ]; then
printf '[HEAL-RESULT] PASS all steps passed\n'
fi
} >> "$LOG_FILE"
''')
script.chmod(0o755)
self.env["INVOKED_STEPS"] = str(self.directory / "invoked-steps")
for name in ("Install RustFS package & start cluster", "Preflight checks", "Run heal test (write -> outage -> heal -> verify)"):
result = self.run_step(name)
self.assertEqual(result.returncode, 0, result.stderr)
self.assertEqual(Path(self.env["INVOKED_STEPS"]).read_text().splitlines(), ["1 2", "", "3 4 5 6 7"])
log = Path(self.env["LOG_FILE"]).read_text()
self.assertNotIn("[HEAL-RESULT]", log)
self.assertEqual(log.count("[HEAL-STEP]"), 7)
report = self.run_step("Generate report")
self.assertEqual(report.returncode, 0, report.stderr)
failed_logs = ["\n".join(line for line in log.splitlines() if not line.startswith(f"[HEAL-STEP] {step} ")) + "\n"
for step in range(1, 8)]
failed_logs += [
log.replace("[HEAL-STEP] 3", "[HEAL-STEP] 3 original failure FAIL\n[HEAL-STEP] 3"),
log + "[HEAL-STEP] 3 later step failure FAIL\n",
log + "[HEAL-RESULT] FAIL earlier failure\n[HEAL-RESULT] PASS later success\n",
log.replace("[HEAL-STEP] 4 fixture step 4 PASS", "[HEAL-STEP] 4 fixture step 4 SKIP"),
]
for failed_log in failed_logs:
with self.subTest(log=failed_log):
Path(self.env["LOG_FILE"]).write_text(failed_log)
report = self.run_step("Generate report")
self.assertNotEqual(report.returncode, 0, report.stderr)
contents = Path(self.env["REPORT_FILE"]).read_text()
self.assertIn("Test Step Outcome: failure", contents)
self.assertNotIn("| PASS |", contents)
if "original failure" in failed_log:
self.assertIn("| 3 | original failure | FAIL |", (self.artifacts / "steps.md").read_text())
if "later step failure" in failed_log:
self.assertIn("| 3 | later step failure | FAIL |", (self.artifacts / "steps.md").read_text())
def test_performance_results_version_and_report_are_bound_to_the_run(self):
self.prepare("performance")
initialized = self.run_step("Initialize functional evidence")
self.assertEqual(initialized.returncode, 0, initialized.stderr)
self.assertEqual(self.env["RUSTFS_RESULT_DIR"], str(self.artifacts / "results"))
self.assertEqual(self.env["VERSION_FILE"], str(self.artifacts / "version.txt"))
version = self.run_step("Collect RustFS version info")
self.assertEqual(version.returncode, 0, version.stderr)
self.assertIn("fixture-version", Path(self.env["VERSION_FILE"]).read_text())
old_summary = self.directory / "old-results/summary.md"
old_summary.parent.mkdir()
old_summary.write_text("OLD RUN EVIDENCE")
upload = "Upload report to dashboard (reports/YYYY-MM-DD.md)"
self.assertNotEqual(self.run_step(upload).returncode, 0)
self.assertFalse(Path(self.env["REPORT_FILE"]).exists())
results = Path(self.env["RUSTFS_RESULT_DIR"])
results.mkdir()
(results / "summary.md").write_text("CURRENT PERFORMANCE RESULTS\n")
report = self.run_step(upload)
self.assertEqual(report.returncode, 0, report.stderr)
contents = Path(self.env["REPORT_FILE"]).read_text()
for value in ("actions/runs/314159", "**Attempt**: 2", "**Workflow Commit**: " + self.context["github.sha"],
"CURRENT PERFORMANCE RESULTS", "fixture-version"):
self.assertIn(value, contents)
self.assertNotIn("OLD RUN EVIDENCE", contents)
if __name__ == "__main__":