From 2b4703e93221c877b1f5f7b3cae567a7099319ba Mon Sep 17 00:00:00 2001
From: Guilherme Costa <guilherme.costa@blockstec.com>
Date: Mon, 19 Oct 2026 14:12:46 +0100
Subject: [PATCH] test(updater): add hermetic end-to-end suite for the update
 and deploy-hold flows

---
 tests/integration/updater/conftest.py         | 484 ++++++++++++++++++
 .../updater/test_updater_daemon_flow.py       | 197 +++++++
 .../updater/test_updater_git_flow.py          | 333 ++++++++++++
 3 files changed, 1014 insertions(+)
 create mode 100644 tests/integration/updater/conftest.py
 create mode 100644 tests/integration/updater/test_updater_daemon_flow.py
 create mode 100644 tests/integration/updater/test_updater_git_flow.py

diff --git a/tests/integration/updater/conftest.py b/tests/integration/updater/conftest.py
new file mode 100644
index 00000000..d8d4b822
--- /dev/null
+++ b/tests/integration/updater/conftest.py
@@ -0,0 +1,484 @@
+"""Updater integration harness: real git, state files, locking and D-Bus layer.
+
+Only the root-owned edges are faked (systemctl, sudo, apt, dpkg, pip, klippy.sock).
+Remotes are local bare repos reached through an https:// insteadOf rewrite, so
+clone/fetch/ls-remote run for real against URLs that pass the https-only checks.
+"""
+
+import asyncio
+import json
+import os
+import shutil
+import subprocess
+import sys
+from collections.abc import Mapping
+from pathlib import Path
+from unittest.mock import AsyncMock, MagicMock
+
+import pytest
+
+REPO_ROOT = Path(__file__).resolve().parents[3]
+REMOTE_BASE = "https://git.test/"
+
+
+class _FakeDbusBase:
+    """Real Python base class so inheritance does not raise TypeError."""
+
+    def __init__(self, *a, **kw) -> None:
+        pass
+
+    def __init_subclass__(cls, **kw) -> None:
+        super().__init_subclass__()
+
+
+@pytest.fixture(scope="module", autouse=True)
+def mock_sdbus():
+    """Swap sdbus for a stub and force a fresh updater.dbus_service import."""
+    mock = MagicMock()
+    mock.DbusInterfaceCommonAsync = _FakeDbusBase
+    mock.dbus_signal_async = lambda *a, **kw: lambda fn: fn
+    mock.dbus_method_async = lambda *a, **kw: lambda fn: fn
+    with pytest.MonkeyPatch.context() as mp:
+        for key in ("sdbus", "updater", "updater.dbus_service"):
+            mp.delitem(sys.modules, key, raising=False)
+        mp.setitem(sys.modules, "sdbus", mock)
+        yield mock
+
+
+def _script(path: Path, body: str) -> Path:
+    """Write an executable bash script."""
+    path.write_text("#!/bin/bash\n" + body)
+    path.chmod(0o755)
+    return path
+
+
+_SYSTEMCTL = """\
+echo "systemctl $*" >>"{state}/calls.log"
+unit="${{!#}}"
+case "$1" in
+  is-active) [ -e "{state}/inactive-$unit" ] && exit 3; exit 0 ;;
+  show)
+    if [ "$2" = "--property=LoadState" ]; then
+      if [ -e "{state}/missing-$unit" ]; then echo not-found; else echo loaded; fi
+    elif [[ " $* " == *" ActiveState "* ]]; then
+      for u in "${{@:2}}"; do
+        case "$u" in -*|ActiveState) continue ;; esac
+        if [ -e "{state}/inactive-$u" ]; then echo inactive; else echo active; fi
+        echo
+      done
+    else
+      echo 0
+    fi
+    exit 0 ;;
+  restart)
+    if [ -e "{state}/fail-restart-$unit" ]; then
+      echo "Job for $unit failed" >&2
+      exit 1
+    fi
+    exit 0 ;;
+  *) exit 0 ;;
+esac
+"""
+
+# Only our fakes run as "root": a hook's bare `sudo cp` into /etc is denied, as on a box.
+_SUDO = """\
+echo "sudo $*" >>"{state}/calls.log"
+case "$1" in
+  "{bin}"/*) exec "$@" ;;
+  *) echo "sudo: not allowed" >&2; exit 1 ;;
+esac
+"""
+
+_APT = """\
+echo "apt $*" >>"{state}/calls.log"
+if [ "$1" = "list" ]; then
+  echo "Listing..."
+  [ -e "{state}/upgradable" ] && cat "{state}/upgradable"
+fi
+exit 0
+"""
+
+_LOGGED = 'echo "{name} $*" >>"{state}/calls.log"\nexit 0\n'
+
+
+class Signal:
+    """Stand-in for an sdbus signal: records every emit in a shared log."""
+
+    def __init__(self, name: str, log: list) -> None:
+        self._name = name
+        self._log = log
+
+    def emit(self, args: tuple) -> None:
+        """Record (signal name, args)."""
+        self._log.append((self._name, args))
+
+
+class Recorder:
+    """ProgressCallback that keeps every call, in order."""
+
+    def __init__(self) -> None:
+        self.events: list[tuple] = []
+
+    def on_step(self, name: str, step: int, total: int) -> None:
+        """Record a progress step."""
+        self.events.append(("on_step", name, step, total))
+
+    def on_component_done(self, name: str, success: bool) -> None:
+        """Record a component outcome."""
+        self.events.append(("on_component_done", name, success))
+
+    def on_error(self, name: str, reason: str) -> None:
+        """Record an error."""
+        self.events.append(("on_error", name, reason))
+
+    def on_rollback(self, name: str, success: bool) -> None:
+        """Record a rollback."""
+        self.events.append(("on_rollback", name, success))
+
+    def on_recover(self, name: str, success: bool) -> None:
+        """Record a recover."""
+        self.events.append(("on_recover", name, success))
+
+    def of(self, method: str) -> list[tuple]:
+        """Arguments of every call to one callback method."""
+        return [e[1:] for e in self.events if e[0] == method]
+
+
+class Remote:
+    """A bare origin plus a scratch worktree used to author upstream commits."""
+
+    def __init__(self, harness: "Harness", name: str) -> None:
+        self.name = name
+        self.url = f"{REMOTE_BASE}{name}.git"
+        self._h = harness
+        self.bare = harness.remotes / f"{name}.git"
+        self.work = harness.root / "authoring" / name
+        harness.git("init", "-q", "--bare", "--initial-branch=master", str(self.bare))
+        harness.git("init", "-q", "--initial-branch=master", str(self.work))
+        harness.git("remote", "add", "origin", str(self.bare), cwd=self.work)
+
+    def commit(
+        self,
+        files: Mapping[str, str | None],
+        msg: str = "change",
+        branch: str = "master",
+    ) -> str:
+        """Write (or delete, for None) files, commit on branch, push; return the sha."""
+        h = self._h
+        # symbolic-ref works on the unborn initial branch, where switch would fail.
+        if h.git("symbolic-ref", "--short", "HEAD", cwd=self.work) != branch:
+            known = h.git("branch", "--format=%(refname:short)", cwd=self.work)
+            create = [] if branch in known.splitlines() else ["-c"]
+            h.git("switch", "-q", *create, branch, cwd=self.work)
+        for rel, content in files.items():
+            target = self.work / rel
+            if content is None:
+                target.unlink()
+                continue
+            target.parent.mkdir(parents=True, exist_ok=True)
+            target.write_text(content)
+        h.git("add", "-A", cwd=self.work)
+        h.git("commit", "-q", "-m", msg, cwd=self.work)
+        h.git("push", "-q", "origin", f"HEAD:refs/heads/{branch}", cwd=self.work)
+        return h.git("rev-parse", "HEAD", cwd=self.work)
+
+
+class Harness:
+    """One isolated device: tmp HOME, fake root tools, origins and components."""
+
+    def __init__(self, root: Path, home: Path, mp: pytest.MonkeyPatch) -> None:
+        self.root = root
+        self.home = home
+        self.remotes = root / "remotes"
+        self.state = root / "fakeroot"
+        self.bin = root / "bin"
+        self.hooks = root / "hooks"
+        self.run_dir = root / "run"
+        self.etc = root / "etc"
+        for d in (self.remotes, self.state, self.bin, self.hooks, self.etc):
+            d.mkdir(parents=True)
+        self.run_dir.mkdir(mode=0o700)
+        self.components: list[dict] = []
+        self.printing = AsyncMock(return_value=False)
+        self.apt_upgrade = AsyncMock()
+        self.dbus_g: dict = {}
+        self.mp = mp
+
+    # -- git -----------------------------------------------------------------
+
+    def git(self, *args: str, cwd: Path | None = None) -> str:
+        """Run git as the test author; return stripped stdout."""
+        out = subprocess.run(
+            ["git", *args], cwd=cwd, check=True, capture_output=True, text=True
+        )
+        return out.stdout.strip()
+
+    def remote(self, name: str, files: dict[str, str] | None = None) -> Remote:
+        """Create an origin with one initial commit."""
+        r = Remote(self, name)
+        r.commit(files or {"README": f"{name}\n"}, msg="initial")
+        return r
+
+    def install(self, remote: Remote, branch: str = "master") -> Path:
+        """Clone the origin into HOME over its https URL, as the installer does."""
+        path = self.home / remote.name
+        self.git("clone", "-q", "--branch", branch, remote.url, str(path))
+        return path
+
+    def head(self, name: str) -> str:
+        """HEAD sha of the installed component."""
+        return self.git("rev-parse", "HEAD", cwd=self.home / name)
+
+    def branch(self, name: str) -> str:
+        """Checked-out branch of the installed component."""
+        return self.git("rev-parse", "--abbrev-ref", "HEAD", cwd=self.home / name)
+
+    # -- components.yaml -----------------------------------------------------
+
+    def component(self, name: str, **fields: object) -> dict:
+        """Register a git component pointing at HOME/<name> and its origin."""
+        entry = {
+            "name": name,
+            "type": "git",
+            "path": str(self.home / name),
+            "url": f"{REMOTE_BASE}{name}.git",
+            "branch": "master",
+            "reset_mode": "hard",
+            "order": 10 + len(self.components),
+        } | fields
+        self.components.append(entry)
+        return entry
+
+    # -- hooks, fakes, observations ------------------------------------------
+
+    def hook(self, name: str, body: str) -> None:
+        """Install a post-update hook for a component."""
+        _script(self.hooks / f"{name}.sh", body)
+
+    def real_hook(self, name: str) -> None:
+        """Install the shipped hook for a component."""
+        dst = self.hooks / f"{name}.sh"
+        shutil.copy(REPO_ROOT / "updater" / "hooks" / f"{name}.sh", dst)
+        dst.chmod(0o755)
+
+    def venv_pip(self, name: str, body: str = "exit 0\n") -> Path:
+        """Give a component a logging venv pip (the sibling .<name>-env layout)."""
+        pip_dir = self.home / f".{name}-env" / "bin"
+        pip_dir.mkdir(parents=True)
+        log = f'echo "venv-pip $*" >>"{self.state}/calls.log"\n'
+        return _script(pip_dir / "pip", log + body)
+
+    def fail_restart(self, unit: str) -> None:
+        """Make `systemctl restart <unit>` fail."""
+        (self.state / f"fail-restart-{unit}").touch()
+
+    def calls(self) -> list[str]:
+        """Every fake root-tool invocation, in order."""
+        log = self.state / "calls.log"
+        return log.read_text().splitlines() if log.exists() else []
+
+    def history(self) -> list[dict]:
+        """Parsed update_history.jsonl."""
+        path = self.home / ".cache" / "blockscreen" / "update_history.jsonl"
+        if not path.exists():
+            return []
+        return [json.loads(line) for line in path.read_text().splitlines()]
+
+    def events(self, component: str) -> list[str]:
+        """History event names for one component."""
+        return [e["event"] for e in self.history() if e["component"] == component]
+
+    def state_file(self) -> dict:
+        """Parsed updater_state.json."""
+        return json.loads(
+            (self.home / ".cache/blockscreen/updater_state.json").read_text()
+        )
+
+    @property
+    def inflight(self) -> Path:
+        """The in-flight batch marker."""
+        return self.home / ".cache" / "blockscreen" / "updater_inflight.json"
+
+    @property
+    def deploy_flag(self) -> Path:
+        """The flag BlocksScreen-deploy.path watches."""
+        return self.home / ".config" / "blockscreen" / ".run-install-updater"
+
+    @property
+    def sentinel(self) -> Path:
+        """The deferred-restart sentinel a self-update hook writes."""
+        return self.run_dir / "updater-restart-needed"
+
+    def service(self):
+        """A fresh UpdateService over the registered components."""
+        import updater.dbus_service as dbus_mod
+
+        recorder = Recorder()
+        svc = dbus_mod.UpdateService(callback=recorder)
+        return svc, recorder
+
+
+def _write_fakes(h: Harness) -> None:
+    """Drop the fake root tools into h.bin."""
+    fmt = {"state": h.state, "bin": h.bin}
+    _script(h.bin / "systemctl", _SYSTEMCTL.format(**fmt))
+    _script(h.bin / "sudo", _SUDO.format(**fmt))
+    _script(h.bin / "apt", _APT.format(**fmt))
+    for tool in ("apt-mark", "dpkg", "bs-apt-helper", "pip3"):
+        _script(h.bin / tool, _LOGGED.format(name=tool, **fmt))
+    # The shipped hook's daemon-reload fallback must not reach the host bus.
+    _script(h.bin / "dbus-send", "exit 1\n")
+
+
+def _patch(mp: pytest.MonkeyPatch, g: dict, **values: object) -> None:
+    """Patch module globals, failing loudly if a name was renamed upstream."""
+    for key, value in values.items():
+        assert key in g, f"harness patches a missing global {key!r}"
+        mp.setitem(g, key, value)
+
+
+@pytest.fixture
+def harness(tmp_path, monkeypatch, mock_sdbus) -> Harness:
+    """Isolated device with every updater path and root tool redirected."""
+    import updater.dbus_service as dbus_mod
+
+    root = tmp_path.resolve()
+    home = root / "home"
+    home.mkdir()
+    h = Harness(root, home, monkeypatch)
+    _write_fakes(h)
+
+    for var in ("XDG_CONFIG_HOME", "GIT_DIR", "GIT_WORK_TREE", "GIT_CONFIG_GLOBAL"):
+        monkeypatch.delenv(var, raising=False)
+    monkeypatch.setenv("HOME", str(home))
+    monkeypatch.setenv("GIT_CONFIG_NOSYSTEM", "1")
+    monkeypatch.setenv("PATH", f"{h.bin}{os.pathsep}{os.environ['PATH']}")
+    (home / ".gitconfig").write_text(
+        "[user]\n\tname = Integration\n\temail = integration@example.invalid\n"
+        "[commit]\n\tgpgsign = false\n"
+        "[init]\n\tdefaultBranch = master\n"
+        "[advice]\n\tdetachedHead = false\n"
+        f'[url "file://{h.remotes}/"]\n\tinsteadOf = {REMOTE_BASE}\n'
+    )
+
+    # sdbus mock re-imports updater; resolve the live modules through the class.
+    svc_g = dbus_mod.UpdateService.__init__.__globals__
+    exec_g = svc_g["git_fetch"].__globals__
+    lock_g = exec_g["restart_sentinel_path"].__globals__
+    comp_g = svc_g["load_components"].__globals__
+    cache = home / ".cache" / "blockscreen"
+    deploy_unit = h.etc / "BlocksScreen-deploy.path"
+    deploy_unit.touch()
+
+    _patch(
+        monkeypatch,
+        svc_g,
+        _STATE_PATH=cache / "updater_state.json",
+        _INFLIGHT_PATH=cache / "updater_inflight.json",
+        _FAULT_MARKER_PATH=cache / "selfheal_fault.json",
+        _HISTORY_PATH=cache / "update_history.jsonl",
+        _APT_STATUS_CACHE=cache / "apt_status_cache.json",
+        _DEPLOY_FLAG=h.deploy_flag,
+        _DEPLOY_PATH_UNIT=deploy_unit,
+        PIP=str(h.bin / "pip3"),
+        SYSTEMCTL=str(h.bin / "systemctl"),
+        klipper_printing=h.printing,
+        load_components=lambda: (comp_g["_build_configs"](h.components), 86_400.0),
+    )
+
+    async def _configured_origin(path: Path) -> str:
+        # get-url expands insteadOf to file://; the raw config keeps the https URL.
+        ok, out = await exec_g["_run"](
+            [exec_g["GIT"], "config", "--get", "remote.origin.url"],
+            cwd=path,
+            timeout=10.0,
+        )
+        return out.strip() if ok else ""
+
+    _patch(
+        monkeypatch,
+        exec_g,
+        SUDO=str(h.bin / "sudo"),
+        SYSTEMCTL=str(h.bin / "systemctl"),
+        APT=str(h.bin / "apt"),
+        APT_MARK=str(h.bin / "apt-mark"),
+        DPKG=str(h.bin / "dpkg"),
+        APT_HELPER=h.bin / "bs-apt-helper",
+        PIP=str(h.bin / "pip3"),
+        KLIPPY_SOCK=root / "no-klippy.sock",
+        _HOOKS_DIR=h.hooks,
+        git_remote_url=_configured_origin,
+    )
+    _patch(monkeypatch, lock_g, _runtime_dir=lambda: h.run_dir)
+    h.dbus_g = dbus_mod.UpdaterDbusService.__init__.__globals__
+    _patch(
+        monkeypatch,
+        h.dbus_g,
+        _STATUS_PATH=h.run_dir / "updater_status.json",
+        _BOOT_DELAY_S=0.0,
+        _DEPLOY_POLL_S=0.05,
+    )
+    # Self-heal supervisors poll systemd on minute timers: out of scope here.
+    for method in ("supervise_ui", "forward_heal_ui", "background_prime_nrestarts"):
+        monkeypatch.setattr(dbus_mod.UpdateService, method, AsyncMock())
+    # Recorded, not run: the daemon must skip it when its own restart is pending.
+    monkeypatch.setattr(dbus_mod.UpdateService, "background_apt_upgrade", h.apt_upgrade)
+    return h
+
+
+class Daemon:
+    """A live UpdaterDbusService plus its recorded signals."""
+
+    def __init__(self, iface) -> None:
+        self.iface = iface
+        self.log: list[tuple[str, tuple]] = []
+        for name in (
+            "step_complete",
+            "component_done",
+            "error",
+            "rollback",
+            "recover_done",
+            "status_ready",
+            "busy_changed",
+            "provisioning_changed",
+        ):
+            setattr(iface, name, Signal(name, self.log))
+
+    def signals(self, name: str) -> list[tuple]:
+        """Args of every emit of one signal."""
+        return [args for sig, args in self.log if sig == name]
+
+    def busy_trace(self) -> list[bool]:
+        """Every busy_changed value, in order."""
+        return [args[0] for args in self.signals("busy_changed")]
+
+    def statuses(self) -> list[dict]:
+        """Every status_ready payload, decoded."""
+        return [json.loads(args[0]) for args in self.signals("status_ready")]
+
+
+@pytest.fixture
+async def daemon(harness):
+    """Factory for live daemons; each is shut down at teardown."""
+    import updater.dbus_service as dbus_mod
+
+    started: list[Daemon] = []
+
+    def start() -> Daemon:
+        d = Daemon(dbus_mod.UpdaterDbusService())
+        started.append(d)
+        return d
+
+    yield start
+    for d in started:
+        await d.iface.shutdown()
+
+
+async def wait_for(predicate, timeout: float = 15.0, what: str = "condition") -> None:
+    """Poll predicate until true; fail with `what` on timeout."""
+    loop = asyncio.get_running_loop()
+    deadline = loop.time() + timeout
+    while not predicate():
+        if loop.time() >= deadline:
+            pytest.fail(f"timed out waiting for {what}")
+        await asyncio.sleep(0.02)
diff --git a/tests/integration/updater/test_updater_daemon_flow.py b/tests/integration/updater/test_updater_daemon_flow.py
new file mode 100644
index 00000000..38d37203
--- /dev/null
+++ b/tests/integration/updater/test_updater_daemon_flow.py
@@ -0,0 +1,197 @@
+"""UpdaterDbusService end to end: busy/overlay contract, boot pass, deploy hold, cancel."""
+
+import asyncio
+
+from conftest import wait_for
+
+UI_FILES = {
+    "updater/dbus_service.py": "# marker\n",
+    "scripts/install-updater.sh": "v1\n",
+}
+
+
+def _idle(d) -> bool:
+    """No update/recover task is still running on the daemon."""
+    return not any(
+        t.get_name().startswith(("update_", "recover_")) and not t.done()
+        for t in d.iface._background_tasks
+    )
+
+
+async def _booted(daemon):
+    """Start a daemon and wait for its first status emit."""
+    d = daemon()
+    await wait_for(lambda: d.statuses(), what="boot status")
+    return d
+
+
+async def test_update_all_round_trip(harness, daemon):
+    origin = harness.remote("klipper")
+    harness.install(origin)
+    new = origin.commit({"klippy.py": "v2\n"})
+    harness.component("klipper", service="klipper.service")
+    d = await _booted(daemon)
+    assert d.statuses()[0]["klipper"]["commits_behind"] == 1
+
+    assert await d.iface.update_all() is True
+    assert await d.iface.update_all() is False  # busy rejects a second run
+    await wait_for(lambda: d.busy_trace() == [True, False], what="busy release")
+
+    assert harness.head("klipper") == new
+    steps = [a for a in d.signals("step_complete") if a[0] == "klipper"]
+    assert steps == [("klipper", n, 4) for n in (1, 2, 3, 4)]
+    assert d.signals("component_done") == [("klipper", True)]
+    await wait_for(lambda: harness.apt_upgrade.await_count == 1, what="apt upgrade")
+
+    seen = len(d.statuses())
+    await d.iface.request_status()
+    await wait_for(lambda: len(d.statuses()) > seen, what="fresh status")
+    assert d.statuses()[-1]["klipper"]["commits_behind"] == 0
+
+
+async def test_nothing_dirty_still_releases_busy(harness, daemon):
+    harness.install(harness.remote("klipper"))
+    harness.component("klipper", service="klipper.service")
+    d = await _booted(daemon)
+
+    assert await d.iface.update_all() is True
+    await wait_for(lambda: d.busy_trace() == [True, False], what="busy release")
+
+    assert d.signals("step_complete") == []
+
+
+async def test_failed_update_releases_busy_and_reports(harness, daemon):
+    origin = harness.remote("klipper")
+    harness.install(origin)
+    old = harness.head("klipper")
+    origin.commit({"klippy.py": "v2\n"})
+    harness.component("klipper", service="klipper.service")
+    harness.hook("klipper", "exit 1\n")
+    d = await _booted(daemon)
+
+    assert await d.iface.update_all() is True
+    await wait_for(lambda: d.busy_trace() == [True, False], what="busy release")
+
+    assert harness.head("klipper") == old
+    assert d.signals("error") == [("klipper", "hook")]
+    assert d.signals("rollback") == [("klipper", True)]
+    assert d.signals("component_done") == [("klipper", False)]
+
+
+async def test_install_update_holds_overlay_until_deploy_finishes(harness, daemon):
+    origin = harness.remote("BlocksScreen", UI_FILES)
+    harness.install(origin)
+    origin.commit({"scripts/install-updater.sh": "v2\n"})
+    harness.component("BlocksScreen", service="BlocksScreen.service")
+    harness.real_hook("BlocksScreen")
+    d = await _booted(daemon)
+
+    assert await d.iface.update_all() is True
+    await wait_for(harness.deploy_flag.exists, what="deploy flag")
+    await wait_for(lambda: _idle(d), what="batch end")
+    await asyncio.sleep(0.3)  # several hold polls
+
+    # Batch done, install-updater still running: the overlay must not drop.
+    assert d.busy_trace() == [True]
+    assert await d.iface.get_busy() is True
+    assert await d.iface.update_all() is False
+    harness.apt_upgrade.assert_not_awaited()
+
+    harness.deploy_flag.unlink()  # install-updater's ExecStopPost
+    await wait_for(lambda: d.busy_trace() == [True, False], what="hold release")
+
+
+async def test_deploy_hold_is_capped(harness, daemon):
+    harness.mp.setitem(harness.dbus_g, "_DEPLOY_HOLD_CAP_S", 0.3)
+    harness.deploy_flag.parent.mkdir(parents=True)
+    harness.deploy_flag.touch()  # a hung install-updater never clears it
+
+    d = daemon()
+    assert await d.iface.get_busy() is True
+
+    await wait_for(lambda: d.busy_trace() == [False], timeout=5.0, what="cap")
+    assert harness.deploy_flag.exists()
+
+    origin = harness.remote("klipper")
+    harness.install(origin)
+    origin.commit({"klippy.py": "v2\n"})
+    harness.component("klipper", service="klipper.service")
+    assert await d.iface.update_all() is True
+    # The same stuck flag must not re-hold at the end of every update.
+    await wait_for(lambda: d.busy_trace() == [False, True, False], what="no re-hold")
+    assert d.iface._deploy_hold is None
+
+
+async def test_stale_flag_under_dead_deploy_units_does_not_hold(harness, daemon):
+    harness.deploy_flag.parent.mkdir(parents=True)
+    harness.deploy_flag.touch()
+    for unit in ("BlocksScreen-deploy.path", "BlocksScreen-deploy.service"):
+        (harness.state / f"inactive-{unit}").touch()  # path unit failed/stopped
+
+    d = daemon()
+    await wait_for(lambda: d.busy_trace() == [False], what="stale flag release")
+    assert harness.deploy_flag.exists()
+
+
+async def test_restarted_daemon_resumes_hold_silently(harness, daemon):
+    harness.deploy_flag.parent.mkdir(parents=True)
+    harness.deploy_flag.touch()  # install-updater restarted us mid-run
+
+    d = daemon()
+    # Not exported yet at __init__: clients read it via get_busy, no signal.
+    assert d.busy_trace() == []
+    assert await d.iface.get_busy() is True
+
+    harness.deploy_flag.unlink()
+    await wait_for(lambda: d.busy_trace() == [False], what="hold release")
+
+
+async def test_boot_provisioning_raises_overlay(harness, daemon):
+    origin = harness.remote("klipper")
+    harness.component("klipper", service="klipper.service", install_if_missing=True)
+
+    d = await _booted(daemon)
+
+    assert [a[0] for a in d.signals("provisioning_changed")] == [True, False]
+    assert d.busy_trace() == [True, False]
+    assert await d.iface.get_provisioning() is False
+    assert harness.head("klipper") == harness.git("rev-parse", "HEAD", cwd=origin.bare)
+    assert d.statuses()[0]["klipper"]["commits_behind"] == 0
+
+
+async def test_boot_reconcile_reverts_power_cut(harness, daemon):
+    origin = harness.remote("klipper")
+    harness.install(origin)
+    old = harness.head("klipper")
+    origin.commit({"klippy.py": "v2\n"})
+    harness.git("pull", "-q", cwd=harness.home / "klipper")
+    harness.inflight.parent.mkdir(parents=True, exist_ok=True)
+    harness.inflight.write_text(f'{{"klipper": "{old}"}}')
+    harness.component("klipper", service="klipper.service")
+
+    await _booted(daemon)
+
+    assert harness.head("klipper") == old
+    assert "boot_rollback" in harness.events("klipper")
+    assert not harness.inflight.exists()
+
+
+async def test_cancel_mid_hook_reverts_and_releases(harness, daemon):
+    origin = harness.remote("klipper")
+    harness.install(origin)
+    old = harness.head("klipper")
+    origin.commit({"klippy.py": "v2\n"})
+    harness.component("klipper", service="klipper.service")
+    harness.hook("klipper", "sleep 20\n")
+    d = await _booted(daemon)
+
+    assert await d.iface.update_all() is True
+    await wait_for(
+        lambda: ("klipper", 3, 4) in d.signals("step_complete"), what="hook start"
+    )
+    await d.iface.cancel()
+
+    assert harness.head("klipper") == old
+    assert d.signals("rollback") == [("klipper", True)]
+    assert d.busy_trace() == [True, False]
+    assert not harness.inflight.exists()
diff --git a/tests/integration/updater/test_updater_git_flow.py b/tests/integration/updater/test_updater_git_flow.py
new file mode 100644
index 00000000..919e3bf7
--- /dev/null
+++ b/tests/integration/updater/test_updater_git_flow.py
@@ -0,0 +1,333 @@
+"""UpdateService end to end against real git repos: update, rollback, heal, install."""
+
+UI_FILES = {
+    "updater/dbus_service.py": "# marker\n",
+    "scripts/install-updater.sh": "v1\n",
+}
+
+
+def _klipper(harness, **fields: object):
+    """Installed klipper one commit behind its origin; returns (origin, old, new)."""
+    origin = harness.remote("klipper")
+    harness.install(origin)
+    old = harness.head("klipper")
+    new = origin.commit({"klippy.py": "v2\n"})
+    harness.component("klipper", service="klipper.service", **fields)
+    return origin, old, new
+
+
+def _rollbacks(harness, name: str) -> list[dict]:
+    return [
+        e
+        for e in harness.history()
+        if e["component"] == name and e["event"] == "rollback"
+    ]
+
+
+async def test_update_moves_to_origin_tip_and_restarts(harness):
+    _, old, new = _klipper(harness)
+    harness.venv_pip("klipper")
+    svc, rec = harness.service()
+
+    assert await svc.update_all({"klipper"}) is True
+
+    assert harness.head("klipper") == new
+    assert harness.state_file()["klipper"]["prev_hash"] == old
+    assert harness.events("klipper") == ["update_start", "update_success"]
+    assert not harness.inflight.exists()
+    assert [s[1] for s in rec.of("on_step")] == [1, 2, 3, 4]
+    assert rec.of("on_component_done") == [("klipper", True)]
+    calls = harness.calls()
+    assert "systemctl restart klipper.service" in calls
+    assert "systemctl is-active klipper.service" in calls
+    # No requirements.txt: the venv is found but pip is never asked to install.
+    assert not [c for c in calls if c.startswith("venv-pip")]
+
+
+async def test_deps_install_runs_component_venv_pip(harness):
+    origin, _, _ = _klipper(harness)
+    req = origin.commit({"requirements.txt": "pyserial\n"})
+    harness.venv_pip("klipper")
+    svc, _ = harness.service()
+
+    assert await svc.update_all({"klipper"}) is True
+
+    assert harness.head("klipper") == req
+    req_path = harness.home / "klipper" / "requirements.txt"
+    assert f"venv-pip install -r {req_path} --quiet" in harness.calls()
+
+
+async def test_hook_failure_reverts(harness):
+    _, old, _ = _klipper(harness)
+    harness.hook("klipper", "exit 3\n")
+    svc, rec = harness.service()
+
+    assert await svc.update_all({"klipper"}) is False
+
+    assert harness.head("klipper") == old
+    [rb] = _rollbacks(harness, "klipper")
+    assert (rb["reason"], rb["reverted_to"], rb["ok"]) == ("hook", old[:12], True)
+    assert rec.of("on_error") == [("klipper", "hook")]
+    assert rec.of("on_rollback") == [("klipper", True)]
+    assert "systemctl restart klipper.service" not in harness.calls()
+    assert not harness.inflight.exists()
+
+
+async def test_hook_receives_hashes(harness):
+    _, old, new = _klipper(harness)
+    out = harness.root / "hook.env"
+    harness.hook(
+        "klipper",
+        f'echo "$COMPONENT_NAME $PREV_HASH $NEW_HASH $COMPONENT_PATH" >"{out}"\n',
+    )
+    svc, _ = harness.service()
+
+    assert await svc.update_all({"klipper"}) is True
+
+    assert out.read_text().split() == [
+        "klipper",
+        old,
+        new,
+        str(harness.home / "klipper"),
+    ]
+
+
+async def test_restart_failure_reverts(harness):
+    _, old, _ = _klipper(harness)
+    harness.fail_restart("klipper.service")
+    svc, rec = harness.service()
+
+    assert await svc.update_all({"klipper"}) is False
+
+    assert harness.head("klipper") == old
+    assert [rb["reason"] for rb in _rollbacks(harness, "klipper")] == ["restart"]
+    assert rec.of("on_component_done") == [("klipper", False)]
+
+
+async def test_deps_failure_reverts(harness):
+    origin, old, _ = _klipper(harness)
+    origin.commit({"requirements.txt": "broken-dep\n"})
+    harness.venv_pip("klipper", 'case "$*" in *"-r "*) exit 1 ;; esac\nexit 0\n')
+    svc, _ = harness.service()
+
+    assert await svc.update_all({"klipper"}) is False
+
+    assert harness.head("klipper") == old
+    assert [rb["reason"] for rb in _rollbacks(harness, "klipper")] == ["deps"]
+
+
+async def test_one_failure_does_not_block_the_rest(harness):
+    _, k_old, _ = _klipper(harness)
+    moon = harness.remote("moonraker")
+    harness.install(moon)
+    m_new = moon.commit({"moonraker.py": "v2\n"})
+    harness.component("moonraker", service="moonraker.service")
+    harness.hook("klipper", "exit 1\n")
+    svc, rec = harness.service()
+
+    assert await svc.update_all({"klipper", "moonraker"}) is False
+
+    assert harness.head("klipper") == k_old
+    assert harness.head("moonraker") == m_new
+    assert sorted(rec.of("on_component_done")) == [
+        ("klipper", False),
+        ("moonraker", True),
+    ]
+    assert harness.events("moonraker") == ["update_start", "update_success"]
+
+
+async def test_dead_branch_is_skipped_untouched(harness):
+    origin = harness.remote("klipper")
+    harness.install(origin)
+    old = harness.head("klipper")
+    harness.component("klipper", service="klipper.service", branch="gone")
+    svc, rec = harness.service()
+
+    assert await svc.update_all({"klipper"}) is False
+
+    assert harness.head("klipper") == old
+    assert rec.of("on_error") == [
+        ("klipper", "branch origin/gone not found - fix components.yaml")
+    ]
+    assert harness.history() == []
+
+
+async def test_branch_switch_follows_components_yaml(harness):
+    origin = harness.remote("klipper")
+    harness.install(origin)
+    feature = origin.commit({"feature.py": "x\n"}, branch="feature")
+    harness.component("klipper", service="klipper.service", branch="feature")
+    svc, _ = harness.service()
+
+    assert await svc.update_all({"klipper"}) is True
+
+    assert harness.branch("klipper") == "feature"
+    assert harness.head("klipper") == feature
+
+
+async def test_hard_reset_discards_local_edits(harness):
+    _, _, new = _klipper(harness)
+    (harness.home / "klipper" / "README").write_text("edited on the printer\n")
+    svc, _ = harness.service()
+
+    status = await svc.check_status(force=True)
+    assert status["klipper"].has_local_changes
+    assert status["klipper"].commits_behind == 1
+    assert await svc.update_all({"klipper"}) is True
+
+    assert harness.head("klipper") == new
+    assert (harness.home / "klipper" / "README").read_text() == "klipper\n"
+
+
+async def test_non_https_remote_is_refused(harness):
+    origin, old, _ = _klipper(harness)
+    harness.git(
+        "remote", "set-url", "origin", str(origin.bare), cwd=harness.home / "klipper"
+    )
+    svc, rec = harness.service()
+
+    assert await svc.update_all({"klipper"}) is False
+
+    assert harness.head("klipper") == old
+    assert rec.of("on_error") == [("klipper", "insecure remote")]
+    assert not harness.inflight.exists()
+
+
+async def test_ui_tip_without_updater_is_refused(harness):
+    origin = harness.remote("BlocksScreen", UI_FILES)
+    harness.install(origin)
+    old = harness.head("BlocksScreen")
+    origin.commit({"updater/dbus_service.py": None}, msg="drop updater")
+    harness.component("BlocksScreen", service="BlocksScreen.service")
+    svc, rec = harness.service()
+
+    assert await svc.update_all({"BlocksScreen"}) is False
+
+    assert harness.head("BlocksScreen") == old
+    assert rec.of("on_error") == [("BlocksScreen", "refusing downgrade past updater")]
+
+
+async def test_install_file_change_hands_off_to_deploy(harness):
+    origin = harness.remote("BlocksScreen", UI_FILES)
+    harness.install(origin)
+    new = origin.commit({"scripts/install-updater.sh": "v2\n"})
+    harness.component("BlocksScreen", service="BlocksScreen.service")
+    harness.real_hook("BlocksScreen")
+    svc, _ = harness.service()
+
+    assert await svc.update_all({"BlocksScreen"}) is True
+
+    assert harness.head("BlocksScreen") == new
+    assert harness.deploy_flag.exists()
+    assert not harness.sentinel.exists()
+    assert svc.deploy_pending()
+    assert svc.daemon_restart_pending
+    calls = harness.calls()
+    assert "systemctl --no-block restart BlocksScreen.service" in calls
+    # The UI is fire-and-forget: never a blocking restart of our own screen.
+    assert "systemctl restart BlocksScreen.service" not in calls
+
+
+async def test_plain_ui_change_sets_no_deploy_flag(harness):
+    origin = harness.remote("BlocksScreen", UI_FILES)
+    harness.install(origin)
+    origin.commit({"README": "docs\n"})
+    harness.component("BlocksScreen", service="BlocksScreen.service")
+    harness.real_hook("BlocksScreen")
+    svc, _ = harness.service()
+
+    assert await svc.update_all({"BlocksScreen"}) is True
+
+    assert not harness.deploy_flag.exists()
+    assert not svc.daemon_restart_pending
+    assert "systemctl --no-block restart BlocksScreen.service" in harness.calls()
+
+
+async def test_reconcile_reverts_power_cut_batch(harness):
+    _, old, new = _klipper(harness)
+    path = harness.home / "klipper"
+    # Power cut mid-batch: the tree moved, the marker still names the rollback point.
+    harness.git("pull", "-q", cwd=path)
+    assert harness.head("klipper") == new
+    harness.inflight.parent.mkdir(parents=True, exist_ok=True)
+    harness.inflight.write_text(f'{{"klipper": "{old}"}}')
+    svc, _ = harness.service()
+
+    assert await svc.reconcile() is True
+
+    assert harness.head("klipper") == old
+    assert "boot_rollback" in harness.events("klipper")
+    assert not harness.inflight.exists()
+
+
+async def test_recover_returns_to_prev_hash(harness):
+    _, old, new = _klipper(harness)
+    svc, rec = harness.service()
+    assert await svc.update_all({"klipper"}) is True
+    assert harness.head("klipper") == new
+    restarts_before = harness.calls().count("systemctl restart klipper.service")
+
+    assert await svc.recover("klipper", hard=True) is True
+
+    assert harness.head("klipper") == old
+    assert rec.of("on_recover") == [("klipper", True)]
+    assert (
+        harness.calls().count("systemctl restart klipper.service")
+        == restarts_before + 1
+    )
+
+
+async def test_single_component_update_and_rollback(harness):
+    _, old, new = _klipper(harness)
+    svc, _ = harness.service()
+    assert await svc.update_component("klipper") is True
+    assert harness.head("klipper") == new
+    assert not harness.inflight.exists()
+
+    harness.hook("klipper", "exit 1\n")
+    harness.git("reset", "-q", "--hard", old, cwd=harness.home / "klipper")
+    svc2, rec = harness.service()
+
+    assert await svc2.update_component("klipper") is False
+
+    assert harness.head("klipper") == old
+    assert rec.of("on_rollback") == [("klipper", True)]
+
+
+async def test_provision_missing_clones_and_enables(harness):
+    origin = harness.remote("klipper")
+    harness.component("klipper", service="klipper.service", install_if_missing=True)
+    svc, rec = harness.service()
+    busy: list[bool] = []
+
+    assert await svc.provision_missing(busy.append) is False
+
+    assert busy == [True, False]
+    assert harness.head("klipper") == harness.git("rev-parse", "HEAD", cwd=origin.bare)
+    assert harness.events("klipper") == ["install_start", "install_success"]
+    assert rec.of("on_component_done") == [("klipper", True)]
+    assert "systemctl enable klipper.service" in harness.calls()
+
+
+async def test_provision_waits_while_printing(harness):
+    harness.remote("klipper")
+    harness.component("klipper", install_if_missing=True)
+    harness.printing.return_value = True
+    svc, _ = harness.service()
+    busy: list[bool] = []
+
+    assert await svc.provision_missing(busy.append) is True
+
+    assert busy == []
+    assert not (harness.home / "klipper").exists()
+
+
+async def test_provision_offline_never_raises_busy(harness):
+    harness.component("klipper", install_if_missing=True)  # no origin exists
+    svc, _ = harness.service()
+    busy: list[bool] = []
+
+    assert await svc.provision_missing(busy.append) is True
+
+    assert busy == []
+    assert not (harness.home / "klipper").exists()