306 lines
12 KiB
Python
306 lines
12 KiB
Python
"""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
|
|
|
|
|
|
def test_pull_stript_de_v_van_een_forgejo_release(su, monkeypatch):
|
|
"""De workflow pusht :0.8.13-beta, niet :v0.8.13-beta."""
|
|
_fake_inspect(su, monkeypatch)
|
|
getrokken = []
|
|
monkeypatch.setattr(su, "_login", lambda log_fn=None: (True, ""))
|
|
monkeypatch.setattr(
|
|
su.docker, "_stream",
|
|
lambda cmd, **kw: getrokken.append(cmd[-1]) or 0,
|
|
)
|
|
ok, image = su.pull("v0.8.13-beta")
|
|
assert ok is True
|
|
assert image == "git.example.com/bes-r/server-up:0.8.13-beta"
|
|
assert getrokken == [image]
|
|
|
|
|
|
# ── SU_TAG wegschrijven ──────────────────────────────────────────────────────
|
|
# De draaiende container kan de deploy-.env niet schrijven: de installatiemap
|
|
# is daar niet gemount. De helper mount de werkmap wel en houdt tag en versie
|
|
# bij elkaar, dus toetsen we het shellfragment rechtstreeks.
|
|
|
|
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").splitlines() == [
|
|
"SU_TAG=0.5.10", "SU_VERSION=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\nSU_VERSION=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",
|
|
"SU_VERSION=0.5.10"]
|
|
assert sum(1 for r in regels if r.startswith("SU_TAG=")) == 1
|
|
assert sum(1 for r in regels if r.startswith("SU_VERSION=")) == 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
|
|
assert "SU_VERSION=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
|
|
assert inhoud.count("SU_VERSION=") == 1
|
|
|
|
|
|
def test_de_helper_zet_de_tag_en_mount_de_werkmap(su, monkeypatch):
|
|
"""De helper gebruikt dezelfde projectnaam en bestanden als de container."""
|
|
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",
|
|
project="eigen-project",
|
|
config_files="/opt/server-up/compose.yml,/opt/server-up/extra.yml",
|
|
)
|
|
# 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 "SU_VERSION=0.5.10" in script, "de helper zet de versie niet"
|
|
assert "--project-directory /opt/server-up" in script
|
|
assert "--project-name eigen-project" in script
|
|
assert "--file /opt/server-up/compose.yml" in script
|
|
assert "--file /opt/server-up/extra.yml" in script
|
|
assert "up -d --remove-orphans --no-build --pull never server-up" in script
|
|
|
|
|
|
def test_apply_endpoint_normaliseert_release_tag(client, monkeypatch):
|
|
"""Ook een oudere UI mag :v0.8.13 nooit aan Docker doorgeven."""
|
|
from core import jobs, selfupdate
|
|
csrf = login(client)
|
|
gezien = []
|
|
monkeypatch.setattr(selfupdate, "status", lambda: {
|
|
"can_apply": True, "reason": "", "previous_tag": "",
|
|
})
|
|
monkeypatch.setattr(
|
|
selfupdate, "apply",
|
|
lambda tag, log_fn=None: (gezien.append(tag) or True, "gestart"),
|
|
)
|
|
monkeypatch.setattr(jobs, "run", lambda fn, *args: fn(*args))
|
|
|
|
r = client.post("/api/update/apply", json={"tag": "v0.8.13-beta"},
|
|
headers={"X-CSRF-Token": csrf})
|
|
assert r.status_code == 200
|
|
assert r.get_json()["tag"] == "0.8.13-beta"
|
|
assert gezien == ["0.8.13-beta"]
|
|
|
|
|
|
def test_apply_onthoudt_versie_in_plaats_van_beweeglijke_alias(
|
|
su, env, monkeypatch):
|
|
_fake_inspect(su, monkeypatch, image="git.example.com/bes-r/server-up:beta",
|
|
version="0.8.12-beta")
|
|
monkeypatch.setattr(
|
|
su, "pull",
|
|
lambda tag, log_fn=None: (
|
|
True, "git.example.com/bes-r/server-up:0.8.13-beta"),
|
|
)
|
|
monkeypatch.setattr(su, "_spawn_helper", lambda *args, **kw: (True, "ok"))
|
|
|
|
ok, _ = su.apply("v0.8.13-beta")
|
|
assert ok is True
|
|
opgeslagen = env["core"].load()
|
|
assert opgeslagen["UPDATE_PREVIOUS_TAG"] == "0.8.12-beta"
|
|
assert opgeslagen["UPDATE_LAST_APPLIED"]["tag"] == "0.8.13-beta"
|
|
|
|
|
|
@pytest.mark.parametrize("workflow", [
|
|
".forgejo/workflows/deploy.yml",
|
|
".forgejo/workflows/deploy-prod.yml",
|
|
])
|
|
def test_deploy_schrijft_tag_en_versie_samen(workflow):
|
|
tekst = (Path(__file__).resolve().parent.parent / workflow).read_text()
|
|
assert "'/^SU_TAG=/d; /^SU_VERSION=/d'" in tekst
|
|
assert "SU_TAG=%s\\nSU_VERSION=%s\\n" in tekst
|
|
|
|
|
|
@pytest.mark.parametrize("workflow,pushregel", [
|
|
(".forgejo/workflows/build.yml", 'docker push "${IMAGE}:v${VERSION}"'),
|
|
(".forgejo/workflows/deploy-prod.yml",
|
|
'docker push "${SU_IMAGE}:v${VERSION}"'),
|
|
])
|
|
def test_release_publiceert_tag_voor_oude_updater(workflow, pushregel):
|
|
tekst = (Path(__file__).resolve().parent.parent / workflow).read_text()
|
|
assert pushregel in tekst
|
|
|
|
|
|
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"
|