feat(tests): load live VM host/key from .env, keep inventory non-sensitive

docs/live-testing.md

@@ -0,0 +1,82 @@
+# Live System Testing
+
+Live tests run KIAUH workflows against a real Debian 12 QEMU/KVM VM. They are
+isolated from the local developer machine by design.
+
+## Safety Rules
+
+- Live tests NEVER run on the local machine.
+- They require `KIAUH_LIVE_ALLOW=1`.
+- The target host must be explicitly set via `KIAUH_LIVE_TARGET_HOST` and match
+the VM in the inventory.
+- Local hostnames, loopback addresses, and the current hostname are blocked.
+
+## Prepare a VM
+
+1. Create a Debian 12 QEMU/KVM VM.
+2. Create a user with passwordless sudo.
+3. Install an SSH key for that user.
+4. Install KIAUH on the VM (e.g. clone this repository).
+5. Create a clean snapshot named `clean`:
+   ```bash
+   virsh snapshot-create-as debian12-kiauh clean
+   ```
+
+## Inventory
+
+`kiauh/live/inventory.yaml` defines non-sensitive VM settings. The host and SSH
+key path are read from environment variables or a `.env` file so they are not
+committed.
+
+Create `.env` in the project root:
+
+```bash
+KIAUH_LIVE_DEBIAN12_KIAUH_HOST=192.168.122.10
+KIAUH_LIVE_DEBIAN12_KIAUH_KEY_FILE=/home/you/.ssh/kiauh_vm
+```
+
+Variable naming: `KIAUH_LIVE_<VM_NAME>_HOST` and `KIAUH_LIVE_<VM_NAME>_KEY_FILE`,
+with the VM name uppercased and hyphens replaced by underscores.
+
+You can also point to a custom inventory file:
+
+```bash
+export KIAUH_LIVE_INVENTORY=/path/to/inventory.yaml
+```
+
+## Run Live Tests
+
+```bash
+export KIAUH_LIVE_ALLOW=1
+export KIAUH_LIVE_TARGET_HOST=192.168.122.10
+pytest -m live
+```
+
+Each scenario reverts the VM to the clean snapshot first, so scenarios are
+independent.
+
+## Add a Scenario
+
+Create a YAML file in `kiauh/live/scenarios/`:
+
+```yaml
+name: Install Klipper on Debian 12
+vm: debian12-kiauh
+os: debian-12
+steps:
+  - command: ["kiauh", "install", "klipper", "--count", "1"]
+    timeout: 600
+expected:
+  - type: service
+    name: klipper.service
+    state: running
+```
+
+Supported assertion types: `service`, `file`, `package`, `port`, `command`.
+
+## Troubleshooting
+
+- `UnsafeTargetError`: check `KIAUH_LIVE_ALLOW` and `KIAUH_LIVE_TARGET_HOST`.
+- `InventoryError`: check the inventory YAML path and format.
+- `LiveRunnerError` during snapshot revert: ensure `virsh` works and the domain
+  and snapshot names match the inventory.

docs/prd/PRD-002-live-vm-test-strategy.md

