337 lines
13 KiB
Python
337 lines
13 KiB
Python
"""TDD — DETTE-022 MAJ silencieuse v2 : NOYAU client de mise à jour Léa.
|
|
|
|
Périmètre testé (parties PURES / testables, GATED, OFF par défaut) :
|
|
- `parse_version` / `is_newer` côté client (R3, self-contained — le bundle
|
|
client n'embarque pas server_v1).
|
|
- `should_update(local_version, server_response)` : décision « faut-il
|
|
updater ? quelle version/type ? » à partir de la réponse serveur.
|
|
- `download_update(...)` via un `downloader` callable INJECTABLE : AUCUN
|
|
réseau réel en test. Vérifie le SHA256, écrit le ZIP dans le staging,
|
|
retourne un plan d'update — SANS toucher aux fichiers vivants.
|
|
- Flag `RPA_AUTO_UPDATE_ENABLED` (défaut OFF) : `auto_update_enabled()`.
|
|
|
|
HORS périmètre (réservé révision humaine — trop risqué pour un agent) :
|
|
swap réel des fichiers, édition Lea.bat, redémarrage. Le module expose des
|
|
STUBS explicites (`apply_update`, `write_boot_ok_marker`) marqués TODO.
|
|
|
|
Le module est chargé par chemin (importlib) pour ne dépendre d'aucun import
|
|
lourd du package client (cf. DETTE-013, comme test_agent_v1_log_shipper).
|
|
"""
|
|
|
|
import hashlib
|
|
import importlib.util
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
_MOD_PATH = (
|
|
Path(__file__).resolve().parents[2]
|
|
/ "agent_v0" / "agent_v1" / "network" / "updater.py"
|
|
)
|
|
|
|
|
|
def _load_module():
|
|
spec = importlib.util.spec_from_file_location("lea_updater", _MOD_PATH)
|
|
mod = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(mod)
|
|
return mod
|
|
|
|
|
|
@pytest.fixture
|
|
def mod():
|
|
return _load_module()
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# R3 — parse_version côté client (self-contained)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestClientParseVersion:
|
|
def test_ordre_semver(self, mod):
|
|
assert mod.parse_version("1.0.2") < mod.parse_version("1.0.10")
|
|
assert mod.is_newer("1.0.10", "1.0.2") is True
|
|
assert mod.is_newer("1.0.1", "1.0.1") is False
|
|
|
|
def test_tolerant_et_fallback(self, mod):
|
|
assert mod.parse_version("v1.2.3") == (1, 2, 3)
|
|
assert mod.parse_version("garbage") == (0,)
|
|
assert mod.parse_version(None) == (0,)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Flag RPA_AUTO_UPDATE_ENABLED — OFF par défaut
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestFlag:
|
|
def test_off_par_defaut(self, mod, monkeypatch):
|
|
monkeypatch.delenv("RPA_AUTO_UPDATE_ENABLED", raising=False)
|
|
assert mod.auto_update_enabled() is False
|
|
|
|
def test_on_si_active(self, mod, monkeypatch):
|
|
for val in ("true", "1", "yes", "on", "TRUE"):
|
|
monkeypatch.setenv("RPA_AUTO_UPDATE_ENABLED", val)
|
|
assert mod.auto_update_enabled() is True
|
|
|
|
def test_off_si_valeur_invalide(self, mod, monkeypatch):
|
|
monkeypatch.setenv("RPA_AUTO_UPDATE_ENABLED", "false")
|
|
assert mod.auto_update_enabled() is False
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# should_update — décision à partir de la réponse serveur
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestShouldUpdate:
|
|
def test_pas_de_maj_si_response_negative(self, mod):
|
|
plan = mod.should_update(
|
|
"1.0.1", {"update_available": False, "latest_version": "1.0.1"}
|
|
)
|
|
assert plan is None
|
|
|
|
def test_maj_si_serveur_propose_version_plus_recente(self, mod):
|
|
plan = mod.should_update(
|
|
"1.0.1",
|
|
{
|
|
"update_available": True,
|
|
"latest_version": "1.0.2",
|
|
"update_type": "code-only",
|
|
"url": "http://srv/api/fleet/download/pc-1?type=code-only&version=1.0.2",
|
|
},
|
|
)
|
|
assert plan is not None
|
|
assert plan["target_version"] == "1.0.2"
|
|
assert plan["update_type"] == "code-only"
|
|
|
|
def test_double_garde_pas_de_downgrade(self, mod):
|
|
# Même si le serveur dit update_available, le client revérifie semver :
|
|
# il ne descend JAMAIS vers une version <= locale (défense en profondeur).
|
|
plan = mod.should_update(
|
|
"1.0.5",
|
|
{"update_available": True, "latest_version": "1.0.2",
|
|
"update_type": "code-only", "url": "http://x"},
|
|
)
|
|
assert plan is None
|
|
|
|
def test_type_inconnu_normalise_code_only(self, mod):
|
|
plan = mod.should_update(
|
|
"1.0.1",
|
|
{"update_available": True, "latest_version": "1.0.2",
|
|
"update_type": "weird", "url": "http://x"},
|
|
)
|
|
assert plan["update_type"] == "code-only"
|
|
|
|
def test_response_malformee_pas_de_crash(self, mod):
|
|
assert mod.should_update("1.0.1", {}) is None
|
|
assert mod.should_update("1.0.1", None) is None
|
|
assert mod.should_update("1.0.1", {"update_available": True}) is None
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# download_update — downloader INJECTABLE, SHA256, aucun réseau réel
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestDownloadUpdate:
|
|
def test_telecharge_et_verifie_sha256_ok(self, mod, tmp_path):
|
|
payload = b"PK\x03\x04 fake zip bytes"
|
|
sha = hashlib.sha256(payload).hexdigest()
|
|
|
|
calls = {}
|
|
|
|
def fake_downloader(url):
|
|
calls["url"] = url
|
|
return payload
|
|
|
|
plan = {
|
|
"target_version": "1.0.2",
|
|
"update_type": "code-only",
|
|
"url": "http://srv/dl?version=1.0.2",
|
|
"sha256": sha,
|
|
}
|
|
result = mod.download_update(
|
|
plan, staging_dir=tmp_path, downloader=fake_downloader
|
|
)
|
|
|
|
assert result["ok"] is True
|
|
assert calls["url"] == "http://srv/dl?version=1.0.2"
|
|
# Le ZIP est écrit dans le staging (Lea_next-like), PAS dans les fichiers vivants.
|
|
staged = Path(result["staged_zip"])
|
|
assert staged.exists()
|
|
assert staged.read_bytes() == payload
|
|
assert staged.parent == tmp_path
|
|
|
|
def test_sha256_mismatch_rejette_et_nettoie(self, mod, tmp_path):
|
|
payload = b"corrupted"
|
|
|
|
def fake_downloader(url):
|
|
return payload
|
|
|
|
plan = {
|
|
"target_version": "1.0.2",
|
|
"update_type": "code-only",
|
|
"url": "http://x",
|
|
"sha256": "0" * 64, # ne correspond pas
|
|
}
|
|
result = mod.download_update(
|
|
plan, staging_dir=tmp_path, downloader=fake_downloader
|
|
)
|
|
assert result["ok"] is False
|
|
assert "sha256" in result["error"].lower()
|
|
# Aucun ZIP corrompu laissé dans le staging.
|
|
assert list(tmp_path.glob("*.zip")) == []
|
|
|
|
def test_sha256_absent_accepte_avec_avertissement(self, mod, tmp_path):
|
|
# Pas de sha256 fourni : best-effort, on accepte mais on le signale.
|
|
payload = b"PK no-sha"
|
|
|
|
plan = {
|
|
"target_version": "1.0.2",
|
|
"update_type": "code-only",
|
|
"url": "http://x",
|
|
}
|
|
result = mod.download_update(
|
|
plan, staging_dir=tmp_path, downloader=lambda u: payload
|
|
)
|
|
assert result["ok"] is True
|
|
assert result.get("sha256_verified") is False
|
|
|
|
def test_downloader_leve_pas_de_crash(self, mod, tmp_path):
|
|
def boom(url):
|
|
raise RuntimeError("réseau down")
|
|
|
|
plan = {"target_version": "1.0.2", "update_type": "code-only",
|
|
"url": "http://x", "sha256": "x"}
|
|
result = mod.download_update(plan, staging_dir=tmp_path, downloader=boom)
|
|
assert result["ok"] is False
|
|
assert "error" in result
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Stubs réservés à la révision humaine — DOIVENT être no-op explicites
|
|
# ---------------------------------------------------------------------------
|
|
|
|
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 test_write_boot_ok_marker_est_un_stub(self, mod):
|
|
result = mod.write_boot_ok_marker("1.0.2")
|
|
assert result["written"] is False
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# run_update_cycle — orchestrateur GATED (check → décide → stage → stub apply)
|
|
# AUCUN réseau réel, AUCUN swap réel : checker/downloader INJECTABLES, le swap
|
|
# reste un stub no-op (réservé révision humaine).
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestRunUpdateCycle:
|
|
def _checker(self, response):
|
|
"""Fabrique un checker injectable qui renvoie `response`."""
|
|
def _c(local_version, machine_id):
|
|
return response
|
|
return _c
|
|
|
|
def test_gate_off_ne_fait_rien(self, mod, tmp_path, monkeypatch):
|
|
# Flag OFF (défaut) : le cycle ne doit RIEN faire (pas d'appel réseau).
|
|
monkeypatch.delenv("RPA_AUTO_UPDATE_ENABLED", raising=False)
|
|
called = {"n": 0}
|
|
|
|
def _checker(local_version, machine_id):
|
|
called["n"] += 1
|
|
return {"update_available": True, "latest_version": "9.9.9",
|
|
"url": "http://x", "sha256": None}
|
|
|
|
result = mod.run_update_cycle(
|
|
local_version="1.0.1",
|
|
machine_id="pc-1",
|
|
staging_dir=tmp_path,
|
|
checker=_checker,
|
|
downloader=lambda u: b"x",
|
|
)
|
|
assert result["status"] == "disabled"
|
|
assert called["n"] == 0 # aucun appel réseau quand OFF
|
|
|
|
def test_a_jour_ne_stage_rien(self, mod, tmp_path, monkeypatch):
|
|
monkeypatch.setenv("RPA_AUTO_UPDATE_ENABLED", "true")
|
|
result = mod.run_update_cycle(
|
|
local_version="1.0.1",
|
|
machine_id="pc-1",
|
|
staging_dir=tmp_path,
|
|
checker=self._checker(
|
|
{"update_available": False, "latest_version": "1.0.1"}
|
|
),
|
|
downloader=lambda u: b"should-not-be-called",
|
|
)
|
|
assert result["status"] == "up_to_date"
|
|
assert list(tmp_path.glob("*.zip")) == []
|
|
|
|
def test_maj_dispo_telecharge_en_staging_mais_ne_swappe_pas(
|
|
self, mod, tmp_path, monkeypatch
|
|
):
|
|
monkeypatch.setenv("RPA_AUTO_UPDATE_ENABLED", "true")
|
|
payload = b"PK\x03\x04 fake"
|
|
sha = hashlib.sha256(payload).hexdigest()
|
|
|
|
result = mod.run_update_cycle(
|
|
local_version="1.0.1",
|
|
machine_id="pc-1",
|
|
staging_dir=tmp_path,
|
|
checker=self._checker({
|
|
"update_available": True,
|
|
"latest_version": "1.0.2",
|
|
"update_type": "code-only",
|
|
"url": "http://srv/dl?version=1.0.2",
|
|
"sha256": sha,
|
|
}),
|
|
downloader=lambda u: payload,
|
|
)
|
|
# Téléchargé + vérifié + STAGÉ, mais PAS appliqué (swap = stub humain).
|
|
assert result["status"] == "staged"
|
|
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
|
|
|
|
def test_sha256_mismatch_ne_stage_pas(self, mod, tmp_path, monkeypatch):
|
|
monkeypatch.setenv("RPA_AUTO_UPDATE_ENABLED", "true")
|
|
result = mod.run_update_cycle(
|
|
local_version="1.0.1",
|
|
machine_id="pc-1",
|
|
staging_dir=tmp_path,
|
|
checker=self._checker({
|
|
"update_available": True,
|
|
"latest_version": "1.0.2",
|
|
"update_type": "code-only",
|
|
"url": "http://x",
|
|
"sha256": "0" * 64,
|
|
}),
|
|
downloader=lambda u: b"corrupted",
|
|
)
|
|
assert result["status"] == "download_failed"
|
|
assert list(tmp_path.glob("*.zip")) == []
|
|
|
|
def test_checker_qui_leve_pas_de_crash(self, mod, tmp_path, monkeypatch):
|
|
monkeypatch.setenv("RPA_AUTO_UPDATE_ENABLED", "true")
|
|
|
|
def _boom(local_version, machine_id):
|
|
raise RuntimeError("serveur down / 503")
|
|
|
|
result = mod.run_update_cycle(
|
|
local_version="1.0.1",
|
|
machine_id="pc-1",
|
|
staging_dir=tmp_path,
|
|
checker=_boom,
|
|
downloader=lambda u: b"x",
|
|
)
|
|
# Best-effort : jamais d'exception ne remonte (ne casse pas Léa).
|
|
assert result["status"] == "check_failed"
|