mirror of
https://github.com/rustfs/rustfs.git
synced 2026-10-13 09:17:41 +08:00
100 lines
4.6 KiB
Python
100 lines
4.6 KiB
Python
#!/usr/bin/env python3
|
|
"""Exercise the consolidated upgrade lane without building RustFS."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import re
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
import unittest
|
|
|
|
from check_test_wiring import yaml_block
|
|
from test_security_workflow import named_steps, shell_body
|
|
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
|
|
|
|
class UpgradeWorkflowTests(unittest.TestCase):
|
|
def setUp(self):
|
|
source = (ROOT / ".github/workflows/e2e-upgrade.yml").read_text().splitlines()
|
|
self.job = yaml_block(source, "upgrade", 2)
|
|
self.steps = named_steps(self.job)
|
|
self.step = self.steps["Run upgrade compatibility tests"]
|
|
# Explicit bash enables pipefail in Actions, so tee cannot hide a failure.
|
|
self.assertIn(" shell: bash", self.step)
|
|
self.assertNotIn(" strategy:", self.job)
|
|
rust = (ROOT / "crates/e2e_test/src/upgrade_compatibility_test.rs").read_text()
|
|
self.expected = set(re.findall(r'#\[ignore[^\n]*\]\s*async fn (\w+)\(', rust))
|
|
self.assertTrue(self.expected)
|
|
|
|
def run_cases(self, failed_test="", fail_tee=False):
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
root = Path(directory)
|
|
commands = root / "commands.jsonl"
|
|
fake_python = root / "python3"
|
|
fake_python.write_text(f"#!{sys.executable}\n" + '''import json, os, sys
|
|
from pathlib import Path
|
|
args = sys.argv[1:]
|
|
assert args[:8] == ["scripts/e2e_binary.py", "run", "--", "cargo", "test", "--locked", "-p", "e2e_test"]
|
|
assert args[9:] == ["--", "--ignored", "--exact", "--nocapture"]
|
|
test = args[8].removeprefix("upgrade_compatibility_test::")
|
|
assert args[8] != test
|
|
assert os.environ["no_proxy"] == "127.0.0.1,localhost,::1"
|
|
with open(os.environ["COMMANDS"], "a") as output:
|
|
output.write(json.dumps({"test": test, "logs": os.environ["RUSTFS_E2E_LOG_DIR"],
|
|
"evidence": os.environ["RUSTFS_SCANNER_HEAL_G09_EVIDENCE_DIR"]}) + "\\n")
|
|
Path(os.environ["RUSTFS_SCANNER_HEAL_G09_EVIDENCE_DIR"]).mkdir(parents=True)
|
|
print("executed " + test)
|
|
sys.exit(17 if test == os.environ["FAILED_TEST"] else 0)
|
|
''')
|
|
fake_python.chmod(0o755)
|
|
if fail_tee:
|
|
tee = root / "tee"
|
|
tee.write_text("#!/bin/sh\ncat >/dev/null\nexit 23\n")
|
|
tee.chmod(0o755)
|
|
env = dict(os.environ, PATH=f"{root}{os.pathsep}{os.environ['PATH']}",
|
|
RUNNER_TEMP=str(root), COMMANDS=str(commands), FAILED_TEST=failed_test,
|
|
NO_PROXY="127.0.0.1,localhost,::1")
|
|
result = subprocess.run(["bash", "--noprofile", "--norc", "-e", "-o", "pipefail", "-c", shell_body(self.step)],
|
|
cwd=root, env=env, capture_output=True, text=True)
|
|
calls = [json.loads(line) for line in commands.read_text().splitlines()]
|
|
self.assertEqual({call["test"] for call in calls}, self.expected)
|
|
self.assertEqual(len(calls), len(self.expected))
|
|
for field in ("logs", "evidence"):
|
|
self.assertEqual(len({call[field] for call in calls}), len(calls))
|
|
self.assertTrue(all(Path(call[field]).is_dir() for call in calls))
|
|
if not fail_tee:
|
|
for call in calls:
|
|
self.assertIn(call["test"], (Path(call["logs"]) / "test.log").read_text())
|
|
return result
|
|
|
|
def test_all_ignored_upgrade_cases_run_once_with_separate_evidence(self):
|
|
result = self.run_cases()
|
|
self.assertEqual(result.returncode, 0, result.stderr)
|
|
|
|
def test_early_or_late_failure_does_not_hide_remaining_cases(self):
|
|
for name in ("direct_upgrade_from_rc2_preserves_object_contracts", "rc5_baseline_replicates_multipart_layouts"):
|
|
with self.subTest(name=name):
|
|
result = self.run_cases(failed_test=name)
|
|
self.assertNotEqual(result.returncode, 0, result.stdout)
|
|
self.assertIn(f"::error::Upgrade compatibility failed: {name}", result.stdout)
|
|
|
|
def test_log_write_failure_fails_the_job(self):
|
|
self.assertNotEqual(self.run_cases(fail_tee=True).returncode, 0)
|
|
|
|
def test_reports_are_uploaded_even_after_failure(self):
|
|
for name, directory in (("Upload scanner/heal G09 evidence", "rustfs-upgrade-g09-evidence"),
|
|
("Upload server logs", "rustfs-upgrade-logs")):
|
|
step = self.steps[name]
|
|
self.assertIn(" if: always()", step)
|
|
self.assertIn(" path: ${{ runner.temp }}/" + directory, step)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|