mirror of
https://github.com/rcourtman/Pulse.git
synced 2026-10-26 13:39:47 +00:00
pulse-triage[bot]
54f40a6a76
The docs-only job kept twelve literal suites while thirty existing privacy, authentication, recovery and other guide suites were omitted. Discover the complete documentation inventory in four isolated bounded shards, trigger on every helper change, and preserve the required check as a fail-closed aggregate. Keep all-suite and explicit smoke execution, safety recipes and deadlines unchanged. Contract-Neutral: Documentation CI admission and smoke-harness coverage only; no product runtime, installer behaviour, shipped guidance, release checks or subsystem contract changes. Change-source: pulse-maintainer
196 lines
9.6 KiB
Python
196 lines
9.6 KiB
Python
#!/usr/bin/env python3
|
|
"""Docs-only CI must execute every existing safety suite and fail closed.
|
|
|
|
Exercise the actual workflow commands with disposable test adapters. Real
|
|
recipes run separately in the source-proof VM, never against customer services.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import fnmatch
|
|
import os
|
|
from pathlib import Path
|
|
import re
|
|
import shlex
|
|
import shutil
|
|
import subprocess
|
|
import tempfile
|
|
import textwrap
|
|
import unittest
|
|
|
|
|
|
ROOT = Path(__file__).resolve().parents[2]
|
|
WORKFLOW = ROOT / ".github/workflows/public-docs.yml"
|
|
SAFETY_STEP = "Exercise documentation safety and diagnostic recipes"
|
|
VERDICT_STEP = "Require every documentation recipe shard"
|
|
# Independent witnesses: all twelve formerly covered suites, plus the critical
|
|
# privacy/access/recovery checks that the former literal workflow omitted.
|
|
REQUIRED_SUITES = (
|
|
"test_production_rollout_docs.py", "test_pmg_docs.py",
|
|
"test_agent_profile_docs.py", "test_pbs_docs.py", "test_audit_recovery_docs.py",
|
|
"test_notification_troubleshooting_docs.py", "test_webhook_verification_docs.py",
|
|
"test_pve_backup_troubleshooting_docs.py", "test_vm_disk_diagnostics.py",
|
|
"test_memory_troubleshooting_docs.py", "test_performance_troubleshooting_docs.py",
|
|
"test_troubleshooting_logs.py", "test_admin_docs.py", "test_api_auth_docs.py",
|
|
"test_authentication_diagnostic_docs.py", "test_update_recovery_docs.py",
|
|
"test_guest_disk_reading_docs.py", "test_guest_network_troubleshooting_docs.py",
|
|
"test_truenas_docs.py", "test_repo_docs_link_drift.py", "test_security_docs.py",
|
|
)
|
|
LEGACY_SUITES = {
|
|
"test_vm_disk_diagnostics.py", "test_troubleshooting_logs.py",
|
|
"test-retired-trial-acquisition-docs.sh",
|
|
}
|
|
|
|
|
|
def documentation_suites() -> list[str]:
|
|
return sorted(path.name for path in (ROOT / "scripts/tests").iterdir()
|
|
if path.is_file() and (fnmatch.fnmatch(path.name, "test_*docs*.py")
|
|
or path.name in LEGACY_SUITES))
|
|
|
|
|
|
def job_body(source: str, name: str) -> str:
|
|
marker = f"\n {name}:\n"
|
|
if source.count(marker) != 1:
|
|
raise AssertionError(f"expected exactly one {name} job")
|
|
return re.split(r"\n [\w-]+:\n", source.split(marker, 1)[1], maxsplit=1)[0]
|
|
|
|
|
|
def step_body(source: str, name: str) -> str:
|
|
marker = f" - name: {name}\n"
|
|
if source.count(marker) != 1:
|
|
raise AssertionError(f"expected exactly one {name} step")
|
|
return source.split(marker, 1)[1].split("\n - name:", 1)[0]
|
|
|
|
|
|
def safety_command(source: str, shard: int) -> str:
|
|
body = step_body(job_body(source, "recipes"), SAFETY_STEP)
|
|
run = re.findall(r"^ run: (.+)$", body, re.MULTILINE)
|
|
if len(run) != 1:
|
|
raise AssertionError("expected one literal docs-shard command")
|
|
return run[0].replace("${{ matrix.shard }}", str(shard))
|
|
|
|
|
|
def verdict_command(source: str) -> str:
|
|
body = step_body(job_body(source, "check"), VERDICT_STEP)
|
|
marker = " run: |\n"
|
|
if body.count(marker) != 1:
|
|
raise AssertionError("expected one literal verdict command")
|
|
return textwrap.dedent(body.split(marker, 1)[1]).strip()
|
|
|
|
|
|
class PublicDocsWorkflowTest(unittest.TestCase):
|
|
def setUp(self):
|
|
self.source = WORKFLOW.read_text(encoding="utf-8")
|
|
|
|
def test_both_events_cover_all_suite_and_helper_changes(self):
|
|
triggers = self.source.split("\non:\n", 1)[1].split("\npermissions:", 1)[0]
|
|
required = {
|
|
"*.md", "docs/**", "frontend-modern/public/docs/**",
|
|
".github/ISSUE_TEMPLATE/**", ".github/PULL_REQUEST_TEMPLATE.md",
|
|
"scripts/check_public_docs.py", "scripts/check_docs_mirror.py",
|
|
"scripts/readme_entrypoints_test.py", "scripts/tests/**",
|
|
"scripts/test-vm-disk.sh", ".github/workflows/public-docs.yml",
|
|
}
|
|
for event in ("pull_request", "push"):
|
|
with self.subTest(event=event):
|
|
block = triggers.split(f" {event}:\n", 1)[1]
|
|
block = re.split(r"\n [a-z_]+:\n", block, maxsplit=1)[0]
|
|
paths = re.findall(r'^ - "([^"]+)"$', block, re.MULTILINE)
|
|
self.assertTrue(required.issubset(paths), required.difference(paths))
|
|
self.assertEqual(len(paths), len(set(paths)))
|
|
self.assertIn(" push:\n branches:\n - main\n", triggers)
|
|
|
|
def test_shards_discover_current_suites_without_dependency_setup(self):
|
|
self.assertTrue(set(REQUIRED_SUITES).issubset(documentation_suites()))
|
|
for shard in range(4):
|
|
self.assertEqual(shlex.split(safety_command(self.source, shard)),
|
|
["scripts/tests/run.sh", "--docs-shard", str(shard)])
|
|
self.assertNotIn("actions/setup-", self.source)
|
|
self.assertNotRegex(self.source, r"\b(?:npm|pip|go) (?:ci|install|build|test)\b")
|
|
|
|
def test_existing_document_admissions_are_preserved(self):
|
|
check = job_body(self.source, "check")
|
|
for name, command in (
|
|
("Validate public documentation", "python3 scripts/check_public_docs.py"),
|
|
("Check getting-started safety and current plans", "python3 scripts/readme_entrypoints_test.py"),
|
|
("Check shipped docs mirror sync", "python3 scripts/check_docs_mirror.py"),
|
|
("Check documentation safety workflow contract", "python3 scripts/tests/test_public_docs_workflow.py"),
|
|
):
|
|
with self.subTest(step=name):
|
|
self.assertIn(f" run: {command}\n", step_body(check, name))
|
|
|
|
def test_all_four_isolated_shards_are_mandatory_and_bounded(self):
|
|
recipes = job_body(self.source, "recipes")
|
|
self.assertIn(" shard: [0, 1, 2, 3]\n", recipes)
|
|
self.assertIn(" fail-fast: false\n", recipes)
|
|
self.assertNotRegex(recipes, r"(?m)^\s*(?:if|continue-on-error|exclude):")
|
|
self.assertNotIn("continue-on-error:", self.source)
|
|
for name in ("recipes", "check"):
|
|
job = job_body(self.source, name)
|
|
self.assertIn(" runs-on: ubuntu-24.04\n", job)
|
|
self.assertIn(" timeout-minutes: 10\n", job)
|
|
self.assertIn(" persist-credentials: false\n", job)
|
|
self.assertIn("permissions:\n contents: read\n", self.source)
|
|
|
|
def test_required_check_runs_even_when_shards_fail_or_are_skipped(self):
|
|
check = job_body(self.source, "check")
|
|
self.assertIn(" needs: recipes\n", check)
|
|
self.assertIn(" if: ${{ always() }}\n", check)
|
|
self.assertNotRegex(check.replace(" if: ${{ always() }}\n", ""), r"(?m)^\s*if:")
|
|
self.assertIn(" RECIPE_RESULT: ${{ needs.recipes.result }}\n", check)
|
|
self.assertLess(check.index(VERDICT_STEP), check.index("Validate public documentation"))
|
|
for state in ("success", "failure", "cancelled", "skipped", "", "unknown"):
|
|
with self.subTest(state=state):
|
|
result = subprocess.run(["bash", "-e", "-o", "pipefail", "-c", verdict_command(self.source)],
|
|
env=dict(os.environ, RECIPE_RESULT=state), text=True,
|
|
capture_output=True, timeout=5)
|
|
self.assertEqual(result.returncode == 0, state == "success", result.stdout + result.stderr)
|
|
|
|
def run_adapters(self, shard: int, failing: str = ""):
|
|
with tempfile.TemporaryDirectory() as temporary:
|
|
root = Path(temporary)
|
|
tests = root / "scripts/tests"
|
|
tests.mkdir(parents=True)
|
|
shutil.copyfile(ROOT / "scripts/tests/run.sh", tests / "run.sh")
|
|
(tests / "run.sh").chmod(0o700)
|
|
for name in documentation_suites():
|
|
status = 1 if name == failing else 0
|
|
if name.endswith(".sh"):
|
|
code = f'#!/bin/bash\nprintf "%s\\n" {shlex.quote(name)} >> "$CALLS"\nexit {status}\n'
|
|
else:
|
|
code = ("import os\nfrom pathlib import Path\n"
|
|
f"with Path(os.environ['CALLS']).open('a') as log: log.write({name!r} + '\\n')\n"
|
|
f"raise SystemExit({status})\n")
|
|
(tests / name).write_text(code, encoding="utf-8")
|
|
(tests / name).chmod(0o700)
|
|
calls = root / "calls"
|
|
result = subprocess.run(["bash", "-e", "-o", "pipefail", "-c", safety_command(self.source, shard)],
|
|
cwd=root, env=dict(os.environ, CALLS=str(calls)),
|
|
text=True, capture_output=True, timeout=15)
|
|
return result, calls.read_text().splitlines() if calls.exists() else []
|
|
|
|
def test_actual_workflow_commands_cover_every_suite_once(self):
|
|
seen = []
|
|
for shard in range(4):
|
|
result, calls = self.run_adapters(shard)
|
|
self.assertEqual(result.returncode, 0, result.stdout + result.stderr)
|
|
self.assertEqual(calls, documentation_suites()[shard::4])
|
|
self.assertIn(f"Summary: {len(calls)}/{len(calls)} passed", result.stdout)
|
|
seen.extend(calls)
|
|
self.assertEqual(sorted(seen), documentation_suites())
|
|
self.assertEqual(len(seen), len(set(seen)))
|
|
|
|
def test_each_failed_suite_fails_its_actual_workflow_command(self):
|
|
for index, failing in enumerate(documentation_suites()):
|
|
with self.subTest(failing=failing):
|
|
shard = index % 4
|
|
result, calls = self.run_adapters(shard, failing)
|
|
self.assertNotEqual(result.returncode, 0, result.stdout + result.stderr)
|
|
self.assertEqual(calls, documentation_suites()[shard::4])
|
|
self.assertIn(f"Summary: {len(calls)-1}/{len(calls)} passed", result.stdout)
|
|
self.assertIn("Failures: 1", result.stdout)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|