Simone/local-ci-cd-system
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
7f2b03029a90232ccdd393c10af033ec3721806c
local-ci-cd-system/tests/python/test_commands_build.py
T
Simone a7a22a29be feat(cache): restore and publish the cargo dependency cache around Linux builds 
Restore runs after the source is in place and before the build; publish runs
only after a successful build and only on a miss, so the steady state uploads
nothing and a failed build can never become an entry. The whole path is
best-effort: a cache problem degrades to a cold build with a message rather
than turning a good build red.

Gated on Cargo.lock being present, so non-Rust jobs are untouched with no flag
to set. The Windows branch is not touched at all.

The prune script is read through importlib.resources rather than a path
relative to the repo root: the production venv is installed non-editable, so a
parents[] walk lands in site-packages where scripts/ does not exist.

The fake transport gained substring-matched canned results; indexing by call
position broke as soon as the cache added calls mid-flow.

Mutation-tested rather than assumed. Three guards were each disabled in turn --
the publish call, the prune-refusal check, the packing-failure check -- and a
test went red for each. The prune-refusal test initially survived its mutant:
it passed because the fake's fetch left no file behind, so publishing failed
for an unrelated reason. Fixed by making fetch write a real file, so the only
thing preventing publication is the guard under test.
2026-07-22 20:59:23 +02:00

1060 lines
36 KiB
Python
Raw Blame History

