mirror of
https://github.com/rcourtman/Pulse.git
synced 2026-10-26 13:39:47 +00:00
rcourtman
64c0a3f9df
An unrelated Microsoft repository returned 403 during fresh-index admission and stopped a browser shard before any test could run. Load the hosted runner's existing Ubuntu Deb822 definition for browser packages, preserving its mirrors and signature policy. Fail when that definition is missing, unreadable or empty and keep strict index and archive hash checks. Exercise the real apt parser, exclusion of unrelated feeds, missing definitions and incomplete index failures in private acquisition directories. Register the package helper and policy with their exact verification routes, and update the fixture helper's governance assertion to its scoped proof contract.
293 lines
15 KiB
Python
293 lines
15 KiB
Python
#!/usr/bin/env python3
|
|
"""Keep failed browser setup out of the E2E test budget, without stale apt success."""
|
|
|
|
import os
|
|
from pathlib import Path
|
|
import re
|
|
import socket
|
|
import subprocess
|
|
import tempfile
|
|
import unittest
|
|
|
|
|
|
ROOT = Path(__file__).resolve().parents[2]
|
|
WORKFLOW = ROOT / ".github/workflows/test-e2e.yml"
|
|
POLICY = ROOT / "scripts/playwright-apt.conf"
|
|
INSTALL_POLICY = (
|
|
"sudo install -m 0644 scripts/playwright-apt.conf "
|
|
"/etc/apt/apt.conf.d/99pulse-playwright-acquire"
|
|
)
|
|
# actions/cache v6.1.0 (Node 24). Restore and save are separate steps so a
|
|
# successful install is cached even when later journeys fail.
|
|
CACHE_PIN = "55cc8345863c7cc4c66a329aec7e433d2d1c52a9"
|
|
CACHE_DIR = '"$HOME/.cache/pulse-browser-os-packages"'
|
|
UBUNTU_SOURCES = "/etc/apt/sources.list.d/ubuntu.sources"
|
|
|
|
|
|
def job_block(workflow, job):
|
|
return re.split(r"(?m)^ \S", workflow.split(f"\n {job}:\n", 1)[1], maxsplit=1)[0]
|
|
|
|
|
|
def step_blocks(job):
|
|
return re.split(r"(?m)^ - name: ", job)[1:]
|
|
|
|
|
|
class PlaywrightAcquisitionTest(unittest.TestCase):
|
|
def assert_workflow_policy(self, workflow):
|
|
for job in ("offline-org-provisioning", "e2e", "agent-registration"):
|
|
with self.subTest(job=job):
|
|
block = job_block(workflow, job)
|
|
steps = step_blocks(block)
|
|
policies = [i for i, step in enumerate(steps) if INSTALL_POLICY in step]
|
|
installers = [
|
|
(i, step) for i, step in enumerate(steps)
|
|
if "npx playwright install --with-deps " in step
|
|
]
|
|
self.assertEqual(len(policies), 1)
|
|
self.assertEqual(len(installers), 1)
|
|
i, install = installers[0]
|
|
self.assertLess(policies[0], i)
|
|
for step in (steps[policies[0]], install):
|
|
self.assertNotIn("continue-on-error", step)
|
|
self.assertNotIn("if:", step)
|
|
self.assertNotIn("||", step)
|
|
# WebKit's complete OS graph was still downloading when the
|
|
# former ten-minute cap killed a job before any test ran.
|
|
# Only that larger setup gets more aggregate acquisition time.
|
|
budget = 30 if job == "e2e" else 10
|
|
self.assertIn(f"timeout-minutes: {budget}\n", install)
|
|
self.assertIn("working-directory: tests/integration", install)
|
|
browsers = "chromium webkit" if job == "e2e" else "chromium"
|
|
self.assertIn(f"npx playwright install --with-deps {browsers}\n", install)
|
|
journey = next(
|
|
n for n, step in enumerate(steps)
|
|
if step.startswith("Run stable-tier E2E suite")
|
|
or step.startswith("Prove authenticated default")
|
|
or step.startswith("Run agent registration lifecycle")
|
|
)
|
|
# Browser setup must finish successfully before real journeys.
|
|
self.assertLess(i, journey)
|
|
self.assert_package_cache(steps, policies[0], i, journey, browsers)
|
|
if job == "e2e":
|
|
self.assertRegex(block, r"(?m)^ timeout-minutes: 60$")
|
|
self.assertNotRegex(install, r"(?m)^ timeout-minutes: 10$")
|
|
|
|
def assert_package_cache(self, steps, policy, install, journey, browsers):
|
|
"""Slow mirrors must not gate setup, and cached files must not bypass apt's authentication."""
|
|
def only(name):
|
|
found = [n for n, step in enumerate(steps) if step.startswith(name + "\n")]
|
|
self.assertEqual(len(found), 1, name)
|
|
return found[0]
|
|
|
|
restore = only("Restore browser OS packages")
|
|
admit = only("Admit verified browser OS packages")
|
|
collect = only("Collect browser OS packages")
|
|
save = only("Save browser OS packages")
|
|
# The bounded apt policy applies to the index refresh inside admission;
|
|
# packages are saved only after a successful install and before any
|
|
# journey can fail the job.
|
|
self.assertLess(policy, admit)
|
|
self.assertLess(restore, admit)
|
|
self.assertLess(admit, install)
|
|
self.assertLess(install, collect)
|
|
self.assertLess(collect, save)
|
|
self.assertLess(save, journey)
|
|
slug = browsers.replace(" ", "-")
|
|
prefix = "browser-os-packages-${{ runner.os }}-" + slug + "-"
|
|
lock = "${{ hashFiles('tests/integration/package-lock.json') }}"
|
|
self.assertIn(f"uses: actions/cache/restore@{CACHE_PIN} # v6.1.0\n", steps[restore])
|
|
# A run-unique key keeps saving possible as Ubuntu publishes new
|
|
# versions; restores take the newest entry by prefix.
|
|
self.assertIn(f"key: {prefix}{lock}-${{{{ github.run_id }}}}-${{{{ github.run_attempt }}}}\n", steps[restore])
|
|
self.assertIn(f" {prefix}{lock}-\n {prefix}\n", steps[restore])
|
|
# Admission and collection go through the verifying helper; a direct
|
|
# copy would let apt accept a same-size file without a hash check.
|
|
self.assertIn(f"run: python3 scripts/browser_os_packages.py seed {CACHE_DIR}\n", steps[admit])
|
|
self.assertIn("id: browser-os-collect\n", steps[collect])
|
|
self.assertIn(f"run: python3 scripts/browser_os_packages.py collect {CACHE_DIR}\n", steps[collect])
|
|
self.assertIn(f"uses: actions/cache/save@{CACHE_PIN} # v6.1.0\n", steps[save])
|
|
self.assertIn("if: steps.browser-os-collect.outputs.changed == 'true'\n", steps[save])
|
|
self.assertIn("key: ${{ steps.browser-os-packages.outputs.cache-primary-key }}\n", steps[save])
|
|
for n in (admit, collect):
|
|
for forbidden in ("${{", "continue-on-error", "if:", "||", " cp ", "install -m"):
|
|
self.assertNotIn(forbidden, steps[n])
|
|
|
|
def test_all_browser_jobs_bound_setup_without_weakening_gates(self):
|
|
self.assert_workflow_policy(WORKFLOW.read_text())
|
|
|
|
def test_policy_and_tests_trigger_both_pr_and_main_validation(self):
|
|
workflow = WORKFLOW.read_text()
|
|
for event in ("pull_request", "push"):
|
|
block = re.split(r"(?m)^ \S", workflow.split(f" {event}:\n", 1)[1], maxsplit=1)[0]
|
|
self.assertIn("'scripts/playwright-apt.conf'", block)
|
|
self.assertIn("'scripts/tests/test_playwright_acquisition.py'", block)
|
|
self.assertIn("'scripts/browser_os_packages.py'", block)
|
|
self.assertIn("'scripts/tests/test_browser_os_packages.py'", block)
|
|
selection = job_block(workflow, "tier-selection")
|
|
for test in ("test_playwright_acquisition.py", "test_browser_os_packages.py"):
|
|
self.assertLess(
|
|
selection.index(f"python3 scripts/tests/{test}"),
|
|
selection.index("npm ci"),
|
|
)
|
|
|
|
def test_unbounded_and_ignored_setup_mutants_are_rejected(self):
|
|
workflow = WORKFLOW.read_text()
|
|
# Each mutation targets the fault, not an unrelated YAML assertion.
|
|
for original, replacement in (
|
|
(" timeout-minutes: 10\n", ""),
|
|
(" timeout-minutes: 30\n", " timeout-minutes: 10\n"),
|
|
(" timeout-minutes: 60\n", " timeout-minutes: 90\n"),
|
|
(INSTALL_POLICY, "echo no acquisition policy"),
|
|
("run: npx playwright install --with-deps chromium\n",
|
|
"run: npx playwright install --with-deps chromium || true\n"),
|
|
(f"actions/cache/restore@{CACHE_PIN}", "actions/cache/restore@v6"),
|
|
("-${{ github.run_id }}-${{ github.run_attempt }}\n", "\n"),
|
|
(f"run: python3 scripts/browser_os_packages.py seed {CACHE_DIR}\n",
|
|
f"run: sudo cp {CACHE_DIR}/*.deb /var/cache/apt/archives/\n"),
|
|
("if: steps.browser-os-collect.outputs.changed == 'true'\n", "if: always()\n"),
|
|
):
|
|
with self.subTest(mutation=replacement):
|
|
# subTest failures cannot be caught as an ordinary assertion,
|
|
# so collect the mutant's real unittest result independently.
|
|
class Mutant(PlaywrightAcquisitionTest):
|
|
def runTest(inner):
|
|
inner.assert_workflow_policy(workflow.replace(original, replacement))
|
|
|
|
result = unittest.TestResult()
|
|
Mutant().run(result)
|
|
self.assertFalse(result.wasSuccessful())
|
|
self.assertTrue(result.failures)
|
|
self.assertFalse(result.errors)
|
|
|
|
def test_real_apt_parser_loads_only_the_expected_controls(self):
|
|
with tempfile.TemporaryDirectory() as temp:
|
|
config = Path(temp) / "apt.conf"
|
|
config.write_text(
|
|
'Dir::Etc::parts "-";\nDir::Etc::main "-";\n' + POLICY.read_text()
|
|
)
|
|
parsed = subprocess.run(
|
|
["apt-config", "dump"], env={**os.environ, "APT_CONFIG": str(config)},
|
|
capture_output=True, text=True, timeout=10, check=True,
|
|
).stdout
|
|
expected = {
|
|
"Dir::Etc::sourcelist": UBUNTU_SOURCES,
|
|
"Dir::Etc::sourceparts": "-",
|
|
"APT::Update::Pre-Invoke::": f"test -r {UBUNTU_SOURCES} && test -s {UBUNTU_SOURCES}",
|
|
"Acquire::http::Timeout": "30",
|
|
"Acquire::https::Timeout": "30",
|
|
"Acquire::Retries": "2",
|
|
"APT::Update::Error-Mode": "any",
|
|
}
|
|
entries = dict(re.findall(r'^(\S+) "([^"\n]*)";', POLICY.read_text(), re.M))
|
|
self.assertEqual(entries, expected)
|
|
for key, value in expected.items():
|
|
self.assertIn(f'{key} "{value}";', parsed)
|
|
self.assertNotIn("AllowUnauthenticated", parsed)
|
|
self.assertNotIn("Verify-Peer", parsed)
|
|
|
|
def test_real_apt_preserves_ubuntu_sources_and_excludes_unrelated_feeds(self):
|
|
# Print acquisition URIs without contacting a repository or touching
|
|
# host state. Both configurations retain the same signed Ubuntu source;
|
|
# the adverse control also admits an unrelated, unreachable vendor feed.
|
|
with tempfile.TemporaryDirectory() as temp:
|
|
root = Path(temp)
|
|
parts = root / "sources.list.d"
|
|
parts.mkdir()
|
|
ubuntu = parts / "ubuntu.sources"
|
|
definition = (
|
|
"Types: deb\nURIs: http://127.0.0.1:1/ubuntu/\n"
|
|
"Suites: noble\nComponents: main\nArchitectures: amd64\n"
|
|
"Signed-By: /usr/share/keyrings/ubuntu-archive-keyring.gpg\n"
|
|
)
|
|
ubuntu.write_text(definition)
|
|
(parts / "unrelated-vendor.list").write_text(
|
|
"deb http://127.0.0.1:1/vendor/ stable main\n"
|
|
)
|
|
(root / "lists/partial").mkdir(parents=True)
|
|
(root / "cache/archives/partial").mkdir(parents=True)
|
|
prefix = (
|
|
'Dir::Etc::parts "-";\nDir::Etc::main "-";\n'
|
|
f'Dir::Etc::sourcelist "-";\nDir::Etc::sourceparts "{parts}";\n'
|
|
f'Dir::State::lists "{root / "lists"}";\n'
|
|
f'Dir::Cache "{root / "cache"}";\n'
|
|
'Dir::State::status "/dev/null";\nDebug::NoLocking "true";\n'
|
|
)
|
|
policy = POLICY.read_text().replace(UBUNTU_SOURCES, str(ubuntu))
|
|
config = root / "apt.conf"
|
|
for scoped in (True, False):
|
|
selected = policy
|
|
if not scoped:
|
|
selected = re.sub(r'^Dir::Etc::source(?:list|parts) .*;\n', '', selected, flags=re.M)
|
|
config.write_text(prefix + selected)
|
|
result = subprocess.run(
|
|
["apt-get", "update", "--print-uris"],
|
|
env={**os.environ, "APT_CONFIG": str(config), "LC_ALL": "C"},
|
|
capture_output=True, text=True, timeout=10, check=True,
|
|
)
|
|
with self.subTest(scoped=scoped):
|
|
self.assertIn("http://127.0.0.1:1/ubuntu/", result.stdout)
|
|
self.assertEqual("http://127.0.0.1:1/vendor/" in result.stdout, not scoped)
|
|
self.assertEqual(ubuntu.read_text(), definition)
|
|
|
|
def test_missing_ubuntu_definition_fails_before_acquisition(self):
|
|
with tempfile.TemporaryDirectory() as temp:
|
|
root = Path(temp)
|
|
(root / "lists/partial").mkdir(parents=True)
|
|
(root / "cache/archives/partial").mkdir(parents=True)
|
|
missing = root / "missing.sources"
|
|
config = root / "apt.conf"
|
|
config.write_text(
|
|
'Dir::Etc::parts "-";\nDir::Etc::main "-";\n'
|
|
f'Dir::State::lists "{root / "lists"}";\n'
|
|
f'Dir::Cache "{root / "cache"}";\n'
|
|
'Dir::State::status "/dev/null";\nDebug::NoLocking "true";\n'
|
|
+ POLICY.read_text().replace(UBUNTU_SOURCES, str(missing))
|
|
)
|
|
result = subprocess.run(
|
|
["apt-get", "update"],
|
|
env={**os.environ, "APT_CONFIG": str(config), "LC_ALL": "C"},
|
|
capture_output=True, text=True, timeout=10,
|
|
)
|
|
self.assertEqual(result.returncode, 100, result.stdout + result.stderr)
|
|
self.assertIn("Pre-Invoke", result.stderr)
|
|
|
|
def test_incomplete_index_is_a_real_apt_failure_not_a_warning_pass(self):
|
|
# An unlistening reserved loopback port gives a definite transport
|
|
# failure. No external request, privileged apt operation, host list,
|
|
# post-update hook or authentication bypass is involved.
|
|
with socket.socket() as sock, tempfile.TemporaryDirectory() as temp:
|
|
sock.bind(("127.0.0.1", 0))
|
|
root = Path(temp)
|
|
(root / "lists/partial").mkdir(parents=True)
|
|
(root / "cache/archives/partial").mkdir(parents=True)
|
|
sources = root / "sources.list"
|
|
sources.write_text(f"deb http://127.0.0.1:{sock.getsockname()[1]} stable main\n")
|
|
config = root / "apt.conf"
|
|
prefix = (
|
|
'Dir::Etc::parts "-";\nDir::Etc::main "-";\n'
|
|
f'Dir::Etc::sourcelist "{sources}";\nDir::Etc::sourceparts "-";\n'
|
|
f'Dir::State::lists "{root / "lists"}";\n'
|
|
f'Dir::Cache "{root / "cache"}";\n'
|
|
'Dir::State::status "/dev/null";\nDebug::NoLocking "true";\n'
|
|
)
|
|
for strict, expected_exit in ((True, 100), (False, 0)):
|
|
policy = POLICY.read_text().replace(UBUNTU_SOURCES, str(sources))
|
|
if not strict:
|
|
policy = policy.replace('APT::Update::Error-Mode "any";', "")
|
|
# The controlled source overrides only the distribution path.
|
|
# Keep this a private loopback failure after policy is loaded.
|
|
config.write_text(policy + prefix)
|
|
result = subprocess.run(
|
|
["apt-get", "update"],
|
|
env={**os.environ, "APT_CONFIG": str(config), "LC_ALL": "C"},
|
|
capture_output=True, text=True, timeout=20,
|
|
)
|
|
with self.subTest(strict=strict):
|
|
self.assertEqual(result.returncode, expected_exit, result.stderr)
|
|
self.assertIn("Failed to fetch http://127.0.0.1:", result.stderr)
|
|
self.assertIn("Connection refused", result.stdout + result.stderr)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|