"""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