feat(update): swap atomique + rollback (Lea.bat) + confirmation boot (main.py)
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) <noreply@anthropic.com>
This commit is contained in:
Dom
@@ -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")
|
||||
|
||||
Reference in New Issue
Block a user