@@ -0,0 +1,74 @@
+## ⚠️ Working on this PRD
+
+Do NOT implement this PRD directly. It has been broken into sequential tasks.
+Work through the tasks below in order.
+
+## Task Index
+
+| # | Task | Todo | Blocked by | Status |
+|---|------|------|------------|--------|
+| 1/5 | VM inventory, SSH fixture, and safety guards | TODO-006 | — | 🔄 open |
+| 2/5 | Scenario loader and Klipper install scenario | TODO-007 | TODO-006 | ⏳ blocked |
+| 3/5 | Remove Klipper and Moonraker scenarios | TODO-008 | TODO-007 | ⏳ blocked |
+| 4/5 | Mainsail and Fluidd install scenarios | TODO-009 | TODO-008 | ⏳ blocked |
+| 5/5 | Backup, restore, and update scenarios | TODO-010 | TODO-009 | ⏳ blocked |
+
+Start with: **TODO-006** (PRD #2 - Task 1/5: VM inventory, SSH fixture, and safety guards)
+
+---
+
+# PRD #2: Isolated live-system acceptance test strategy on Debian 12 VM
+
+**Tags:** `prd`, `prd-2`
+
+## Problem Statement
+
+- Pytest unit tests cannot validate real package installs, systemd services, git clones, and OS-specific behavior.
+- Running workflow tests on a local developer machine risks destroying the environment.
+- Need reproducible, isolated acceptance tests with explicit expected outcomes.
+
+## Solution
+
+- Acceptance tests run only on a pre-built Debian 12 QEMU/KVM VM.
+- Test harness connects via SSH; never executes on the local host.
+- YAML scenarios define workflows and expected outcomes.
+- VM snapshot reverted before every scenario.
+- Multi-layer safety prevents local execution.
+
+## User Stories
+
+1. As a maintainer, I want install/remove Klipper workflow tested on a real VM, so I know the installer still works.
+2. As a maintainer, I want install/remove Moonraker workflow tested, so API stack compatibility is verified.
+3. As a maintainer, I want Mainsail/Fluidd install workflow tested, so web client setup works end-to-end.
+4. As a maintainer, I want backup/restore workflow tested, so user data survives the cycle.
+5. As a maintainer, I want tests parameterized by VM inventory, so future Ubuntu/Debian versions can be added without code changes.
+6. As a maintainer, I want local execution blocked by multiple guards, so the developer machine is never modified.
+7. As a CI operator, I want scenario results to show expected vs actual outcome, so failures are actionable.
+
+## Implementation Decisions
+
+- VM inventory config supplies host/IP, SSH user/key, OS family. No auto-provisioning; base images are prepared in advance.
+- Test harness: pytest + SSH fixture + Testinfra assertions. Commands are routed through SSH; assertions use Testinfra modules for service/file/package/port state.
+- Scenario schema YAML: `name`, `os`, `steps` (commands/options), `expected` (assertions for service running, file exists, package installed, port reachable, process present).
+- Snapshot reset: revert VM overlay before every scenario. Scenarios must be independent.
+- Safety guards: require `KIAUH_LIVE_TARGET_HOST`; abort if value is `localhost`, `127.*`, or matches current hostname; abort if target resolves to a local interface; verify SSH host key differs from local; optional explicit confirmation prompt.
+- Workflow priority: (1) install/remove Klipper; (2) install/remove Moonraker; (3) install Mainsail/Fluidd; (4) backup/restore; (5) update flows.
+- Test user on VM has passwordless sudo; VM has internet access; long installs use timeouts.
+- Scenario runner exposes expected outcome per step; failure shows command, expected assertion, and actual result.
+
+## Testing Decisions
+
+- Acceptance tests verify observable system state, not internal functions.
+- Each scenario defines exact pre-state (clean snapshot) and post-state assertions.
+- Flaky network commands are retried with timeout; failures attach relevant VM logs.
+- Live suite is marked with a `live` pytest marker and excluded from the default `pytest` run.
+
+## Out of Scope
+
+- Running live tests on the local machine or bare metal.
+- Auto-provisioning or building VM images.
+- Testing every extension in the first iteration; only core workflows.
+
+## Further Notes
+
+- Future OS matrix (Ubuntu 22.04/24.04) is enabled by adding inventory entries and matching base images; no harness changes needed.

docs/todo/TODO-006.md

@@ -0,0 +1,31 @@
+# PRD #2 - Task 1/5: VM inventory, SSH fixture, and safety guards
+
+**Tags:** `task`, `prd-2`
+
+## Parent PRD
+
+PRD #2: Isolated live-system acceptance test strategy on Debian 12 VM (`docs/prd/PRD-002-live-vm-test-strategy.md`)
+
+## What to build
+
+Create the VM inventory config schema (host/IP, SSH user/key, OS family). Implement the SSH connection fixture and the multi-layer safety guards that abort if the target could be the local machine. Write tests for the guards without executing any workflow.
+
+## Acceptance criteria
+
+- [ ] Inventory config schema documented and validated.
+- [ ] SSH fixture connects only when target is explicitly allowed.
+- [ ] Guards block `localhost`, `127.*`, current hostname, local interfaces, and unknown SSH host keys.
+- [ ] Guard tests run on the local machine and prove the blocks work.
+
+## Blocked by
+
+None — can start immediately.
+
+## Next task
+
+- TODO-007 (PRD #2 - Task 2/5: Scenario loader and Klipper install scenario)
+
+## User stories addressed
+
+- User story 5
+- User story 6

docs/todo/TODO-007.md

@@ -0,0 +1,32 @@
+# PRD #2 - Task 2/5: Scenario loader and Klipper install scenario
+
+**Tags:** `task`, `prd-2`
+
+## Parent PRD
+
+PRD #2: Isolated live-system acceptance test strategy on Debian 12 VM (`docs/prd/PRD-002-live-vm-test-strategy.md`)
+
+## What to build
+
+Implement the YAML scenario loader and the `live` pytest marker. Write the first end-to-end scenario: install Klipper on the Debian 12 VM, define expected outcomes (service file, env file, folders), and run it with snapshot revert before the scenario.
+
+## Acceptance criteria
+
+- [ ] YAML scenario loader parses `name`, `os`, `steps`, and `expected` assertions.
+- [ ] `pytest -m live` runs only live scenarios; default run skips them.
+- [ ] Klipper install scenario runs on the VM and passes.
+- [ ] Snapshot revert happens before the scenario.
+- [ ] Expected outcomes include file, service, and folder assertions.
+
+## Blocked by
+
+- TODO-006 (PRD #2 - Task 1/5: VM inventory, SSH fixture, and safety guards)
+
+## Next task
+
+- TODO-008 (PRD #2 - Task 3/5: Remove Klipper and Moonraker scenarios)
+
+## User stories addressed
+
+- User story 1
+- User story 7

docs/todo/TODO-008.md

@@ -0,0 +1,30 @@
+# PRD #2 - Task 3/5: Remove Klipper and Moonraker scenarios
+
+**Tags:** `task`, `prd-2`
+
+## Parent PRD
+
+PRD #2: Isolated live-system acceptance test strategy on Debian 12 VM (`docs/prd/PRD-002-live-vm-test-strategy.md`)
+
+## What to build
+
+Add remove-Klipper and install/remove-Moonraker scenarios. Each scenario starts from a clean snapshot. Capture relevant VM logs when a scenario fails to make debugging actionable.
+
+## Acceptance criteria
+
+- [ ] Remove Klipper scenario runs and verifies service/files are gone.
+- [ ] Install Moonraker scenario runs and verifies service/config/log files.
+- [ ] Remove Moonraker scenario runs and verifies cleanup.
+- [ ] Failure output includes tail of installer/service logs.
+
+## Blocked by
+
+- TODO-007 (PRD #2 - Task 2/5: Scenario loader and Klipper install scenario)
+
+## Next task
+
+- TODO-009 (PRD #2 - Task 4/5: Mainsail and Fluidd install scenarios)
+
+## User stories addressed
+
+- User story 2

docs/todo/TODO-009.md

@@ -0,0 +1,30 @@
+# PRD #2 - Task 4/5: Mainsail and Fluidd install scenarios
+
+**Tags:** `task`, `prd-2`
+
+## Parent PRD
+
+PRD #2: Isolated live-system acceptance test strategy on Debian 12 VM (`docs/prd/PRD-002-live-vm-test-strategy.md`)
+
+## What to build
+
+Add install scenarios for Mainsail and Fluidd web clients. Assert that the static files are deployed and the reverse-proxy/service config is in place. Harden the snapshot-revert fixture so it runs reliably before every scenario.
+
+## Acceptance criteria
+
+- [ ] Mainsail install scenario passes on the VM.
+- [ ] Fluidd install scenario passes on the VM.
+- [ ] Expected outcomes check webroot directory and reverse-proxy config.
+- [ ] Snapshot revert fixture is robust (wait for SSH, error on revert failure).
+
+## Blocked by
+
+- TODO-008 (PRD #2 - Task 3/5: Remove Klipper and Moonraker scenarios)
+
+## Next task
+
+- TODO-010 (PRD #2 - Task 5/5: Backup, restore, and update scenarios)
+
+## User stories addressed
+
+- User story 3

docs/todo/TODO-010.md

@@ -0,0 +1,32 @@
+# PRD #2 - Task 5/5: Backup, restore, and update scenarios
+
+**Tags:** `task`, `prd-2`
+
+## Parent PRD
+
+PRD #2: Isolated live-system acceptance test strategy on Debian 12 VM (`docs/prd/PRD-002-live-vm-test-strategy.md`)
+
+## What to build
+
+Add backup/restore and update scenarios. Document the live test runbook: how to prepare the VM, set inventory, run scenarios, and read results. Ensure the whole live suite can be executed in one command.
+
+## Acceptance criteria
+
+- [ ] Backup scenario creates an archive with expected content.
+- [ ] Restore scenario returns config files to expected state.
+- [ ] Update scenario changes a component version/config observable on the VM.
+- [ ] Runbook `docs/live-testing.md` covers VM setup, inventory, execution, and troubleshooting.
+- [ ] Full `pytest -m live` run completes end-to-end.
+
+## Blocked by
+
+- TODO-009 (PRD #2 - Task 4/5: Mainsail and Fluidd install scenarios)
+
+## Next task
+
+None — this is the last task.
+
+## User stories addressed
+
+- User story 4
+- User story 7

kiauh/live/guards.py

@@ -0,0 +1,93 @@
+# ======================================================================= #
+#  Copyright (C) 2020 - 2026 Dominik Willner <th33xitus@gmail.com>        #
+#                                                                         #
+#  This file is part of KIAUH - Klipper Installation And Update Helper    #
+#  https://github.com/dw-0/kiauh                                          #
+#                                                                         #
+#  This file may be distributed under the terms of the GNU GPLv3 license  #
+# ======================================================================= #
+from __future__ import annotations
+
+import os
+import socket
+from ipaddress import ip_address
+from typing import List, Set
+
+
+class UnsafeTargetError(Exception):
+    """Raised when a live test target is considered unsafe."""
+
+
+LOCAL_HOSTNAMES = {"localhost", "localhost.localdomain"}
+LOCAL_ADDRESSES = {"127.0.0.1", "::1"}
+
+
+def _local_interface_ips() -> Set[str]:
+    """Return all IP addresses assigned to local network interfaces."""
+    ips: Set[str] = set()
+    try:
+        hostname = socket.gethostname()
+        infos = socket.getaddrinfo(hostname, None)
+        for info in infos:
+            addr = info[4][0]
+            ips.add(addr)
+    except socket.gaierror:
+        pass
+    return ips
+
+
+def _resolve(host: str) -> List[str]:
+    """Resolve a hostname to its IP addresses."""
+    try:
+        infos = socket.getaddrinfo(host, None)
+        return [info[4][0] for info in infos]
+    except socket.gaierror:
+        return []
+
+
+def is_local_host(host: str) -> bool:
+    """Return True if the host refers to the local machine."""
+    host_lower = host.lower().strip()
+
+    if host_lower in LOCAL_HOSTNAMES:
+        return True
+
+    if host_lower in LOCAL_ADDRESSES:
+        return True
+
+    if host_lower == socket.gethostname().lower():
+        return True
+
+    resolved = _resolve(host_lower)
+    local_ips = _local_interface_ips() | LOCAL_ADDRESSES
+    for addr in resolved:
+        if addr in local_ips:
+            return True
+        try:
+            if ip_address(addr).is_loopback:
+                return True
+        except ValueError:
+            pass
+
+    return False
+
+
+def assert_safe_to_run(vm) -> None:
+    """Multi-layer safety check before running live tests against a VM."""
+    if os.environ.get("KIAUH_LIVE_ALLOW") != "1":
+        raise UnsafeTargetError(
+            "Live tests disabled. Set KIAUH_LIVE_ALLOW=1 to enable."
+        )
+
+    if not vm.host:
+        raise UnsafeTargetError("VM host is empty")
+
+    if is_local_host(vm.host):
+        raise UnsafeTargetError(
+            f"Refusing to run live tests against local host: {vm.host}"
+        )
+
+    if os.environ.get("KIAUH_LIVE_TARGET_HOST") != vm.host:
+        raise UnsafeTargetError(
+            "KIAUH_LIVE_TARGET_HOST must match the selected VM host"
+        )

kiauh/live/inventory.py

@@ -0,0 +1,90 @@
+# ======================================================================= #
+#  Copyright (C) 2020 - 2026 Dominik Willner <th33xitus@gmail.com>        #
+#                                                                         #
+#  This file is part of KIAUH - Klipper Installation And Update Helper    #
+#  https://github.com/dw-0/kiauh                                          #
+#                                                                         #
+#  This file may be distributed under the terms of the GNU GPLv3 license  #
+# ======================================================================= #
+from __future__ import annotations
+
+import os
+from dataclasses import dataclass
+from pathlib import Path
+from typing import List
+
+import yaml
+
+try:
+    from dotenv import load_dotenv
+
+    load_dotenv(Path(__file__).parents[2].joinpath(".env"))
+except ImportError:
+    load_dotenv = None
+
+DEFAULT_INVENTORY_PATH = Path(__file__).parent.joinpath("inventory.yaml")
+
+
+@dataclass
+class VM:
+    name: str
+    host: str
+    user: str
+    key_file: str
+    os: str
+    domain: str | None = None
+    snapshot: str | None = None
+
+
+class InventoryError(Exception):
+    pass
+
+
+def _env_var_name(vm_name: str, suffix: str) -> str:
+    safe_name = vm_name.replace("-", "_").upper()
+    return f"KIAUH_LIVE_{safe_name}_{suffix.upper()}"
+
+
+def load_inventory(path: Path | None = None) -> List[VM]:
+    inventory_path = Path(path or os.environ.get("KIAUH_LIVE_INVENTORY", DEFAULT_INVENTORY_PATH))
+    if not inventory_path.exists():
+        raise InventoryError(f"Inventory file not found: {inventory_path}")
+
+    data = yaml.safe_load(inventory_path.read_text())
+    if not data or "vms" not in data:
+        raise InventoryError("Inventory must contain a 'vms' list")
+
+    vms = []
+    for item in data["vms"]:
+        for required in ("name", "user", "os"):
+            if required not in item:
+                raise InventoryError(f"VM '{item.get('name', '?')}' missing '{required}'")
+
+        name = item["name"]
+        host = os.environ.get(_env_var_name(name, "host"), item.get("host"))
+        key_file = os.environ.get(_env_var_name(name, "key_file"), item.get("key_file"))
+
+        if not host:
+            raise InventoryError(f"VM '{name}' missing 'host' (inventory or env)")
+        if not key_file:
+            raise InventoryError(f"VM '{name}' missing 'key_file' (inventory or env)")
+
+        vms.append(
+            VM(
+                name=name,
+                host=host,
+                user=item["user"],
+                key_file=key_file,
+                os=item["os"],
+                domain=item.get("domain"),
+                snapshot=item.get("snapshot"),
+            )
+        )
+    return vms
+
+
+def get_vm(name: str, path: Path | None = None) -> VM:
+    for vm in load_inventory(path):
+        if vm.name == name:
+            return vm
+    raise InventoryError(f"VM '{name}' not found in inventory")

kiauh/live/inventory.yaml

@@ -0,0 +1,11 @@
+vms:
+  - name: debian12-kiauh
+    # Override host/key_file via environment variables or a .env file.
+    # KIAUH_LIVE_DEBIAN12_KIAUH_HOST
+    # KIAUH_LIVE_DEBIAN12_KIAUH_KEY_FILE
+    host: null
+    user: kiauh
+    key_file: null
+    os: debian-12
+    domain: debian12-kiauh
+    snapshot: clean

kiauh/live/runner.py

@@ -0,0 +1,93 @@
+# ======================================================================= #
+#  Copyright (C) 2020 - 2026 Dominik Willner <th33xitus@gmail.com>        #
+#                                                                         #
+#  This file is part of KIAUH - Klipper Installation And Update Helper    #
+#  https://github.com/dw-0/kiauh                                          #
+#                                                                         #
+#  This file may be distributed under the terms of the GNU GPLv3 license  #
+# ======================================================================= #
+from __future__ import annotations
+
+import subprocess
+from pathlib import Path
+from typing import List
+
+import paramiko
+import testinfra
+
+from live.guards import assert_safe_to_run, is_local_host
+from live.inventory import VM
+
+
+class LiveRunnerError(Exception):
+    pass
+
+
+class LiveRunner:
+    """SSH runner for live VM tests."""
+
+    def __init__(self, vm: VM) -> None:
+        assert_safe_to_run(vm)
+        self.vm = vm
+        self._ssh: paramiko.SSHClient | None = None
+
+    def connect(self) -> paramiko.SSHClient:
+        if self._ssh is not None:
+            return self._ssh
+
+        client = paramiko.SSHClient()
+        client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
+        key_file = Path(self.vm.key_file).expanduser()
+        client.connect(
+            self.vm.host,
+            username=self.vm.user,
+            key_filename=str(key_file),
+            look_for_keys=False,
+            timeout=30,
+        )
+        self._ssh = client
+        return client
+
+    def run(self, command: List[str], timeout: int = 120) -> subprocess.CompletedProcess:
+        client = self.connect()
+        cmd = " ".join(command) if isinstance(command, list) else command
+        stdin, stdout, stderr = client.exec_command(cmd, timeout=timeout)
+        rc = stdout.channel.recv_exit_status()
+        return subprocess.CompletedProcess(
+            args=command,
+            returncode=rc,
+            stdout=stdout.read().decode("utf-8", errors="replace"),
+            stderr=stderr.read().decode("utf-8", errors="replace"),
+        )
+
+    def get_host(self) -> testinfra.host.Host:
+        """Return a Testinfra host for assertions."""
+        key_file = Path(self.vm.key_file).expanduser()
+        return testinfra.get_host(
+            f"paramiko://{self.vm.user}@{self.vm.host}",
+            ssh_identity_file=str(key_file),
+        )
+
+    def close(self) -> None:
+        if self._ssh is not None:
+            self._ssh.close()
+            self._ssh = None
+
+
+def revert_vm_snapshot(vm: VM) -> None:
+    """Revert a VM to the configured snapshot before a scenario."""
+    if not vm.domain or not vm.snapshot:
+        raise LiveRunnerError("VM inventory missing domain or snapshot")
+
+    if is_local_host(vm.host):
+        raise LiveRunnerError("Refusing to revert a local VM snapshot")
+
+    result = subprocess.run(
+        ["virsh", "snapshot-revert", vm.domain, vm.snapshot, "--running"],
+        capture_output=True,
+        text=True,
+    )
+    if result.returncode != 0:
+        raise LiveRunnerError(
+            f"Failed to revert snapshot: {result.stderr.strip() or result.stdout.strip()}"
+        )

kiauh/live/scenarios.py

@@ -0,0 +1,85 @@
+# ======================================================================= #
+#  Copyright (C) 2020 - 2026 Dominik Willner <th33xitus@gmail.com>        #
+#                                                                         #
+#  This file is part of KIAUH - Klipper Installation And Update Helper    #
+#  https://github.com/dw-0/kiauh                                          #
+#                                                                         #
+#  This file may be distributed under the terms of the GNU GPLv3 license  #
+# ======================================================================= #
+from __future__ import annotations
+
+from pathlib import Path
+from typing import Any, Dict, List
+
+import yaml
+
+SCENARIOS_DIR = Path(__file__).parent.joinpath("scenarios")
+
+
+class ScenarioError(Exception):
+    pass
+
+
+def load_scenarios(directory: Path | None = None) -> List[Dict[str, Any]]:
+    path = Path(directory or SCENARIOS_DIR)
+    if not path.exists():
+        return []
+
+    scenarios = []
+    for file in sorted(path.glob("*.yaml")):
+        data = yaml.safe_load(file.read_text())
+        if not isinstance(data, dict):
+            raise ScenarioError(f"Scenario {file.name} is not a mapping")
+        data.setdefault("file", str(file))
+        scenarios.append(data)
+    return scenarios
+
+
+def assert_expected(host, expected: List[Dict[str, Any]]) -> None:
+    """Evaluate expected outcomes using Testinfra assertions."""
+    failures = []
+
+    for item in expected:
+        try:
+            _assert_item(host, item)
+        except AssertionError as e:
+            failures.append(f"{item}: {e}")
+
+    if failures:
+        raise AssertionError("\n".join(failures))
+
+
+def _assert_item(host, item: Dict[str, Any]) -> None:
+    assertion_type = item.get("type")
+
+    if assertion_type == "service":
+        service = host.service(item["name"])
+        state = item.get("state")
+        if state == "running":
+            assert service.is_running, f"service {item['name']} is not running"
+        elif state == "enabled":
+            assert service.is_enabled, f"service {item['name']} is not enabled"
+
+    elif assertion_type == "file":
+        file = host.file(item["path"])
+        if item.get("exists", True):
+            assert file.exists, f"file {item['path']} does not exist"
+        else:
+            assert not file.exists, f"file {item['path']} should not exist"
+
+    elif assertion_type == "package":
+        pkg = host.package(item["name"])
+        assert pkg.is_installed, f"package {item['name']} is not installed"
+
+    elif assertion_type == "port":
+        socket = host.socket(f"tcp://{item['address']}:{item['port']}")
+        assert socket.is_listening, f"port {item['port']} is not listening"
+
+    elif assertion_type == "command":
+        result = host.run(item["command"])
+        assert result.rc == item.get("returncode", 0), (
+            f"command {item['command']} returned {result.rc}: {result.stderr}"
+        )
+
+    else:
+        raise ScenarioError(f"Unknown assertion type: {assertion_type}")

kiauh/live/scenarios/backup_restore.yaml

@@ -0,0 +1,24 @@
+name: Backup and restore Klipper data on Debian 12
+vm: debian12-kiauh
+os: debian-12
+steps:
+  - command: ["kiauh", "install", "klipper", "--count", "1"]
+    timeout: 600
+  - command: ["mkdir", "-p", "/home/kiauh/backups"]
+    timeout: 10
+  - command: ["tar", "-czf", "/home/kiauh/backups/printer_data.tar.gz", "-C", "/home/kiauh", "printer_data"]
+    timeout: 60
+  - command: ["rm", "-rf", "/home/kiauh/printer_data"]
+    timeout: 30
+  - command: ["tar", "-xzf", "/home/kiauh/backups/printer_data.tar.gz", "-C", "/home/kiauh"]
+    timeout: 60
+expected:
+  - type: file
+    path: /home/kiauh/backups/printer_data.tar.gz
+    exists: true
+  - type: file
+    path: /home/kiauh/printer_data
+    exists: true
+  - type: service
+    name: klipper.service
+    state: running

kiauh/live/scenarios/klipper_install.yaml

@@ -0,0 +1,18 @@
+name: Install Klipper on Debian 12
+vm: debian12-kiauh
+os: debian-12
+steps:
+  - command: ["kiauh", "install", "klipper", "--count", "1"]
+    timeout: 600
+  - command: ["sudo", "systemctl", "is-active", "klipper.service"]
+    timeout: 30
+expected:
+  - type: service
+    name: klipper.service
+    state: running
+  - type: file
+    path: /home/kiauh/klipper/klipper
+    exists: true
+  - type: command
+    command: systemctl is-enabled klipper.service
+    returncode: 0

kiauh/live/scenarios/klipper_remove.yaml

@@ -0,0 +1,18 @@
+name: Remove Klipper on Debian 12
+vm: debian12-kiauh
+os: debian-12
+steps:
+  - command: ["kiauh", "install", "klipper", "--count", "1"]
+    timeout: 600
+  - command: ["kiauh", "remove", "klipper", "--service", "--dir", "--env"]
+    timeout: 120
+expected:
+  - type: file
+    path: /etc/systemd/system/klipper.service
+    exists: false
+  - type: file
+    path: /home/kiauh/klipper
+    exists: false
+  - type: file
+    path: /home/kiauh/klippy-env
+    exists: false

kiauh/live/scenarios/klipper_update.yaml

@@ -0,0 +1,15 @@
+name: Update Klipper on Debian 12
+vm: debian12-kiauh
+os: debian-12
+steps:
+  - command: ["kiauh", "install", "klipper", "--count", "1"]
+    timeout: 600
+  - command: ["kiauh", "update", "klipper"]
+    timeout: 300
+expected:
+  - type: service
+    name: klipper.service
+    state: running
+  - type: command
+    command: test -d /home/kiauh/klipper/.git
+    returncode: 0

kiauh/live/scenarios/moonraker_install.yaml

@@ -0,0 +1,15 @@
+name: Install Moonraker on Debian 12
+vm: debian12-kiauh
+os: debian-12
+steps:
+  - command: ["kiauh", "install", "klipper", "--count", "1"]
+    timeout: 600
+  - command: ["kiauh", "install", "moonraker"]
+    timeout: 600
+expected:
+  - type: service
+    name: moonraker.service
+    state: running
+  - type: file
+    path: /home/kiauh/moonraker/moonraker
+    exists: true

kiauh/live/scenarios/moonraker_remove.yaml

@@ -0,0 +1,17 @@
+name: Remove Moonraker on Debian 12
+vm: debian12-kiauh
+os: debian-12
+steps:
+  - command: ["kiauh", "install", "klipper", "--count", "1"]
+    timeout: 600
+  - command: ["kiauh", "install", "moonraker"]
+    timeout: 600
+  - command: ["kiauh", "remove", "moonraker", "--service", "--dir", "--env"]
+    timeout: 120
+expected:
+  - type: file
+    path: /etc/systemd/system/moonraker.service
+    exists: false
+  - type: file
+    path: /home/kiauh/moonraker
+    exists: false

kiauh/live/tests/conftest.py

@@ -0,0 +1,74 @@
+from __future__ import annotations
+
+import os
+import sys
+from pathlib import Path
+from typing import Any, Dict, Generator, List
+
+import pytest
+
+PROJECT_ROOT = Path(__file__).resolve().parents[3]
+if str(PROJECT_ROOT) not in sys.path:
+    sys.path.insert(0, str(PROJECT_ROOT))
+
+from live.inventory import VM, get_vm
+from live.runner import LiveRunner, revert_vm_snapshot
+from live.scenarios import load_scenarios
+
+
+@pytest.fixture(autouse=True)
+def silence_logger(monkeypatch: pytest.MonkeyPatch) -> None:
+    for name in (
+        "print_info",
+        "print_ok",
+        "print_warn",
+        "print_error",
+        "print_status",
+        "print_dialog",
+    ):
+        monkeypatch.setattr(f"core.logger.Logger.{name}", lambda *a, **k: None)
+
+
+def pytest_configure(config: pytest.Config) -> None:
+    config.addinivalue_line(
+        "markers",
+        "live: tests that run against a real VM (isolated, never local)",
+    )
+
+
+def pytest_collection_modifyitems(config: pytest.Config, items: List[pytest.Item]) -> None:
+    skip_live = pytest.mark.skip(reason="live tests skipped by default; use -m live")
+    for item in items:
+        if item.get_closest_marker("live") and not config.getoption("-m"):
+            item.add_marker(skip_live)
+
+
+@pytest.fixture(scope="session")
+def live_vm() -> VM:
+    """Single VM used for live tests."""
+    vm_name = os.environ.get("KIAUH_LIVE_VM", "debian12-kiauh")
+    return get_vm(vm_name)
+
+
+@pytest.fixture(scope="function")
+def live_runner(live_vm: VM) -> Generator[LiveRunner, None, None]:
+    runner = LiveRunner(live_vm)
+    yield runner
+    runner.close()
+
+
+@pytest.fixture(scope="function")
+def fresh_vm(live_vm: VM) -> VM:
+    """Revert the VM to its clean snapshot before each scenario."""
+    revert_vm_snapshot(live_vm)
+    return live_vm
+
+
+def pytest_generate_tests(metafunc: pytest.Metafunc) -> None:
+    if "scenario" in metafunc.fixturenames:
+        scenarios = load_scenarios()
+        metafunc.parametrize(
+            "scenario",
+            scenarios,
+            ids=[s.get("name", s.get("file", "unknown")) for s in scenarios],
+        )

kiauh/live/tests/test_guards.py

@@ -0,0 +1,55 @@
+from __future__ import annotations
+
+import socket
+
+import pytest
+from live.guards import UnsafeTargetError, assert_safe_to_run, is_local_host
+from live.inventory import VM
+
+
+class TestIsLocalHost:
+    def test_localhost_is_local(self) -> None:
+        assert is_local_host("localhost") is True
+
+    def test_127_is_local(self) -> None:
+        assert is_local_host("127.0.0.1") is True
+        assert is_local_host("::1") is True
+
+    def test_current_hostname_is_local(self) -> None:
+        assert is_local_host(socket.gethostname()) is True
+
+    def test_remote_host_is_not_local(self) -> None:
+        assert is_local_host("192.168.122.10") is False
+        assert is_local_host("example.com") is False
+
+
+class TestAssertSafeToRun:
+    def test_requires_live_allow_flag(self, monkeypatch: pytest.MonkeyPatch) -> None:
+        monkeypatch.delenv("KIAUH_LIVE_ALLOW", raising=False)
+        vm = VM(name="vm", host="192.168.122.10", user="u", key_file="k", os="debian-12")
+
+        with pytest.raises(UnsafeTargetError, match="disabled"):
+            assert_safe_to_run(vm)
+
+    def test_blocks_localhost(self, monkeypatch: pytest.MonkeyPatch) -> None:
+        monkeypatch.setenv("KIAUH_LIVE_ALLOW", "1")
+        monkeypatch.setenv("KIAUH_LIVE_TARGET_HOST", "localhost")
+        vm = VM(name="vm", host="localhost", user="u", key_file="k", os="debian-12")
+
+        with pytest.raises(UnsafeTargetError, match="local host"):
+            assert_safe_to_run(vm)
+
+    def test_requires_matching_target_host(self, monkeypatch: pytest.MonkeyPatch) -> None:
+        monkeypatch.setenv("KIAUH_LIVE_ALLOW", "1")
+        monkeypatch.setenv("KIAUH_LIVE_TARGET_HOST", "192.168.122.10")
+        vm = VM(name="vm", host="192.168.122.11", user="u", key_file="k", os="debian-12")
+
+        with pytest.raises(UnsafeTargetError, match="must match"):
+            assert_safe_to_run(vm)
+
+    def test_passes_for_safe_remote_host(self, monkeypatch: pytest.MonkeyPatch) -> None:
+        monkeypatch.setenv("KIAUH_LIVE_ALLOW", "1")
+        monkeypatch.setenv("KIAUH_LIVE_TARGET_HOST", "192.168.122.10")
+        vm = VM(name="vm", host="192.168.122.10", user="u", key_file="k", os="debian-12")
+
+        assert assert_safe_to_run(vm) is None

kiauh/live/tests/test_inventory.py

@@ -0,0 +1,54 @@
+from __future__ import annotations
+
+from pathlib import Path
+
+import pytest
+from live.inventory import InventoryError, VM, load_inventory
+
+
+class TestLoadInventory:
+    def test_loads_vm_from_inventory(self, tmp_path: Path) -> None:
+        inv = tmp_path / "inv.yaml"
+        inv.write_text(
+            "vms:\n"
+            "  - name: debian12\n"
+            "    host: 10.0.0.5\n"
+            "    user: kiauh\n"
+            "    key_file: /key\n"
+            "    os: debian-12\n"
+        )
+        vms = load_inventory(inv)
+        assert len(vms) == 1
+        assert vms[0].host == "10.0.0.5"
+        assert vms[0].key_file == "/key"
+
+    def test_env_overrides_host_and_key_file(
+        self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+    ) -> None:
+        monkeypatch.setenv("KIAUH_LIVE_DEBIAN12_HOST", "192.168.1.50")
+        monkeypatch.setenv("KIAUH_LIVE_DEBIAN12_KEY_FILE", "/secret/key")
+
+        inv = tmp_path / "inv.yaml"
+        inv.write_text(
+            "vms:\n"
+            "  - name: debian12\n"
+            "    host: 10.0.0.5\n"
+            "    user: kiauh\n"
+            "    key_file: /key\n"
+            "    os: debian-12\n"
+        )
+        vms = load_inventory(inv)
+        assert vms[0].host == "192.168.1.50"
+        assert vms[0].key_file == "/secret/key"
+
+    def test_missing_host_raises(self, tmp_path: Path) -> None:
+        inv = tmp_path / "inv.yaml"
+        inv.write_text(
+            "vms:\n"
+            "  - name: debian12\n"
+            "    user: kiauh\n"
+            "    key_file: /key\n"
+            "    os: debian-12\n"
+        )
+        with pytest.raises(InventoryError, match="missing 'host'"):
+            load_inventory(inv)

kiauh/live/tests/test_scenarios.py

@@ -0,0 +1,27 @@
+from __future__ import annotations
+
+from typing import Any, Dict, List
+
+import pytest
+
+from live.inventory import VM
+from live.runner import LiveRunner
+from live.scenarios import assert_expected
+
+
+@pytest.mark.live
+def test_scenario(fresh_vm: VM, live_runner: LiveRunner, scenario: Dict[str, Any]) -> None:
+    """Run a live scenario against the VM and verify expected outcomes."""
+    steps: List[Dict[str, Any]] = scenario.get("steps", [])
+    expected = scenario.get("expected", [])
+
+    for step in steps:
+        command = step["command"]
+        timeout = step.get("timeout", 120)
+        result = live_runner.run(command, timeout=timeout)
+        assert result.returncode == 0, (
+            f"Step failed: {' '.join(command)}\nstdout: {result.stdout}\nstderr: {result.stderr}"
+        )
+
+    host = live_runner.get_host()
+    assert_expected(host, expected)

pyproject.toml

@@ -2,7 +2,7 @@
 requires-python = ">=3.8"
 
 [project.optional-dependencies]
-dev=["ruff", "mypy", "pytest"]
+dev=["ruff", "mypy", "pytest", "paramiko", "pytest-testinfra", "pyyaml", "python-dotenv"]
 
 [tool.ruff]
 required-version = ">=0.9.10"
@@ -33,5 +33,12 @@ warn_unreachable = true
 
 [tool.pytest.ini_options]
 minversion = "8.2.1"
-testpaths = ["kiauh/core/simple_config_parser/tests", "kiauh/utils/tests"]
+testpaths = [
+    "kiauh/core/simple_config_parser/tests",
+    "kiauh/live/tests",
+    "kiauh/utils/tests",
+]
 pythonpath = ["kiauh"]
+markers = [
+    "live: tests that run against a real VM (isolated, never local)",
+]

requirements-dev.txt

@@ -1,3 +1,7 @@
 ruff (>=0.9.10)
 mypy
 pytest
+paramiko
+pytest-testinfra
+pyyaml
+python-dotenv