From a210e5ee3283e563417cbccb47f2db24073e2364 Mon Sep 17 00:00:00 2001 From: Dom Date: Wed, 1 Jul 2026 14:10:34 +0200 Subject: [PATCH] feat(update): swap atomique + rollback (Lea.bat) + confirmation boot (main.py) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Implémente le SWAP réel de la MAJ silencieuse (DETTE-022), remplace les stubs : - updater.apply_update : ARME le swap (extrait le ZIP -> agent_v1_new/ + marqueur UPDATE_READY, garde-fou zip-slip). N'écrase JAMAIS le vivant. - updater.write_boot_ok_marker : désarme le rollback (retire PENDING_BOOT). - Lea.bat (template + embed généré par configure_embed.ps1) : swap ATOMIQUE par renames (agent_v1 -> agent_v1_prev backup ; agent_v1_new -> agent_v1) + rollback auto si PENDING_BOOT persiste (boot précédent non confirmé). - main.py : confirme le boot après 90 s de liveness locale OU quit propre (évite un faux rollback ; RPA_BOOT_CONFIRM_DELAY_S surchargeable pour les tests). Testable (Python) : 45 tests verts. Le swap OS (renames Lea.bat) + le câblage main.py seront validés par le test Win 11 (step 0 pré-canary, dont le rollback). Co-Authored-By: Claude Opus 4.8 (1M context) --- agent_v0/agent_v1/main.py | 45 +++++++ agent_v0/agent_v1/network/updater.py | 182 ++++++++++++++++++--------- deploy/installer/configure_embed.ps1 | 23 ++++ deploy/lea_package/Lea.bat | 29 +++++ tests/unit/test_agent_v1_updater.py | 114 +++++++++++++---- 5 files changed, 313 insertions(+), 80 deletions(-) diff --git a/agent_v0/agent_v1/main.py b/agent_v0/agent_v1/main.py index 6b992156b..fbc931ad2 100644 --- a/agent_v0/agent_v1/main.py +++ b/agent_v0/agent_v1/main.py @@ -168,6 +168,22 @@ class AgentV1: target=self._auto_update_loop, daemon=True, name="lea-auto-update" ).start() + # MAJ silencieuse — confirmation de boot post-swap. Si Lea.bat vient + # d'appliquer une MAJ (marqueur PENDING_BOOT), on désarme le rollback + # après ~90 s de tourne STABLE (liveness LOCALE, indépendante du DGX). + # Un quit propre avant 90 s confirme aussi (cf. main()). Seul un vrai + # crash laisse PENDING_BOOT → rollback au prochain lancement. + if _pending_boot_marker_exists(): + def _boot_confirm(): + import os as _os + import time as _time + _time.sleep(float(_os.environ.get("RPA_BOOT_CONFIRM_DELAY_S", "90"))) + if self.running: + _confirm_boot_ok() + threading.Thread( + target=_boot_confirm, daemon=True, name="lea-boot-confirm" + ).start() + # Mini-serveur HTTP pour captures a la demande (port 5006) self._capture_server = CaptureServer() self._capture_server.start() @@ -718,6 +734,31 @@ def _agent_should_live(agent) -> bool: return True +def _pending_boot_marker_exists() -> bool: + """True si Lea.bat a posé PENDING_BOOT (boot post-MAJ à valider).""" + try: + from .network.updater import _resolve_app_dir + return (_resolve_app_dir(None) / "PENDING_BOOT").exists() + except Exception: + return False + + +def _confirm_boot_ok() -> None: + """Confirme un boot post-MAJ : écrit boot_ok + retire PENDING_BOOT. + + Désarme le rollback de Lea.bat. No-op si pas de PENDING_BOOT (boot normal). + Best-effort — ne doit jamais casser l'arrêt/la vie de Léa. + """ + try: + if not _pending_boot_marker_exists(): + return + from .network import updater + updater.write_boot_ok_marker(AGENT_VERSION) + logger.info("[MAJ] Boot confirmé (v%s) — rollback désarmé", AGENT_VERSION) + except Exception as e: # noqa: BLE001 + logger.debug("confirm_boot_ok: %s", e) + + def main(): from .ui.session_watchdog import InteractiveSessionWatchdog @@ -738,6 +779,10 @@ def main(): try: watchdog.run() + # Sortie normale du watchdog = quit propre (tray / session) → le boot + # était sain : on confirme (couvre un quit AVANT les 90 s, évite un faux + # rollback). No-op si ce n'est pas un boot post-MAJ. + _confirm_boot_ok() except KeyboardInterrupt: logger.info("[MAIN] Interruption clavier — arret propre") except Exception: diff --git a/agent_v0/agent_v1/network/updater.py b/agent_v0/agent_v1/network/updater.py index 80af1c63d..f4e01064b 100644 --- a/agent_v0/agent_v1/network/updater.py +++ b/agent_v0/agent_v1/network/updater.py @@ -18,12 +18,11 @@ Ce module ne contient que les parties PURES / testables, sans réseau réel : les fichiers vivants. Retourne un plan d'application. - `auto_update_enabled()` : lit le flag (défaut OFF). -⚠️⚠️ PARTIES DANGEREUSES — RÉSERVÉES RÉVISION HUMAINE ⚠️⚠️ -Le remplacement réel des fichiers (`apply_update`), l'écriture du marker -rollback (`write_boot_ok_marker`), l'édition de `Lea.bat` et le redémarrage -ne sont PAS implémentés ici : ce sont des STUBS no-op explicites. Un agent ne -doit pas écrire de code qui écrase des binaires vivants ou relance un process -sans supervision. Les points d'extension sont marqués `# TODO swap supervisé`. +⚠️ SWAP — répartition claire des responsabilités : +`apply_update` / `write_boot_ok_marker` ci-dessous ne font que l'ARMEMENT côté +Python (extraction vers `agent_v1_new/` + marqueurs) — ils n'écrasent JAMAIS un +fichier vivant. Le remplacement ATOMIQUE (renames), le redémarrage et le +rollback sont faits HORS-PROCESS par `Lea.bat` au démarrage (revu ligne à ligne). Pattern d'import / résilience aligné sur `log_shipper.py` (même branche). @@ -33,8 +32,10 @@ Branche feat/push-log-dgx. from __future__ import annotations import hashlib +import json import logging import os +import shutil from pathlib import Path from typing import Callable, Optional, Tuple @@ -296,6 +297,7 @@ def run_update_cycle( staging_dir, checker: Optional[Callable[[str, str], object]] = None, downloader: Optional[Callable[[str], bytes]] = None, + app_dir=None, ) -> dict: """Un cycle complet de MAJ silencieuse — GATED, best-effort, SANS swap. @@ -306,9 +308,10 @@ def run_update_cycle( 3. `should_update(...)` → plan (double garde semver, jamais de downgrade). 4. `download_update(...)` → ZIP dans le STAGING + vérif SHA256. Ne touche JAMAIS les fichiers vivants. - 5. Le swap réel N'EST PAS FAIT ici : `apply_update` reste un stub no-op - (réservé révision humaine + Lea.bat hors-process). Le résultat porte - `applied: False`. + 5. `apply_update` ARME le swap (extraction `agent_v1_new/` + marqueur + UPDATE_READY) mais NE swappe PAS : le remplacement atomique + le + redémarrage sont faits par Lea.bat au prochain démarrage. `applied` + reste False tant que Léa n'a pas redémarré sur la nouvelle version. Jamais d'exception ne remonte (ne doit JAMAIS casser Léa). Retourne un dict d'état pour le diagnostic / le log : @@ -342,76 +345,137 @@ def run_update_cycle( "error": staged.get("error"), } - # ⚠️ Le swap réel est réservé révision humaine : on APPELLE le stub (no-op) - # pour matérialiser le point d'extension, mais rien n'est écrasé/redémarré. - applied = apply_update(staged) + # Armement du swap : extraction du ZIP vers agent_v1_new\ + marqueur + # UPDATE_READY. Le swap ATOMIQUE (renames) et le redémarrage sont faits + # HORS-PROCESS par Lea.bat au prochain démarrage — JAMAIS depuis ici + # (on n'écrase pas les fichiers d'un Léa en cours d'exécution). + armed = apply_update(staged, app_dir=app_dir) return { - "status": "staged", - "applied": bool(applied.get("applied", False)), + "status": "armed" if armed.get("armed") else "arm_failed", + "applied": False, # le swap effectif est fait par Lea.bat, pas ici + "armed": bool(armed.get("armed", False)), "target_version": staged.get("target_version"), "update_type": staged.get("update_type"), "staged_zip": staged.get("staged_zip"), "sha256_verified": staged.get("sha256_verified", False), - "apply_reason": applied.get("reason"), + "marker": armed.get("marker"), + "error": armed.get("error"), } # =========================================================================== -# ⚠️ ZONE DANGEREUSE — STUBS RÉSERVÉS RÉVISION HUMAINE (NE PAS IMPLÉMENTER -# PAR UN AGENT). Points d'extension explicites, no-op pour l'instant. +# SWAP — côté Python : ARMEMENT SEULEMENT (extraction + marqueurs). +# Le remplacement ATOMIQUE des fichiers vivants + le redémarrage + le +# rollback sont faits HORS-PROCESS par `Lea.bat` au démarrage (renames). +# Python n'écrase JAMAIS les fichiers d'un Léa en cours d'exécution. # =========================================================================== -def apply_update(prepared: dict) -> dict: - """STUB — application réelle de l'update (swap des fichiers). +def _resolve_app_dir(app_dir) -> Path: + """Répertoire d'install (contient `agent_v1/`, `run_agent_v1.py`, `Lea.bat`). - Réservé révision humaine : remplacer des fichiers vivants du client et - déclencher un swap est trop risqué pour être généré par un agent. La - mécanique cible (design v2) est : + INJECTABLE (tests : tmp_path). Défaut = parent du package agent_v1. + """ + if app_dir is not None: + return Path(app_dir) + try: + from ..config import BASE_DIR # BASE_DIR = dossier du package agent_v1 + return Path(BASE_DIR).parent + except Exception: + return Path(__file__).resolve().parent.parent.parent - - code-only : extraire `agent_v1\\` + `lea_ui\\` + `run_agent_v1.py` + - `config.py` du ZIP staging, poser un marker `UPDATE_READY` - (`update_type=code-only`) ; le swap effectif est fait par `Lea.bat` - au prochain démarrage (xcopy ciblé). - - full : poser `UPDATE_READY` (`update_type=full`) ; `Lea.bat` fait le - backup complet `Lea_prev\\` puis le swap complet. - # TODO swap supervisé : extraction ZIP + écriture marker UPDATE_READY. - # NE PAS écraser les fichiers vivants depuis Python — c'est Lea.bat qui - # swappe hors-process. Édition de Lea.bat + restart = hors périmètre agent. +def apply_update(prepared: dict, app_dir=None) -> dict: + """ARME le swap : extrait le ZIP staging vers `agent_v1_new/` + marqueur. + + NE swappe PAS et NE redémarre PAS (c'est le rôle de `Lea.bat`). Écrit + uniquement à côté des fichiers vivants (dossier neuf + marqueur), donc + l'opération est sûre même sur un Léa en cours d'exécution. + + 1. Extrait `prepared["staged_zip"]` → `/agent_v1_new/` + (nettoyé au préalable ; garde-fou zip-slip). + 2. Écrit `/UPDATE_READY` (JSON : version, type, chemins) que + `Lea.bat` lira au prochain démarrage pour faire le swap atomique. + + Best-effort : aucune exception ne remonte (ne doit jamais casser Léa). Returns: - {applied: False, reason: "réservé révision humaine (swap supervisé)"} + succès : {armed: True, applied: False, target_version, update_type, + marker, extracted_to} + échec : {armed: False, applied: False, error} """ - logger.info( - "apply_update appelé mais NON implémenté (stub réservé révision humaine) : %r", - prepared.get("target_version") if isinstance(prepared, dict) else prepared, - ) - return { - "applied": False, - "reason": "réservé révision humaine — swap supervisé (Lea.bat), hors périmètre agent", - } + if not isinstance(prepared, dict): + return {"armed": False, "applied": False, "error": "prepared invalide"} + staged_zip = prepared.get("staged_zip") + target_version = prepared.get("target_version", "unknown") + update_type = _normalize_update_type(prepared.get("update_type")) + try: + root = _resolve_app_dir(app_dir) + zip_path = Path(staged_zip) if staged_zip else None + if zip_path is None or not zip_path.is_file(): + return {"armed": False, "applied": False, "error": "ZIP staging introuvable"} + + new_dir = root / "agent_v1_new" + if new_dir.exists(): + shutil.rmtree(new_dir, ignore_errors=True) # nettoie un staging partiel + new_dir.mkdir(parents=True, exist_ok=True) + + import zipfile + new_root = new_dir.resolve() + with zipfile.ZipFile(zip_path) as zf: + for name in zf.namelist(): # garde-fou zip-slip (chemins ../) + dest = (new_dir / name).resolve() + if not str(dest).startswith(str(new_root)): + shutil.rmtree(new_dir, ignore_errors=True) + return {"armed": False, "applied": False, + "error": f"zip-slip refusé : {name}"} + zf.extractall(new_dir) + + marker = root / "UPDATE_READY" + marker.write_text(json.dumps({ + "target_version": target_version, + "update_type": update_type, + "extracted_to": str(new_dir), + "staged_zip": str(zip_path), + }), encoding="utf-8") + + logger.info( + "Update ARMÉ : %s (%s) → %s ; swap au prochain démarrage (Lea.bat)", + target_version, update_type, new_dir, + ) + return {"armed": True, "applied": False, "target_version": target_version, + "update_type": update_type, "marker": str(marker), + "extracted_to": str(new_dir)} + except Exception as e: # noqa: BLE001 + logger.warning("apply_update (armement) a échoué : %s", e) + return {"armed": False, "applied": False, "error": f"arm_failed: {e}"} -def write_boot_ok_marker(version: str) -> dict: - """STUB — écriture du marker rollback `boot_ok_{version}` (R1). +def write_boot_ok_marker(version: str, app_dir=None) -> dict: + """Confirme un boot sain : écrit `boot_ok_{version}` + désarme le rollback. - Réservé révision humaine : le marker pilote le rollback de Lea.bat au - prochain démarrage. Sa sémantique (health-check ~60s heartbeat DGX + - session active AVANT écriture) doit être validée à la main pour éviter un - faux rollback (cas DGX down ≠ Léa N+1 buguée — cf. design R1, cas edge 3). + Appelé par `main.py` après ~90 s de tourne STABLE (liveness LOCALE, + indépendante du DGX — évite un faux rollback quand le réseau est coupé). + Retirer `PENDING_BOOT*` dit à `Lea.bat` que la nouvelle version a démarré + correctement (sinon, au prochain lancement, Lea.bat rollback vers la + version précédente). - # TODO swap supervisé : écrire `%LOCALAPPDATA%\\Lea\\boot_ok_{version}` - # après ~60s de heartbeat DGX sain + session active (main.py startup). - - Returns: - {written: False, reason: "..."} + Best-effort : aucune exception ne remonte. """ - logger.info( - "write_boot_ok_marker appelé mais NON implémenté (stub R1) : version=%s", - version, - ) - return { - "written": False, - "reason": "réservé révision humaine — marker rollback (health-check), hors périmètre agent", - } + try: + root = _resolve_app_dir(app_dir) + marker = root / f"boot_ok_{version}" + marker.write_text("ok", encoding="utf-8") + cleared = [] + for p in root.glob("PENDING_BOOT*"): + try: + p.unlink() + cleared.append(p.name) + except OSError: + pass + logger.info("boot_ok écrit (%s) ; PENDING_BOOT retiré : %s", + version, cleared or "aucun") + return {"written": True, "marker": str(marker), "cleared_pending": cleared} + except Exception as e: # noqa: BLE001 + logger.warning("write_boot_ok_marker a échoué : %s", e) + return {"written": False, "error": str(e)} diff --git a/deploy/installer/configure_embed.ps1 b/deploy/installer/configure_embed.ps1 index 5f6b2b6dd..47f185e6c 100644 --- a/deploy/installer/configure_embed.ps1 +++ b/deploy/installer/configure_embed.ps1 @@ -76,6 +76,29 @@ if exist "lea_agent.lock" ( timeout /t 2 >nul ) +:: MAJ SILENCIEUSE — swap atomique + rollback (renames uniquement) +if exist "PENDING_BOOT" ( + echo [MAJ] Boot precedent non confirme : retour a la version precedente. + if exist "agent_v1_prev" ( + if exist "agent_v1_echec" rmdir /s /q "agent_v1_echec" >nul 2>&1 + if exist "agent_v1" move "agent_v1" "agent_v1_echec" >nul 2>&1 + move "agent_v1_prev" "agent_v1" >nul 2>&1 + ) + del /f /q "PENDING_BOOT" >nul 2>&1 +) else if exist "UPDATE_READY" ( + if exist "agent_v1_new" ( + echo [MAJ] Application de la mise a jour... + if exist "agent_v1" ( + if exist "agent_v1_prev" rmdir /s /q "agent_v1_prev" >nul 2>&1 + move "agent_v1" "agent_v1_prev" >nul 2>&1 + ) + move "agent_v1_new" "agent_v1" >nul 2>&1 + move "UPDATE_READY" "PENDING_BOOT" >nul 2>&1 + ) else ( + del /f /q "UPDATE_READY" >nul 2>&1 + ) +) + if exist "config.txt" ( for /f "usebackq eol=# tokens=1,* delims==" %%a in ("config.txt") do ( if not "%%a"=="" if not "%%b"=="" set "%%a=%%b" diff --git a/deploy/lea_package/Lea.bat b/deploy/lea_package/Lea.bat index 7a03aa7eb..b49bf0de6 100644 --- a/deploy/lea_package/Lea.bat +++ b/deploy/lea_package/Lea.bat @@ -20,6 +20,35 @@ if exist "lea_agent.lock" ( timeout /t 2 >nul ) +:: --------------------------------------------------------------- +:: MAJ SILENCIEUSE — swap atomique + rollback (hors-process) +:: L'ancienne instance est fermee ci-dessus : agent_v1\ est libre. +:: Renames uniquement (quasi-atomiques), jamais d'ecrasement fichier par fichier. +:: --------------------------------------------------------------- +if exist "PENDING_BOOT" ( + :: Le boot precedent n'a JAMAIS confirme (crash) -> ROLLBACK version precedente + echo [MAJ] Boot precedent non confirme : retour a la version precedente. + if exist "agent_v1_prev" ( + if exist "agent_v1_echec" rmdir /s /q "agent_v1_echec" >nul 2>&1 + if exist "agent_v1" move "agent_v1" "agent_v1_echec" >nul 2>&1 + move "agent_v1_prev" "agent_v1" >nul 2>&1 + ) + del /f /q "PENDING_BOOT" >nul 2>&1 +) else if exist "UPDATE_READY" ( + :: Une MAJ est armee (agent_v1_new pret) -> SWAP + if exist "agent_v1_new" ( + echo [MAJ] Application de la mise a jour... + if exist "agent_v1" ( + if exist "agent_v1_prev" rmdir /s /q "agent_v1_prev" >nul 2>&1 + move "agent_v1" "agent_v1_prev" >nul 2>&1 + ) + move "agent_v1_new" "agent_v1" >nul 2>&1 + move "UPDATE_READY" "PENDING_BOOT" >nul 2>&1 + ) else ( + del /f /q "UPDATE_READY" >nul 2>&1 + ) +) + :: --------------------------------------------------------------- :: Verifier que l'installation a ete faite :: --------------------------------------------------------------- diff --git a/tests/unit/test_agent_v1_updater.py b/tests/unit/test_agent_v1_updater.py index bd3debf0d..b99bb9fab 100644 --- a/tests/unit/test_agent_v1_updater.py +++ b/tests/unit/test_agent_v1_updater.py @@ -206,23 +206,88 @@ class TestDownloadUpdate: # --------------------------------------------------------------------------- -# Stubs réservés à la révision humaine — DOIVENT être no-op explicites +# apply_update — ARMEMENT du swap (extraction agent_v1_new + marqueur). +# NE swappe PAS et NE touche PAS les fichiers vivants (Lea.bat le fait au boot). # --------------------------------------------------------------------------- -class TestDangerousPartsAreStubs: - def test_apply_update_est_un_stub_non_implemente(self, mod, tmp_path): - # Le swap réel est réservé révision humaine : le stub NE TOUCHE RIEN - # et signale qu'il n'est pas implémenté. - result = mod.apply_update( - {"target_version": "1.0.2", "update_type": "code-only", - "staged_zip": str(tmp_path / "x.zip")} - ) - assert result["applied"] is False - assert "human" in result["reason"].lower() or "supervis" in result["reason"].lower() +def _make_zip(path, entries): + """Fabrique un ZIP {nom: contenu} pour les tests.""" + import zipfile + with zipfile.ZipFile(path, "w") as zf: + for name, content in entries.items(): + zf.writestr(name, content) + return path - def test_write_boot_ok_marker_est_un_stub(self, mod): - result = mod.write_boot_ok_marker("1.0.2") - assert result["written"] is False + +class TestApplyUpdateArm: + def test_arme_extrait_et_pose_marqueur(self, mod, tmp_path): + app = tmp_path / "app"; app.mkdir() + z = _make_zip(tmp_path / "u.zip", {"main.py": "v2", "sub/x.py": "y"}) + res = mod.apply_update( + {"target_version": "1.0.2", "update_type": "code-only", "staged_zip": str(z)}, + app_dir=app, + ) + assert res["armed"] is True and res["applied"] is False + new_dir = app / "agent_v1_new" + assert (new_dir / "main.py").read_text() == "v2" + assert (new_dir / "sub" / "x.py").read_text() == "y" + import json as _j + data = _j.loads((app / "UPDATE_READY").read_text()) + assert data["target_version"] == "1.0.2" + assert data["update_type"] == "code-only" + + def test_ne_touche_pas_le_agent_v1_vivant(self, mod, tmp_path): + app = tmp_path / "app"; (app / "agent_v1").mkdir(parents=True) + live = app / "agent_v1" / "sentinelle.txt" + live.write_text("VERSION_VIVANTE") + z = _make_zip(tmp_path / "u.zip", {"main.py": "v2"}) + mod.apply_update( + {"target_version": "1.0.2", "update_type": "code-only", "staged_zip": str(z)}, + app_dir=app, + ) + assert live.read_text() == "VERSION_VIVANTE" # swap différé à Lea.bat + + def test_zip_introuvable_pas_de_crash_ni_marqueur(self, mod, tmp_path): + app = tmp_path / "app"; app.mkdir() + res = mod.apply_update( + {"target_version": "1.0.2", "update_type": "code-only", + "staged_zip": str(tmp_path / "absent.zip")}, + app_dir=app, + ) + assert res["armed"] is False and "error" in res + assert not (app / "UPDATE_READY").exists() + + def test_relance_nettoie_agent_v1_new_precedent(self, mod, tmp_path): + app = tmp_path / "app"; app.mkdir() + stale = app / "agent_v1_new"; stale.mkdir() + (stale / "vieux.txt").write_text("obsolete") + z = _make_zip(tmp_path / "u.zip", {"main.py": "v2"}) + mod.apply_update( + {"target_version": "1.0.3", "update_type": "code-only", "staged_zip": str(z)}, + app_dir=app, + ) + assert not (app / "agent_v1_new" / "vieux.txt").exists() + assert (app / "agent_v1_new" / "main.py").read_text() == "v2" + + def test_zip_slip_refuse(self, mod, tmp_path): + app = tmp_path / "app"; app.mkdir() + z = _make_zip(tmp_path / "evil.zip", {"../evil.py": "pwn"}) + res = mod.apply_update( + {"target_version": "1.0.2", "update_type": "code-only", "staged_zip": str(z)}, + app_dir=app, + ) + assert res["armed"] is False + assert not (app / "evil.py").exists() + + +class TestWriteBootOkMarker: + def test_ecrit_boot_ok_et_desarme_pending(self, mod, tmp_path): + app = tmp_path / "app"; app.mkdir() + (app / "PENDING_BOOT_1.0.2").write_text("x") + res = mod.write_boot_ok_marker("1.0.2", app_dir=app) + assert res["written"] is True + assert (app / "boot_ok_1.0.2").exists() + assert not (app / "PENDING_BOOT_1.0.2").exists() # --------------------------------------------------------------------------- @@ -272,12 +337,18 @@ class TestRunUpdateCycle: assert result["status"] == "up_to_date" assert list(tmp_path.glob("*.zip")) == [] - def test_maj_dispo_telecharge_en_staging_mais_ne_swappe_pas( + def test_maj_dispo_arme_le_swap_mais_ne_swappe_pas( self, mod, tmp_path, monkeypatch ): monkeypatch.setenv("RPA_AUTO_UPDATE_ENABLED", "true") - payload = b"PK\x03\x04 fake" + # payload = un VRAI ZIP (le download le stage, apply_update l'extrait) + import io, zipfile + buf = io.BytesIO() + with zipfile.ZipFile(buf, "w") as zf: + zf.writestr("main.py", "code v1.0.2") + payload = buf.getvalue() sha = hashlib.sha256(payload).hexdigest() + app = tmp_path / "app"; app.mkdir() result = mod.run_update_cycle( local_version="1.0.1", @@ -291,15 +362,16 @@ class TestRunUpdateCycle: "sha256": sha, }), downloader=lambda u: payload, + app_dir=app, ) - # Téléchargé + vérifié + STAGÉ, mais PAS appliqué (swap = stub humain). - assert result["status"] == "staged" + # Téléchargé + vérifié + ARMÉ (agent_v1_new + UPDATE_READY), mais PAS + # swappé : le remplacement atomique est fait par Lea.bat au reboot. + assert result["status"] == "armed" assert result["target_version"] == "1.0.2" assert result["sha256_verified"] is True - staged = Path(result["staged_zip"]) - assert staged.exists() and staged.parent == tmp_path - # Le swap est explicitement NON fait (réservé révision humaine). assert result["applied"] is False + assert (app / "UPDATE_READY").exists() + assert (app / "agent_v1_new" / "main.py").read_text() == "code v1.0.2" def test_sha256_mismatch_ne_stage_pas(self, mod, tmp_path, monkeypatch): monkeypatch.setenv("RPA_AUTO_UPDATE_ENABLED", "true")