Skip to content
Merged
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
42 changes: 34 additions & 8 deletions tests/test_cookbook.py
Original file line number Diff line number Diff line change
Expand Up @@ -18,6 +18,7 @@
from __future__ import annotations

import os
import shutil
import subprocess
import sys
from pathlib import Path
Expand All @@ -40,15 +41,37 @@
)


def _run(recipe: str, no_network: Path) -> subprocess.CompletedProcess[str]:
def _sandbox(recipe: str, tmp_path: Path) -> Path:
"""A private copy of one recipe's directory.

**Recipes used to run in `examples/cookbook/<name>/` itself, and two tests share a recipe.**
Under `pytest -n auto` they land on different workers and run concurrently in that one
directory, so `verify-in-github-actions`, whose script writes `verify-report.json`, reads it
and then `rm -f`s it, raced itself: one worker's delete landed between the other's write and
read, `bash -euo pipefail` turned the `FileNotFoundError` into a non-zero exit, and the run
went red with nothing wrong in the library.

It reddened two different branches in one afternoon before it was reproduced, which is the
cost worth naming: a suite that fails once in a while teaches people to re-run it, and this
project's gate is only worth anything while green means green.

Copying also stops the suite writing into the working tree at all, so `git status` after a
test run says what it should.
"""
destination = tmp_path / recipe
shutil.copytree(COOKBOOK / recipe, destination)
return destination


def _run(recipe: str, no_network: Path, where: Path) -> subprocess.CompletedProcess[str]:
environment = dict(os.environ)
environment["PYTHONPATH"] = os.pathsep.join(
part for part in (str(no_network), environment.get("PYTHONPATH", "")) if part
)
environment["PYTHONDONTWRITEBYTECODE"] = "1"
for name in ("CTRLRUN_CONFIG", "CTRLRUN_STATE", "CTRLRUN_STORE_URL"):
environment.pop(name, None)
script = COOKBOOK / recipe / "main.py"
script = where / "main.py"
command = (
[sys.executable, str(script)] if script.exists() else ["bash", "-euo", "pipefail", "run.sh"]
)
Expand All @@ -58,7 +81,7 @@ def _run(recipe: str, no_network: Path) -> subprocess.CompletedProcess[str]:
)
return subprocess.run(
command,
cwd=COOKBOOK / recipe,
cwd=where,
env=environment,
capture_output=True,
text=True,
Expand All @@ -68,17 +91,20 @@ def _run(recipe: str, no_network: Path) -> subprocess.CompletedProcess[str]:


@pytest.mark.parametrize("recipe", RECIPES)
def test_every_recipe_runs_offline_and_is_repeatable(recipe, no_network):
first = _run(recipe, no_network)
def test_every_recipe_runs_offline_and_is_repeatable(recipe, no_network, tmp_path):
"""Twice in **one** directory, which is the point, and that directory is this test's own."""
where = _sandbox(recipe, tmp_path)

first = _run(recipe, no_network, where)
assert first.returncode == 0, f"{recipe} failed:\n{first.stdout}\n{first.stderr}"
second = _run(recipe, no_network)
second = _run(recipe, no_network, where)
assert second.returncode == 0, f"{recipe} is not repeatable:\n{second.stdout}\n{second.stderr}"


@pytest.mark.parametrize("recipe", RECIPES)
def test_every_recipe_refuses_something_and_says_so(recipe, no_network):
def test_every_recipe_refuses_something_and_says_so(recipe, no_network, tmp_path):
"""The share unit is a failure and a refusal: every recipe's output shows one."""
output = _run(recipe, no_network).stdout.lower()
output = _run(recipe, no_network, _sandbox(recipe, tmp_path)).stdout.lower()
refusals = (
"refused",
"blocked",
Expand Down
Loading