"""Zelf-update: image-afleiding, veiligheidscontroles en de tag-administratie.""" import sys from pathlib import Path import pytest sys.path.insert(0, str(Path(__file__).resolve().parent.parent / "server-up")) from conftest import login @pytest.fixture def su(env): from core import selfupdate return selfupdate def _fake_inspect(su, monkeypatch, **overrides): info = {"name": "server-up", "image": "git.example.com/bes-r/server-up:0.5.00", "image_id": "sha256:abc", "project": "server-up", "working_dir": "/opt/server-up", "config_files": "/opt/server-up/docker-compose.yml", "service": "server-up"} info.update(overrides) monkeypatch.setattr(su, "inspect_self", lambda: info) return info @pytest.mark.parametrize("image,verwacht", [ ("git.example.com/bes-r/server-up:0.5.00", "git.example.com/bes-r/server-up"), ("git.example.com/bes-r/server-up", "git.example.com/bes-r/server-up"), ("server-up:latest", "server-up"), # Poortnummer in de registryhost mag niet als tag gelezen worden. ("git.example.com:3000/bes-r/server-up", "git.example.com:3000/bes-r/server-up"), ("git.example.com:3000/bes-r/server-up:1.2.3", "git.example.com:3000/bes-r/server-up"), ]) def test_image_repo_afleiding(su, monkeypatch, image, verwacht): _fake_inspect(su, monkeypatch, image=image) assert su.image_repo() == verwacht def test_expliciet_ingesteld_image_wint(su, env, monkeypatch): _fake_inspect(su, monkeypatch) env["core"].patch({"UPDATE_IMAGE": "registry.intern/team/server-up"}) assert su.image_repo() == "registry.intern/team/server-up" def test_status_ok_bij_registry_image(su, monkeypatch): _fake_inspect(su, monkeypatch) st = su.status() assert st["can_apply"] is True assert st["image"] == "git.example.com/bes-r/server-up" def test_status_weigert_lokaal_gebouwd_image(su, monkeypatch): """Een zelfgebouwd image is geen storing maar de normale opzet bij een server die na een push zelf bouwt; de melding moet dat ook zeggen.""" _fake_inspect(su, monkeypatch, image="server-up:latest") st = su.status() assert st["can_apply"] is False assert st["mode"] == "local-build" assert "zelfgebouwd" in st["reason"] assert "deploy" in st["reason"] # wijst naar de Git-route assert "SU_IMAGE" in st["reason"] # én naar het alternatief def test_status_modus_bij_registry_image(su, monkeypatch): _fake_inspect(su, monkeypatch) assert su.status()["mode"] == "registry" @pytest.mark.parametrize("overrides,modus", [ ({"working_dir": ""}, "no-compose"), ]) def test_status_modi(su, monkeypatch, overrides, modus): _fake_inspect(su, monkeypatch, **overrides) assert su.status()["mode"] == modus def test_status_modus_zonder_container(su, monkeypatch): monkeypatch.setattr(su, "inspect_self", lambda: {}) assert su.status()["mode"] == "no-container" def test_status_weigert_zonder_compose_labels(su, monkeypatch): _fake_inspect(su, monkeypatch, working_dir="") st = su.status() assert st["can_apply"] is False assert "compose" in st["reason"] def test_status_zonder_container(su, monkeypatch): monkeypatch.setattr(su, "inspect_self", lambda: {}) st = su.status() assert st["can_apply"] is False assert "niet te vinden" in st["reason"] @pytest.mark.parametrize("tag", [ "../../etc", "0.5.0; rm -rf /", "-x", "", "a" * 200, "tag met spatie", ]) def test_pull_weigert_ongeldige_tag(su, monkeypatch, tag): _fake_inspect(su, monkeypatch) ok, msg = su.pull(tag) assert ok is False assert "Ongeldige tag" in msg # ── SU_TAG wegschrijven ────────────────────────────────────────────────────── # Dit gebeurde eerder in Python, hier in de container. Maar de installatiemap is # hier nergens gemount — er zijn maar drie mounts: de docker-socket, su-data en # BASE_DIR. Bijwerken vanuit de interface kon daardoor nooit werken. Het gebeurt # nu in de helper, die de werkmap wél mount, dus toetsen we het shellfragment. def _draai_tagscript(su, map_: Path, tag: str): import subprocess r = subprocess.run(["sh", "-c", su._tag_script(tag)], cwd=str(map_), capture_output=True, text=True, timeout=30) assert r.returncode == 0, r.stderr return (map_ / ".env").read_text(encoding="utf-8") def test_tagscript_maakt_env_aan(su, tmp_path): assert _draai_tagscript(su, tmp_path, "0.5.10").strip() == "SU_TAG=0.5.10" def test_tagscript_vervangt_bestaande_regel(su, tmp_path): (tmp_path / ".env").write_text("BIND=0.0.0.0\nSU_TAG=0.5.00\nPORT=5000\n") regels = _draai_tagscript(su, tmp_path, "0.5.10").splitlines() assert regels == ["BIND=0.0.0.0", "PORT=5000", "SU_TAG=0.5.10"] assert sum(1 for r in regels if r.startswith("SU_TAG=")) == 1 def test_tagscript_behoudt_andere_instellingen(su, tmp_path): (tmp_path / ".env").write_text("SU_IMAGE=git.example.com/bes-r/server-up\n") inhoud = _draai_tagscript(su, tmp_path, "0.5.10") assert "SU_IMAGE=git.example.com/bes-r/server-up" in inhoud assert "SU_TAG=0.5.10" in inhoud def test_tagscript_laat_zich_niet_uitbreiden(su, tmp_path): """De tag wordt al door _TAG_RE gefilterd, maar dit fragment belandt in een `sh -c` in een container met de docker-socket. Twee sloten op de deur.""" inhoud = _draai_tagscript(su, tmp_path, "1.0.0'; touch /tmp/su-inbraak; echo '") assert not Path("/tmp/su-inbraak").exists(), "de tag brak uit het commando" assert inhoud.count("SU_TAG=") == 1 def test_de_helper_zet_de_tag_en_mount_de_werkmap(su, monkeypatch): """Zonder de mount schrijft de helper in zijn eigen overlay, en zonder het tag-fragment start compose gewoon het oude image opnieuw.""" aanroepen = [] def nep_run(cmd, **kw): aanroepen.append(cmd) return type("R", (), {"returncode": 0, "stdout": "", "stderr": ""})() monkeypatch.setattr(su.docker, "_run", nep_run) su._spawn_helper("git.example.com/bes-r/server-up:0.5.10", "/opt/server-up", "server-up", "0.5.10") # De eerste aanroep ruimt een blijven hangen helper op; de tweede is de start. cmd = next(c for c in aanroepen if "run" in c) assert "-v" in cmd and "/opt/server-up:/opt/server-up" in cmd script = cmd[-1] assert "SU_TAG=0.5.10" in script, "de helper zet de tag niet" assert "docker compose up -d" in script def test_rollback_zonder_vorige_versie(su): ok, msg = su.rollback() assert ok is False assert "Geen vorige versie" in msg def test_apply_endpoint_weigert_zonder_registry(client, monkeypatch): from core import selfupdate csrf = login(client) monkeypatch.setattr(selfupdate, "inspect_self", lambda: {"image": "server-up:latest", "working_dir": "/opt/x", "service": "server-up"}) r = client.post("/api/update/apply", json={"tag": "0.5.10"}, headers={"X-CSRF-Token": csrf}) assert r.status_code == 400 assert "registry" in r.get_json()["msg"] def test_rollback_endpoint_weigert_zonder_vorige_tag(client, monkeypatch): from core import selfupdate csrf = login(client) monkeypatch.setattr(selfupdate, "inspect_self", lambda: {"image": "git.example.com/bes-r/server-up:0.5.00", "working_dir": "/opt/x", "service": "server-up"}) r = client.post("/api/update/rollback", headers={"X-CSRF-Token": csrf}) assert r.status_code == 400 assert "Geen vorige versie" in r.get_json()["msg"] def test_registry_token_lekt_niet_via_settings(client, env): csrf = login(client) client.put("/api/settings", json={"UPDATE_REGISTRY_USER": "bes-r", "UPDATE_REGISTRY_TOKEN": "geheim-package-token"}, headers={"X-CSRF-Token": csrf}) body = client.get("/api/settings").get_data(as_text=True) assert "geheim-package-token" not in body assert "has_update_registry_token" in body # Wel echt opgeslagen: assert env["core"].load()["UPDATE_REGISTRY_TOKEN"] == "geheim-package-token" def test_leeg_registry_token_wist_bestaande_niet(client, env): csrf = login(client) client.put("/api/settings", json={"UPDATE_REGISTRY_TOKEN": "blijf-staan"}, headers={"X-CSRF-Token": csrf}) client.put("/api/settings", json={"UPDATE_REGISTRY_TOKEN": ""}, headers={"X-CSRF-Token": csrf}) assert env["core"].load()["UPDATE_REGISTRY_TOKEN"] == "blijf-staan"