"""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] = {} self.default_result = _FakeResult(stdout="ok") type(self).instances.append(self) 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 = 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