"""Tests for ``ci_orchestrator build run``."""
from __future__ import annotations
from pathlib import Path
from typing import Any, ClassVar
import pytest
from click.testing import CliRunner
import ci_orchestrator.commands.build as build_module
from ci_orchestrator.__main__ import cli
from ci_orchestrator.config import BackendConfig, Config, Paths
from ci_orchestrator.credentials import Credential
from ci_orchestrator.transport.errors import TransportCommandError, TransportError
class _FakeResult:
def __init__(self, stdout: str = "", stderr: str = "", returncode: int = 0) -> None:
self.stdout = stdout
self.stderr = stderr
self.returncode = returncode
@property
def ok(self) -> bool:
return self.returncode == 0
class _FakeTransport:
"""Records every ``run/copy/fetch`` call."""
instances: ClassVar[list[_FakeTransport]] = []
def __init__(self, *args: Any, **kwargs: Any) -> None:
self.args = args
self.kwargs = kwargs
self.runs: list[tuple[str, bool]] = []
self.copies: list[tuple[str, str]] = []
self.fetches: list[tuple[str, str]] = []
self.run_results: dict[int, _FakeResult] = {}
# Matched by substring against the script, checked before the
# positional run_results. Index-based fixtures get brittle once a
# feature adds calls in the middle of the flow (the cargo cache does).
self.script_results: list[tuple[str, _FakeResult]] = []
self.default_result = _FakeResult(stdout="ok")
type(self).instances.append(self)
def result_for(
self, needle: str, *, stdout: str = "", stderr: str = "", returncode: int = 0
) -> None:
self.script_results.append(
(needle, _FakeResult(stdout=stdout, stderr=stderr, returncode=returncode))
)
def __enter__(self) -> _FakeTransport:
return self
def __exit__(self, *_e: object) -> None:
return None
def run(self, script: str, *, check: bool = True) -> _FakeResult:
self.runs.append((script, check))
result: _FakeResult | None = None
for needle, canned in self.script_results:
if needle in script:
result = canned
break
if result is None:
result = self.run_results.get(len(self.runs) - 1, self.default_result)
if check and not result.ok:
raise TransportCommandError(result.returncode, result.stdout, result.stderr)
return result
def run_streaming(
self, script: str, *, check: bool = True, timeout: float | None = None
) -> _FakeResult:
# Tests don't exercise live streaming; delegate to run() so
# capture + check semantics (and _BuildFails overrides) hold.
return self.run(script, check=check)
def copy(self, local_path: str, remote_path: str) -> None:
self.copies.append((str(local_path), remote_path))
def fetch(self, remote_path: str, local_path: str) -> None:
self.fetches.append((remote_path, str(local_path)))
@pytest.fixture(autouse=True)
def _reset_instances() -> None:
_FakeTransport.instances = []
def _patch(monkeypatch: pytest.MonkeyPatch) -> type[_FakeTransport]:
class _Store:
def get(self, _t: str) -> Credential:
return Credential("ci", "pwd")
monkeypatch.setattr(build_module, "WinRmTransport", _FakeTransport)
monkeypatch.setattr(build_module, "SshTransport", _FakeTransport)
monkeypatch.setattr(build_module, "KeyringCredentialStore", lambda: _Store())
return _FakeTransport
# ── parameter validation ───────────────────────────────────────────────────
def test_build_run_requires_source(monkeypatch: pytest.MonkeyPatch) -> None:
_patch(monkeypatch)
result = CliRunner().invoke(
cli, ["build", "run", "--ip-address", "10.0.0.1"]
)
assert result.exit_code != 0
assert "either --host-source-dir or --clone-url" in result.output
def test_build_run_rejects_both_modes(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None:
_patch(monkeypatch)
src = tmp_path / "src"
src.mkdir()
result = CliRunner().invoke(
cli,
[
"build",
"run",
"--ip-address",
"10.0.0.1",
"--host-source-dir",
str(src),
"--clone-url",
"https://example/repo.git",
],
)
assert result.exit_code != 0
assert "not both" in result.output
def test_build_run_rejects_missing_source_dir(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None:
_patch(monkeypatch)
result = CliRunner().invoke(
cli,
[
"build",
"run",
"--ip-address",
"10.0.0.1",
"--host-source-dir",
str(tmp_path / "ghost"),
],
)
assert result.exit_code != 0
assert "does not exist" in result.output
def test_build_run_rejects_bad_extra_env(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None:
_patch(monkeypatch)
src = tmp_path / "src"
src.mkdir()
result = CliRunner().invoke(
cli,
[
"build",
"run",
"--ip-address",
"10.0.0.1",
"--guest-os",
"linux",
"--host-source-dir",
str(src),
"--build-command",
"make",
"--extra-env",
"NO_EQUALS",
],
)
assert result.exit_code != 0
assert "KEY=VALUE" in result.output
# ── happy path: Linux ──────────────────────────────────────────────────────
def test_build_run_linux_clone_url_happy_path(
monkeypatch: pytest.MonkeyPatch, tmp_path: Path
) -> None:
_patch(monkeypatch)
result = CliRunner().invoke(
cli,
[
"build",
"run",
"--ip-address",
"10.0.0.5",
"--guest-os",
"linux",
"--clone-url",
"https://example/repo.git",
"--clone-branch",
"feature/x",
"--clone-commit",
"deadbeef",
"--clone-submodules",
"--build-command",
"make all",
"--guest-linux-work-dir",
"/work",
"--guest-linux-output-dir",
"/out",
"--guest-artifact-source",
"build",
"--extra-env",
"FOO=bar baz",
"--extra-env",
"TOKEN=secret",
"--ssh-key-path",
str(tmp_path / "key"),
],
)
assert result.exit_code == 0, result.output
instance = _FakeTransport.instances[0]
cmds = [c[0] for c in instance.runs]
assert any("rm -rf /work && mkdir -p /work" in c for c in cmds)
assert any("git clone --depth 1 --branch feature/x" in c for c in cmds)
assert any("--recurse-submodules" in c for c in cmds)
assert any("checkout deadbeef" in c for c in cmds)
# extra-env exported before build
assert any("export FOO='bar baz'" in c and "export TOKEN=secret" in c for c in cmds)
assert any("cd /work && make all" in c for c in cmds)
# artifact stage uses the requested source dir
assert any("/work/build" in c for c in cmds)
def test_build_run_linux_xvfb_wraps_command(
monkeypatch: pytest.MonkeyPatch, tmp_path: Path
) -> None:
"""--xvfb wraps the Linux build in xvfb-run; env exports stay outside."""
_patch(monkeypatch)
result = CliRunner().invoke(
cli,
[
"build", "run",
"--ip-address", "10.0.0.5",
"--guest-os", "linux",
"--clone-url", "https://example/repo.git",
"--build-command", "make all",
"--guest-linux-work-dir", "/work",
"--extra-env", "FOO=bar",
"--xvfb",
"--ssh-key-path", str(tmp_path / "key"),
],
)
assert result.exit_code == 0, result.output
cmds = [c[0] for c in _FakeTransport.instances[0].runs]
# Build wrapped under xvfb-run via sh -c; env export precedes it.
assert any(
"xvfb-run -a sh -c 'cd /work && make all'" in c
and "export FOO=bar" in c
for c in cmds
)
def test_build_run_linux_no_xvfb_runs_plain(
monkeypatch: pytest.MonkeyPatch, tmp_path: Path
) -> None:
"""Without --xvfb the build runs directly (no xvfb-run wrapper)."""
_patch(monkeypatch)
result = CliRunner().invoke(
cli,
[
"build", "run",
"--ip-address", "10.0.0.5",
"--guest-os", "linux",
"--clone-url", "https://example/repo.git",
"--build-command", "make all",
"--guest-linux-work-dir", "/work",
"--ssh-key-path", str(tmp_path / "key"),
],
)
assert result.exit_code == 0, result.output
cmds = [c[0] for c in _FakeTransport.instances[0].runs]
assert any("cd /work && make all" in c for c in cmds)
assert not any("xvfb-run" in c for c in cmds)
def test_build_run_xvfb_ignored_on_windows(
monkeypatch: pytest.MonkeyPatch, tmp_path: Path
) -> None:
"""--xvfb is a no-op (with a notice) for Windows guests."""
_patch(monkeypatch)
src = tmp_path / "src"
src.mkdir()
(src / "f.txt").write_text("x")
result = CliRunner().invoke(
cli,
[
"build", "run",
"--ip-address", "10.0.0.9",
"--guest-os", "windows",
"--host-source-dir", str(src),
"--build-command", "echo hi",
"--xvfb",
],
)
assert result.exit_code == 0, result.output
assert "--xvfb ignored for Windows guest" in result.output
cmds = [c[0] for c in _FakeTransport.instances[0].runs]
assert not any("xvfb-run" in c for c in cmds)
def test_build_run_linux_host_source_dir(
monkeypatch: pytest.MonkeyPatch, tmp_path: Path
) -> None:
_patch(monkeypatch)
src = tmp_path / "src"
src.mkdir()
(src / "hello.txt").write_text("hi", encoding="utf-8")
result = CliRunner().invoke(
cli,
[
"build",
"run",
"--ip-address",
"10.0.0.5",
"--guest-os",
"linux",
"--host-source-dir",
str(src),
"--build-command",
"true",
"--ssh-key-path",
str(tmp_path / "key"),
],
)
assert result.exit_code == 0, result.output
instance = _FakeTransport.instances[0]
# A single tarball was uploaded.
assert len(instance.copies) == 1
local, remote = instance.copies[0]
assert local.endswith(".tar.gz")
assert remote.startswith("/tmp/")
# Extraction command was invoked.
assert any("tar -xzf" in c[0] for c in instance.runs)
def test_build_run_linux_skip_artifact(
monkeypatch: pytest.MonkeyPatch, tmp_path: Path
) -> None:
_patch(monkeypatch)
result = CliRunner().invoke(
cli,
[
"build",
"run",
"--ip-address",
"10.0.0.5",
"--guest-os",
"linux",
"--clone-url",
"https://example/repo.git",
"--build-command",
"make",
"--skip-artifact",
"--ssh-key-path",
str(tmp_path / "key"),
],
)
assert result.exit_code == 0, result.output
cmds = [c[0] for c in _FakeTransport.instances[0].runs]
# No staging command should run.
assert not any("/opt/ci/output" in c for c in cmds)
def test_build_run_linux_build_failure_exits_nonzero(
monkeypatch: pytest.MonkeyPatch, tmp_path: Path
) -> None:
_patch(monkeypatch)
# Make the build command (4th run: rm/git clone/checkout-skipped/build)
# return non-zero. Index of build run varies; easiest: make any non-zero
# result for the run whose script contains 'cd '.
src = tmp_path / "src"
src.mkdir()
class _BuildFails(_FakeTransport):
def run(self, script: str, *, check: bool = True) -> _FakeResult:
if "cd /opt/ci/build && make" in script:
self.runs.append((script, check))
return _FakeResult(stdout="boom\n", stderr="err\n", returncode=2)
return super().run(script, check=check)
monkeypatch.setattr(build_module, "SshTransport", _BuildFails)
result = CliRunner().invoke(
cli,
[
"build",
"run",
"--ip-address",
"10.0.0.5",
"--guest-os",
"linux",
"--host-source-dir",
str(src),
"--ssh-key-path",
str(tmp_path / "key"),
],
)
assert result.exit_code != 0
assert "build command failed (exit 2)" in result.output
# ── happy path: Windows ────────────────────────────────────────────────────
def test_build_run_windows_clone_url_default_dotnet(
monkeypatch: pytest.MonkeyPatch, tmp_path: Path
) -> None:
_patch(monkeypatch)
result = CliRunner().invoke(
cli,
[
"build",
"run",
"--ip-address",
"10.0.0.7",
"--guest-os",
"windows",
"--clone-url",
"https://example/repo.git",
"--use-shared-cache",
],
)
assert result.exit_code == 0, result.output
cmds = [c[0] for c in _FakeTransport.instances[0].runs]
assert any("git clone --depth 1" in c for c in cmds)
assert any("dotnet restore" in c for c in cmds)
assert any("NUGET_PACKAGES" in c for c in cmds)
def test_build_run_windows_host_source_dir(
monkeypatch: pytest.MonkeyPatch, tmp_path: Path
) -> None:
_patch(monkeypatch)
src = tmp_path / "src"
src.mkdir()
(src / "a.cs").write_text("//", encoding="utf-8")
result = CliRunner().invoke(
cli,
[
"build",
"run",
"--ip-address",
"10.0.0.7",
"--host-source-dir",
str(src),
"--build-command",
"echo built > dist\\out.txt",
"--guest-artifact-source",
"dist",
],
)
assert result.exit_code == 0, result.output
instance = _FakeTransport.instances[0]
assert len(instance.copies) == 1
assert instance.copies[0][1] == "C:\\CI\\src-transfer.zip"
cmds = [c[0] for c in instance.runs]
assert any("Expand-Archive" in c for c in cmds)
# Staging copies the raw source into the collect dir (no zip-in-zip).
assert any("Copy-Item" in c and "'dist'" in c for c in cmds)
assert not any("Compress-Archive" in c and "'dist'" in c for c in cmds)
# ── _parse_extra_env unit tests ────────────────────────────────────────────
def test_parse_extra_env_empty_string_skipped() -> None:
"""Line 62: empty strings in the iterable are silently skipped (continue branch)."""
result = build_module._parse_extra_env(["", "FOO=bar", ""])
assert result == {"FOO": "bar"}
def test_parse_extra_env_no_equals_raises() -> None:
"""Line 63-66: value with no '=' must raise BadParameter."""
import click
with pytest.raises(click.BadParameter, match="KEY=VALUE"):
build_module._parse_extra_env(["BADVALUE"])
def test_parse_extra_env_empty_key_raises() -> None:
"""Line 69: '=value' with empty key must raise BadParameter."""
import click
with pytest.raises(click.BadParameter, match="key cannot be empty"):
build_module._parse_extra_env(["=value"])
# ── _zip_dir skips subdirectories ─────────────────────────────────────────
def test_zip_dir_skips_subdirectories(tmp_path: Path) -> None:
"""Line 80->79: _zip_dir iterates rglob('*'); directories are skipped (is_file() == False)."""
import zipfile
src = tmp_path / "src"
src.mkdir()
(src / "file.txt").write_text("hello", encoding="utf-8")
subdir = src / "subdir"
subdir.mkdir()
(subdir / "nested.txt").write_text("world", encoding="utf-8")
archive = tmp_path / "out.zip"
build_module._zip_dir(src, archive)
with zipfile.ZipFile(archive) as zf:
names = zf.namelist()
# Only files should appear; the bare subdirectory entry must not be present.
assert "file.txt" in names
assert str(Path("subdir") / "nested.txt") in names or "subdir/nested.txt" in names
# No entry that ends in '/' (a directory entry).
assert not any(n.endswith("/") for n in names)
# ── _report_build_output: stdout empty, stderr present ────────────────────
def test_report_build_output_stderr_only(capsys: pytest.CaptureFixture[str]) -> None:
"""Lines 108->111, 112-113: when stdout is empty, only stderr branch executes."""
result = build_module._report_build_output("", "some error text")
captured = capsys.readouterr()
assert "some error text" in captured.err
assert captured.out.count("[build run]") == 2 # header + footer lines
# The tail should reflect the stderr content.
assert "some error text" in result
# ── _linux_build: neither host_source_dir nor clone_url ───────────────────
def test_linux_build_no_source_raises(monkeypatch: pytest.MonkeyPatch) -> None:
"""Line 189: _linux_build with no source raises ClickException."""
import click
monkeypatch.setattr(build_module, "SshTransport", _FakeTransport)
with pytest.raises(click.ClickException, match="either --host-source-dir or --clone-url"):
build_module._linux_build(
ip_address="10.0.0.1",
ssh_user="ci_build",
key_path=None,
known_hosts=None,
workdir="/opt/ci/build",
output_dir="/opt/ci/output",
host_source_dir=None,
clone_url=None,
clone_branch="main",
clone_commit="",
clone_submodules=False,
build_command="make",
artifact_source="dist",
extra_env={},
skip_artifact=False,
)
# ── _linux_build: clone_url + clone_commit triggers fetch+checkout ─────────
def test_linux_build_clone_commit_fetch_checkout(monkeypatch: pytest.MonkeyPatch) -> None:
"""Lines 230-233: clone_commit causes 3 subprocess calls (clone, fetch, checkout) via SSH."""
monkeypatch.setattr(build_module, "SshTransport", _FakeTransport)
build_module._linux_build(
ip_address="10.0.0.1",
ssh_user="ci_build",
key_path=None,
known_hosts=None,
workdir="/opt/ci/build",
output_dir="/opt/ci/output",
host_source_dir=None,
clone_url="http://repo/example.git",
clone_branch="main",
clone_commit="abc123",
clone_submodules=False,
build_command="true",
artifact_source="dist",
extra_env={},
skip_artifact=True,
)
instance = _FakeTransport.instances[0]
cmds = [c[0] for c in instance.runs]
assert any("git clone" in c for c in cmds)
assert any("fetch --depth 1 origin" in c and "abc123" in c for c in cmds)
assert any("checkout" in c and "abc123" in c for c in cmds)
# ── _windows_build: clone_url + clone_commit triggers fetch+checkout ───────
def test_windows_build_clone_commit_fetch_checkout(monkeypatch: pytest.MonkeyPatch) -> None:
"""Lines 315-321: clone_commit in windows path causes fetch+checkout run."""
_patch(monkeypatch)
build_module._windows_build(
ip_address="10.0.0.7",
credential_target="BuildVMGuest",
workdir="C:\\CI\\build",
output_dir="C:\\CI\\output",
host_source_dir=None,
clone_url="https://example/repo.git",
clone_branch="main",
clone_commit="deadbeef",
clone_submodules=False,
build_command="echo done",
artifact_source="dist",
configuration="Release",
extra_env={},
skip_artifact=True,
use_shared_cache=False,
)
instance = _FakeTransport.instances[0]
cmds = [c[0] for c in instance.runs]
assert any("git clone" in c for c in cmds)
assert any("fetch --depth 1 origin" in c and "deadbeef" in c for c in cmds)
assert any("checkout" in c and "deadbeef" in c for c in cmds)
# ── _windows_build: no source raises ClickException ────────────────────────
def test_windows_build_no_source_raises(monkeypatch: pytest.MonkeyPatch) -> None:
"""Line 321: _windows_build with no source raises ClickException."""
import click
_patch(monkeypatch)
with pytest.raises(click.ClickException, match="either --host-source-dir or --clone-url"):
build_module._windows_build(
ip_address="10.0.0.7",
credential_target="BuildVMGuest",
workdir="C:\\CI\\build",
output_dir="C:\\CI\\output",
host_source_dir=None,
clone_url=None,
clone_branch="main",
clone_commit="",
clone_submodules=False,
build_command="echo done",
artifact_source="dist",
configuration="Release",
extra_env={},
skip_artifact=False,
use_shared_cache=False,
)
def test_windows_build_extra_env_semicolon_appended(monkeypatch: pytest.MonkeyPatch) -> None:
"""Line 330: when extra_env is non-empty, env_setup gets '; ' appended before build command."""
_patch(monkeypatch)
build_module._windows_build(
ip_address="10.0.0.7",
credential_target="BuildVMGuest",
workdir="C:\\CI\\build",
output_dir="C:\\CI\\output",
host_source_dir=None,
clone_url="https://example/repo.git",
clone_branch="main",
clone_commit="",
clone_submodules=False,
build_command="echo done",
artifact_source="dist",
configuration="Release",
extra_env={"MY_VAR": "hello"},
skip_artifact=True,
use_shared_cache=False,
)
instance = _FakeTransport.instances[0]
cmds = [c[0] for c in instance.runs]
# The build command should contain the injected env var followed by '; '
assert any("$env:MY_VAR = 'hello'" in c for c in cmds)
def test_windows_build_build_failure_raises(monkeypatch: pytest.MonkeyPatch) -> None:
"""Lines 365-367: when the build command returns non-zero, ClickException is raised."""
import click
_patch(monkeypatch)
class _FailsBuild(_FakeTransport):
def run(self, script: str, *, check: bool = True) -> _FakeResult:
if "echo fail" in script:
return _FakeResult(stdout="", stderr="boom", returncode=1)
return super().run(script, check=check)
monkeypatch.setattr(build_module, "WinRmTransport", _FailsBuild)
with pytest.raises(click.ClickException, match="build command failed"):
build_module._windows_build(
ip_address="10.0.0.7",
credential_target="BuildVMGuest",
workdir="C:\\CI\\build",
output_dir="C:\\CI\\output",
host_source_dir=None,
clone_url="https://example/repo.git",
clone_branch="main",
clone_commit="",
clone_submodules=False,
build_command="echo fail",
artifact_source="dist",
configuration="Release",
extra_env={},
skip_artifact=False,
use_shared_cache=False,
)
# ── _linux_build: artifact_source == output_dir → collecting in place ─────
def test_linux_build_artifact_source_equals_output_dir(
monkeypatch: pytest.MonkeyPatch,
) -> None:
"""Lines 365, 370-371: when artifact_source is the output_dir itself, the
'collecting in place' branch runs an existence/non-empty check."""
monkeypatch.setattr(build_module, "SshTransport", _FakeTransport)
output_dir = "/opt/ci/output"
build_module._linux_build(
ip_address="10.0.0.1",
ssh_user="ci_build",
key_path=None,
known_hosts=None,
workdir="/opt/ci/build",
output_dir=output_dir,
host_source_dir=None,
clone_url="https://example/repo.git",
clone_branch="main",
clone_commit="",
clone_submodules=False,
build_command="true",
# Use the absolute output dir so srcpath == out
artifact_source=output_dir,
extra_env={},
skip_artifact=False,
)
instance = _FakeTransport.instances[0]
cmds = [c[0] for c in instance.runs]
# The 'collecting in place' guard command must appear.
assert any("if [ ! -d" in c and output_dir in c for c in cmds)
# The wipe-and-copy command must NOT appear.
assert not any("rm -rf" in c and "mkdir -p" in c and output_dir in c for c in cmds)
# ── build run CLI: ssh_key_path from config ────────────────────────────────
def _make_config(ssh_key_path: Path | None = None) -> Config:
paths = Paths(
root=Path("/var/lib/ci"),
templates=Path("/var/lib/ci/templates"),
build_vms=Path("/var/lib/ci/build-vms"),
artifacts=Path("/var/lib/ci/artifacts"),
keys=Path("/var/lib/ci/keys"),
)
return Config(paths=paths, backend=BackendConfig(), ssh_key_path=ssh_key_path)
def test_build_run_linux_ssh_key_from_config(
monkeypatch: pytest.MonkeyPatch, tmp_path: Path
) -> None:
"""Line 548: when --ssh-key-path is omitted but config.ssh_key_path is set, it is used."""
_patch(monkeypatch)
key = tmp_path / "ci_linux"
monkeypatch.setattr(build_module, "load_config", lambda: _make_config(ssh_key_path=key))
result = CliRunner().invoke(
cli,
[
"build",
"run",
"--ip-address",
"10.0.0.5",
"--guest-os",
"linux",
"--clone-url",
"https://example/repo.git",
"--build-command",
"true",
"--skip-artifact",
# deliberately NO --ssh-key-path
],
)
assert result.exit_code == 0, result.output
# The transport must have been instantiated with the config key path.
instance = _FakeTransport.instances[0]
assert str(key) in (instance.kwargs.get("key_path") or instance.args[1] if len(instance.args) > 1 else "") or instance.kwargs.get("key_path") == str(key)
# ── build run CLI: linux TransportError → ClickException ──────────────────
def test_build_run_linux_transport_error(
monkeypatch: pytest.MonkeyPatch, tmp_path: Path
) -> None:
"""Line 568: TransportError from _linux_build is caught and re-raised as ClickException."""
_patch(monkeypatch)
monkeypatch.setattr(build_module, "load_config", lambda: _make_config())
class _RaisesTransport(_FakeTransport):
def __enter__(self) -> _RaisesTransport:
raise TransportError("connection refused")
monkeypatch.setattr(build_module, "SshTransport", _RaisesTransport)
result = CliRunner().invoke(
cli,
[
"build",
"run",
"--ip-address",
"10.0.0.5",
"--guest-os",
"linux",
"--clone-url",
"https://example/repo.git",
"--build-command",
"true",
"--ssh-key-path",
str(tmp_path / "key"),
],
)
assert result.exit_code != 0
assert "connection refused" in result.output
# ── build run CLI: windows TransportError → ClickException ────────────────
def test_build_run_windows_transport_error(
monkeypatch: pytest.MonkeyPatch, tmp_path: Path
) -> None:
"""Lines 591-592: TransportError from _windows_build is caught and re-raised as ClickException."""
_patch(monkeypatch)
monkeypatch.setattr(build_module, "load_config", lambda: _make_config())
class _RaisesTransport(_FakeTransport):
def __enter__(self) -> _RaisesTransport:
raise TransportError("winrm unreachable")
monkeypatch.setattr(build_module, "WinRmTransport", _RaisesTransport)
result = CliRunner().invoke(
cli,
[
"build",
"run",
"--ip-address",
"10.0.0.7",
"--guest-os",
"windows",
"--clone-url",
"https://example/repo.git",
],
)
assert result.exit_code != 0
assert "winrm unreachable" in result.output
# ── _linux_build: cargo dependency cache ───────────────────────────────────
def _linux_kwargs(**overrides: Any) -> dict[str, Any]:
base: dict[str, Any] = {
"ip_address": "10.0.0.1",
"ssh_user": "ci_build",
"key_path": None,
"known_hosts": None,
"workdir": "/opt/ci/build",
"output_dir": "/opt/ci/output",
"host_source_dir": None,
"clone_url": "http://repo/example.git",
"clone_branch": "main",
"clone_commit": "",
"clone_submodules": False,
"build_command": "true",
"artifact_source": "dist",
"extra_env": {},
"skip_artifact": True,
}
base.update(overrides)
return base
def _rust_guest(fake: _FakeTransport, *, digest: str = "cafe") -> None:
"""Make the fake answer like a Rust workspace with a known toolchain."""
fake.result_for("sha256sum", stdout=f"{digest} Cargo.lock\n")
fake.result_for("rustc -V", stdout="rustc 1.97.1 (abc 2026-01-01)\n")
fake.result_for("rustc -vV", stdout="x86_64-unknown-linux-gnu\n")
def _expected_key(digest: str = "cafe") -> str:
from ci_orchestrator.cargo_cache import cache_key
return cache_key(
lock_digest=digest,
rustc_version="rustc 1.97.1 (abc 2026-01-01)",
host_triple="x86_64-unknown-linux-gnu",
)
def _fetch_writes_a_real_file(
monkeypatch: pytest.MonkeyPatch, fake: _FakeTransport
) -> None:
"""Make ``fetch`` produce an actual file on disk.
Without this the default fake leaves nothing behind, ``publish`` dies on
FileNotFoundError, and a "nothing was published" assertion passes for the
wrong reason -- it would hold even with the guard under test removed.
Mutation-tested: dropping the prune-refusal check must make those tests go
red, and it only does when a publish could otherwise have succeeded.
"""
def _fetch(remote: str, local: str) -> None:
fake.fetches.append((remote, str(local)))
Path(local).write_bytes(b"fresh-entry")
monkeypatch.setattr(fake, "fetch", _fetch)
def test_linux_build_without_cache_root_runs_no_cache_commands(
monkeypatch: pytest.MonkeyPatch,
) -> None:
monkeypatch.setattr(build_module, "SshTransport", _FakeTransport)
build_module._linux_build(**_linux_kwargs())
cmds = [c[0] for c in _FakeTransport.instances[0].runs]
assert not any("cargo-cache-prune" in c for c in cmds)
assert not any("sha256sum" in c for c in cmds)
def test_linux_build_skips_cache_when_no_cargo_lock(
monkeypatch: pytest.MonkeyPatch, tmp_path: Path
) -> None:
monkeypatch.setattr(build_module, "SshTransport", _FakeTransport)
fake = _FakeTransport()
_FakeTransport.instances = []
monkeypatch.setattr(build_module, "SshTransport", lambda *a, **k: fake)
fake.result_for("test -f", returncode=1)
build_module._linux_build(**_linux_kwargs(cargo_cache_root=tmp_path))
cmds = [c[0] for c in fake.runs]
assert not any("cargo-cache-prune" in c for c in cmds)
assert not any("sha256sum" in c for c in cmds)
def test_linux_build_restores_an_existing_entry(
monkeypatch: pytest.MonkeyPatch, tmp_path: Path
) -> None:
from ci_orchestrator.cargo_cache import CargoCacheStore
fake = _FakeTransport()
monkeypatch.setattr(build_module, "SshTransport", lambda *a, **k: fake)
_rust_guest(fake)
store = CargoCacheStore(tmp_path)
seed = tmp_path / "seed.tar.zst"
seed.write_bytes(b"entry")
store.publish(_expected_key(), seed)
build_module._linux_build(**_linux_kwargs(cargo_cache_root=tmp_path))
cmds = [c[0] for c in fake.runs]
assert any("tar --zstd -xf" in c for c in cmds), "a hit must unpack the entry"
assert fake.copies, "a hit must upload the entry to the guest"
# A hit must not re-publish: the entry is already there.
assert not any("cargo-cache-prune" in c for c in cmds)
def test_linux_build_publishes_on_a_miss(
monkeypatch: pytest.MonkeyPatch, tmp_path: Path
) -> None:
from ci_orchestrator.cargo_cache import CargoCacheStore
fake = _FakeTransport()
monkeypatch.setattr(build_module, "SshTransport", lambda *a, **k: fake)
_rust_guest(fake)
_fetch_writes_a_real_file(monkeypatch, fake)
build_module._linux_build(**_linux_kwargs(cargo_cache_root=tmp_path))
assert CargoCacheStore(tmp_path).lookup(_expected_key()) is not None
def test_linux_build_does_not_publish_when_the_prune_assertion_fails(
monkeypatch: pytest.MonkeyPatch, tmp_path: Path
) -> None:
from ci_orchestrator.cargo_cache import CargoCacheStore
fake = _FakeTransport()
monkeypatch.setattr(build_module, "SshTransport", lambda *a, **k: fake)
_rust_guest(fake)
_fetch_writes_a_real_file(monkeypatch, fake)
fake.result_for(
"cargo-cache-prune",
returncode=2,
stderr="prune: REFUSING to publish -- artifacts still reference workspace source",
)
build_module._linux_build(**_linux_kwargs(cargo_cache_root=tmp_path))
assert CargoCacheStore(tmp_path).lookup(_expected_key()) is None, (
"a failed prune assertion must never reach the store"
)
def test_linux_build_does_not_publish_when_packing_fails(
monkeypatch: pytest.MonkeyPatch, tmp_path: Path
) -> None:
from ci_orchestrator.cargo_cache import CargoCacheStore
fake = _FakeTransport()
monkeypatch.setattr(build_module, "SshTransport", lambda *a, **k: fake)
_rust_guest(fake)
_fetch_writes_a_real_file(monkeypatch, fake)
fake.result_for("tar --zstd -cf", returncode=1, stderr="no space left on device")
build_module._linux_build(**_linux_kwargs(cargo_cache_root=tmp_path))
assert CargoCacheStore(tmp_path).lookup(_expected_key()) is None
def test_cache_probe_failure_does_not_fail_the_build(
monkeypatch: pytest.MonkeyPatch, tmp_path: Path
) -> None:
fake = _FakeTransport()
monkeypatch.setattr(build_module, "SshTransport", lambda *a, **k: fake)
fake.result_for("sha256sum", returncode=1, stderr="boom")
# Must not raise: the cache is an optimisation, never a failure mode.
build_module._linux_build(**_linux_kwargs(cargo_cache_root=tmp_path))
def test_cache_transport_error_does_not_fail_the_build(
monkeypatch: pytest.MonkeyPatch, tmp_path: Path
) -> None:
fake = _FakeTransport()
monkeypatch.setattr(build_module, "SshTransport", lambda *a, **k: fake)
_rust_guest(fake)
def _boom(*_a: Any, **_k: Any) -> None:
raise TransportError("ssh died mid-restore")
monkeypatch.setattr(fake, "copy", _boom)
build_module._linux_build(**_linux_kwargs(cargo_cache_root=tmp_path))
def test_cache_key_differs_when_the_lockfile_differs(
monkeypatch: pytest.MonkeyPatch, tmp_path: Path
) -> None:
"""A Cargo.lock change must miss the entry built for the previous one."""
from ci_orchestrator.cargo_cache import CargoCacheStore
store = CargoCacheStore(tmp_path)
seed = tmp_path / "seed.tar.zst"
seed.write_bytes(b"entry")
store.publish(_expected_key("cafe"), seed)
fake = _FakeTransport()
monkeypatch.setattr(build_module, "SshTransport", lambda *a, **k: fake)
_rust_guest(fake, digest="beef")
build_module._linux_build(**_linux_kwargs(cargo_cache_root=tmp_path))
cmds = [c[0] for c in fake.runs]
assert not any("tar --zstd -xf" in c for c in cmds), "different lock must miss"
Reference in New Issue View Git Blame Copy Permalink