diff --git a/bootstrap/update.sh b/bootstrap/update.sh new file mode 100755 index 0000000..ece2145 --- /dev/null +++ b/bootstrap/update.sh @@ -0,0 +1,56 @@ +#!/usr/bin/env bash +# Lightweight idempotent update script. +# Run after `git pull` to apply code + config changes without re-registering +# the runner or touching /etc/hil-bench/config.yaml. +# +# Usage: sudo ./bootstrap/update.sh [config-path] +set -euo pipefail + +CONFIG_PATH="${1:-/etc/hil-bench/config.yaml}" +SCRIPT_DIR="$(cd "$(dirname "$0")" && pwd)" +REPO_DIR="$(dirname "$SCRIPT_DIR")" +VENV="/opt/hil-bench/venv" +RUNNER_DIR="/opt/hil-bench/actions-runner" + +echo "=== HIL Bench Update ===" +echo "Repo: ${REPO_DIR}" +echo "Config: ${CONFIG_PATH}" +echo "" + +# ── 1. Reinstall benchctl into venv ────────────────────────────────────── + +if [[ -d "$VENV" ]]; then + echo "--- Updating benchctl in venv ---" + "${VENV}/bin/pip" install --quiet -e "$REPO_DIR" + echo "benchctl updated" +else + echo "WARNING: venv not found at ${VENV} — run install_python_env.sh first" +fi + +# ── 2. Ensure work directory exists ────────────────────────────────────── + +mkdir -p /opt/hil-bench/_work +echo "Work directory: /opt/hil-bench/_work" + +# ── 3. Refresh runner .env (if runner is installed) ────────────────────── + +if [[ -d "$RUNNER_DIR" ]]; then + echo "--- Refreshing runner .env ---" + "${SCRIPT_DIR}/write_runner_env.sh" "$RUNNER_DIR" "$VENV" "$CONFIG_PATH" +else + echo "Runner not installed — skipping .env" +fi + +# ── 4. Refresh systemd health timer ───────────────────────────────────── + +echo "--- Refreshing health timer ---" +"${SCRIPT_DIR}/install_health_timer.sh" "$REPO_DIR" + +# ── 5. Done ────────────────────────────────────────────────────────────── + +echo "" +echo "=== Update complete ===" +if [[ -d "$RUNNER_DIR" ]]; then + echo "REMINDER: Restart the runner service to pick up changes:" + echo " sudo systemctl restart actions.runner.*.service" +fi diff --git a/bootstrap/write_runner_env.sh b/bootstrap/write_runner_env.sh new file mode 100755 index 0000000..1ccf925 --- /dev/null +++ b/bootstrap/write_runner_env.sh @@ -0,0 +1,15 @@ +#!/usr/bin/env bash +# Write the runner .env file so jobs see the venv on PATH. +# Usage: write_runner_env.sh +set -euo pipefail + +RUNNER_DIR="${1:?Usage: $0 }" +VENV="${2:?Usage: $0 }" +CONFIG_PATH="${3:?Usage: $0 }" + +cat > "$RUNNER_DIR/.env" < Path: - """Resolve a firmware path — absolute or relative to runner workspace. + """Resolve a firmware path — absolute, relative to CWD, or relative to workspace. - Accepts: - - Absolute path: returned as-is if it exists - - Relative path: resolved against the runner workspace directory - - Glob pattern: if exactly one match, return it + Resolution order: + 1. Absolute path: returned as-is + 2. Relative to CWD: if it exists + 3. Relative to runner workspace: if it exists + 4. Glob against CWD, then workspace: if exactly one match """ path = Path(firmware) @@ -29,17 +30,35 @@ def resolve_firmware_path( if path.is_absolute(): return path - # Relative to workspace — check direct match first, then try glob - resolved = workspace / path - if resolved.exists(): - return resolved + cwd = Path.cwd() - # Try glob + # Relative to CWD first + cwd_resolved = cwd / path + if cwd_resolved.exists(): + return cwd_resolved + + # Relative to workspace + ws_resolved = workspace / path + if ws_resolved.exists(): + return ws_resolved + + # Try glob — CWD first, then workspace pattern = str(firmware) - matches = sorted(workspace.glob(pattern)) - if len(matches) == 1: - return matches[0] - if len(matches) > 1: - raise ArtifactError(f"ambiguous firmware path {firmware!r}: matched {len(matches)} files") - raise ArtifactError(f"firmware {firmware!r} not found in workspace {workspace}") + cwd_matches = sorted(cwd.glob(pattern)) + if len(cwd_matches) == 1: + return cwd_matches[0] + if len(cwd_matches) > 1: + msg = f"ambiguous firmware path {firmware!r}: matched {len(cwd_matches)} files in CWD" + raise ArtifactError(msg) + + ws_matches = sorted(workspace.glob(pattern)) + if len(ws_matches) == 1: + return ws_matches[0] + if len(ws_matches) > 1: + msg = f"ambiguous firmware path {firmware!r}: matched {len(ws_matches)} files in workspace" + raise ArtifactError(msg) + + raise ArtifactError( + f"firmware {firmware!r} not found in CWD ({cwd}) or workspace ({workspace})" + ) diff --git a/src/hilbench/cli/health_cmd.py b/src/hilbench/cli/health_cmd.py index d7d3e0f..74540d6 100644 --- a/src/hilbench/cli/health_cmd.py +++ b/src/hilbench/cli/health_cmd.py @@ -8,20 +8,29 @@ from rich.console import Console from rich.table import Table -from hilbench.health import results_to_dicts, run_all_checks +from hilbench.health import CHECK_CATEGORIES, results_to_dicts, run_checks @click.command() @click.option("--json-output", "--json", "as_json", is_flag=True, help="Output as JSON.") +@click.option( + "--check", + "-c", + "checks", + multiple=True, + type=click.Choice(CHECK_CATEGORIES, case_sensitive=False), + help="Run only specific checks (repeatable). Default: all.", +) @click.pass_obj -def health(ctx: object, as_json: bool) -> None: +def health(ctx: object, as_json: bool, checks: tuple[str, ...]) -> None: """Run health checks on the bench.""" from hilbench.cli.main import Context assert isinstance(ctx, Context) console = Console() cfg = ctx.config - results = run_all_checks(cfg) + categories = list(checks) if checks else None + results = run_checks(cfg, categories=categories) try: from hilbench.publisher import on_health_complete diff --git a/src/hilbench/health.py b/src/hilbench/health.py index 42fa7de..c9ea04d 100644 --- a/src/hilbench/health.py +++ b/src/hilbench/health.py @@ -11,6 +11,8 @@ from hilbench.probe import probe_factory if TYPE_CHECKING: + from collections.abc import Callable + from hilbench.config import BenchConfig logger = logging.getLogger(__name__) @@ -96,12 +98,30 @@ def check_runner_service() -> CheckResult: ) -def run_all_checks(config: BenchConfig) -> list[CheckResult]: - """Run all health checks and return results.""" +_CHECK_RUNNERS: dict[str, Callable[[BenchConfig], list[CheckResult]]] = { + "config": lambda cfg: [check_config(cfg)], + "probe": check_probe, + "serial": check_serial, + "gpio_chip": lambda cfg: [check_gpio_chip()], + "runner_service": lambda cfg: [check_runner_service()], +} + +CHECK_CATEGORIES: list[str] = list(_CHECK_RUNNERS.keys()) + + +def run_checks( + config: BenchConfig, + categories: list[str] | None = None, +) -> list[CheckResult]: + """Run health checks, optionally filtered to specific categories.""" + selected = categories if categories else CHECK_CATEGORIES results: list[CheckResult] = [] - results.append(check_config(config)) - results.extend(check_probe(config)) - results.extend(check_serial(config)) - results.append(check_gpio_chip()) - results.append(check_runner_service()) + for cat in selected: + runner = _CHECK_RUNNERS[cat] + results.extend(runner(config)) return results + + +def run_all_checks(config: BenchConfig) -> list[CheckResult]: + """Run all health checks and return results.""" + return run_checks(config) diff --git a/tests/test_artifacts.py b/tests/test_artifacts.py new file mode 100644 index 0000000..ca25fde --- /dev/null +++ b/tests/test_artifacts.py @@ -0,0 +1,103 @@ +"""Tests for firmware artifact path resolution.""" + +from __future__ import annotations + +from typing import TYPE_CHECKING + +import pytest + +if TYPE_CHECKING: + from pathlib import Path + +from hilbench.artifacts import resolve_firmware_path +from hilbench.exceptions import ArtifactError + + +class TestResolveFirmwarePath: + def test_absolute_path_returned_as_is(self, tmp_path: Path) -> None: + fw = tmp_path / "firmware.elf" + fw.write_bytes(b"\x00") + result = resolve_firmware_path(str(fw), workspace=tmp_path / "ws") + assert result == fw + + def test_cwd_takes_priority_over_workspace( + self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch + ) -> None: + """A file in CWD is found before the workspace is checked.""" + cwd_dir = tmp_path / "cwd" + ws_dir = tmp_path / "ws" + cwd_dir.mkdir() + ws_dir.mkdir() + + cwd_fw = cwd_dir / "test.bin" + cwd_fw.write_bytes(b"\x01") + ws_fw = ws_dir / "test.bin" + ws_fw.write_bytes(b"\x02") + + monkeypatch.chdir(cwd_dir) + result = resolve_firmware_path("test.bin", workspace=ws_dir) + assert result == cwd_fw + + def test_workspace_fallback(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + """Falls back to workspace when CWD doesn't contain the file.""" + cwd_dir = tmp_path / "cwd" + ws_dir = tmp_path / "ws" + cwd_dir.mkdir() + ws_dir.mkdir() + + ws_fw = ws_dir / "test.bin" + ws_fw.write_bytes(b"\x02") + + monkeypatch.chdir(cwd_dir) + result = resolve_firmware_path("test.bin", workspace=ws_dir) + assert result == ws_fw + + def test_glob_cwd_first(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + """Glob matches CWD before workspace.""" + cwd_dir = tmp_path / "cwd" + ws_dir = tmp_path / "ws" + cwd_dir.mkdir() + ws_dir.mkdir() + + cwd_fw = cwd_dir / "firmware.elf" + cwd_fw.write_bytes(b"\x01") + + monkeypatch.chdir(cwd_dir) + result = resolve_firmware_path("*.elf", workspace=ws_dir) + assert result == cwd_fw + + def test_glob_workspace_fallback( + self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch + ) -> None: + """Glob falls back to workspace when no CWD matches.""" + cwd_dir = tmp_path / "cwd" + ws_dir = tmp_path / "ws" + cwd_dir.mkdir() + ws_dir.mkdir() + + ws_fw = ws_dir / "firmware.elf" + ws_fw.write_bytes(b"\x02") + + monkeypatch.chdir(cwd_dir) + result = resolve_firmware_path("*.elf", workspace=ws_dir) + assert result == ws_fw + + def test_not_found_raises(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + cwd_dir = tmp_path / "cwd" + ws_dir = tmp_path / "ws" + cwd_dir.mkdir() + ws_dir.mkdir() + monkeypatch.chdir(cwd_dir) + + with pytest.raises(ArtifactError, match="not found"): + resolve_firmware_path("nope.bin", workspace=ws_dir) + + def test_ambiguous_raises(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + cwd_dir = tmp_path / "cwd" + cwd_dir.mkdir() + (cwd_dir / "a.elf").write_bytes(b"\x01") + (cwd_dir / "b.elf").write_bytes(b"\x02") + monkeypatch.chdir(cwd_dir) + + with pytest.raises(ArtifactError, match="ambiguous"): + resolve_firmware_path("*.elf", workspace=tmp_path / "ws") diff --git a/tests/test_cli.py b/tests/test_cli.py index aa84553..2955b40 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -115,6 +115,14 @@ def test_health_no_config(self) -> None: # Should fail — no config at default path assert result.exit_code != 0 + def test_health_check_config_only(self, sample_config_path: Path) -> None: + runner = CliRunner() + result = runner.invoke( + cli, ["--config", str(sample_config_path), "health", "--check", "config"] + ) + assert result.exit_code == 0 + assert "config" in result.output + def test_publish_config_no_env(self) -> None: runner = CliRunner() result = runner.invoke(cli, ["publish", "config"]) diff --git a/tests/test_health.py b/tests/test_health.py index 2f139ad..bfe560f 100644 --- a/tests/test_health.py +++ b/tests/test_health.py @@ -10,6 +10,7 @@ check_gpio_chip, check_runner_service, run_all_checks, + run_checks, ) if TYPE_CHECKING: @@ -41,3 +42,34 @@ def test_run_all_checks(self, sample_config: BenchConfig) -> None: assert len(results) >= 4 names = [r.name for r in results] assert "config" in names + + def test_run_checks_filtered(self, sample_config: BenchConfig) -> None: + """Only config check runs when filtered to ['config'].""" + results = run_checks(sample_config, categories=["config"]) + assert len(results) == 1 + assert results[0].name == "config" + assert results[0].passed + + def test_run_checks_multiple_categories(self, sample_config: BenchConfig) -> None: + """Multiple categories run their respective checks.""" + with patch("hilbench.health.probe_factory") as mock_pf: + mock_probe = type("P", (), {"is_connected": lambda self: False})() + mock_pf.return_value = mock_probe + results = run_checks(sample_config, categories=["config", "probe"]) + + names = [r.name for r in results] + assert "config" in names + assert any(n.startswith("probe:") for n in names) + # serial, gpio_chip, runner_service should NOT be present + assert not any(n.startswith("serial:") for n in names) + assert "gpio_chip" not in names + assert "runner_service" not in names + + def test_run_checks_none_means_all(self, sample_config: BenchConfig) -> None: + """Passing categories=None runs all checks (same as run_all_checks).""" + with patch("hilbench.health.probe_factory") as mock_pf: + mock_probe = type("P", (), {"is_connected": lambda self: False})() + mock_pf.return_value = mock_probe + results = run_checks(sample_config, categories=None) + + assert len(results) >= 4