server-up/tests/test_install_script.py

"""Het installatiescript: syntaxis, opties en de dingen die stuk mogen gaan.

Een installer die je met `curl | sh` draait moet in elk geval syntactisch
kloppen onder de shells die distributies als /bin/sh gebruiken, en niet
stilzwijgend het verkeerde doen bij een typefout in de opties.
"""
import re
import shutil
import subprocess
from pathlib import Path

import pytest

WORTEL = Path(__file__).resolve().parent.parent
SCRIPT = WORTEL / "install.sh"


def _draai(*args, **kw):
    return subprocess.run(["sh", str(SCRIPT), *args], capture_output=True,
                          text=True, timeout=60, **kw)


def test_script_bestaat_en_is_uitvoerbaar():
    assert SCRIPT.is_file()
    assert SCRIPT.stat().st_mode & 0o111, "niet uitvoerbaar"


@pytest.mark.parametrize("shell", ["sh", "dash", "bash"])
def test_syntaxis_klopt_onder(shell):
    """Debian en Ubuntu gebruiken dash als /bin/sh; bashismen breken daar."""
    if not shutil.which(shell):
        pytest.skip(f"{shell} niet aanwezig")
    r = subprocess.run([shell, "-n", str(SCRIPT)], capture_output=True, text=True)
    assert r.returncode == 0, r.stderr


def test_geen_bashismen():
    tekst = SCRIPT.read_text(encoding="utf-8")
    for patroon, uitleg in [
        (r"\[\[", "[[ ]] is bash-only"),
        (r"^\s*declare ", "declare is bash-only"),
        (r"\$\{[A-Za-z_]+\[", "arrays zijn bash-only"),
        (r"==", "== in test is bash-only"),
    ]:
        treffers = [l for l in tekst.splitlines()
                    if re.search(patroon, l) and not l.strip().startswith("#")]
        assert not treffers, f"{uitleg}: {treffers[:2]}"


def test_help_werkt_en_noemt_alle_opties():
    r = _draai("--help")
    assert r.returncode == 0
    for optie in ("--dir", "--port", "--bind", "--branch", "--token",
                  "--update", "--uninstall", "--yes", "--dry-run",
                  "--admin", "--admin-password-file", "--create-admin"):
        assert optie in r.stdout, f"{optie} niet gedocumenteerd in --help"


def test_onbekende_optie_stopt():
    r = _draai("--bestaatniet")
    assert r.returncode != 0
    assert "Onbekende optie" in r.stderr


@pytest.mark.parametrize("poort", ["abc", "", "80a", "-1"])
def test_ongeldige_poort_stopt(poort):
    r = _draai("--port", poort)
    assert r.returncode != 0
    # Een lege waarde valt al af bij het uitlezen van de optie, de rest bij de
    # controle daarna; beide meldingen gaan over de poort.
    assert "poort" in r.stderr.lower() or "port" in r.stderr.lower()


def test_update_zonder_installatie_zegt_dat(tmp_path):
    r = _draai("--update", "--dir", str(tmp_path / "bestaatniet"))
    assert r.returncode != 0
    assert "Geen installatie gevonden" in r.stderr


def test_dry_run_wijzigt_niets(tmp_path):
    doel = tmp_path / "server-up"
    r = _draai("--dry-run", "--dir", str(doel), "--yes")
    assert not doel.exists(), "proefdraai heeft toch iets aangemaakt"
    assert "[droog]" in r.stdout


def test_dry_run_toont_de_hele_gang_van_zaken(tmp_path):
    """Ook op een machine zonder Docker moet je de voorvertoning kunnen zien."""
    r = _draai("--dry-run", "--dir", str(tmp_path / "su"), "--yes")
    for kop in ("Systeem controleren", "Broncode ophalen", "Instellingen",
                "Bouwen en starten"):
        assert kop in r.stdout, f"stap ontbreekt: {kop}"


def _pad_zonder(tmp_path, *weglaten):
    """Een PATH met alles erin behalve de genoemde commando's."""
    nep = tmp_path / "bin"
    nep.mkdir()
    import os
    for map_ in os.environ.get("PATH", "").split(os.pathsep):
        p = Path(map_)
        if not p.is_dir():
            continue
        for f in p.iterdir():
            if f.name in weglaten or (nep / f.name).exists():
                continue
            try:
                (nep / f.name).symlink_to(f)
            except OSError:
                pass
    return str(nep)


def test_proefdraai_werkt_zonder_curl_en_tar(tmp_path):
    """De CI-runner draait in python:3.12-slim, waar curl ontbreekt. Een
    proefdraai hoort dan nog steeds de hele voorvertoning te tonen: dat is juist
    het moment waarop je wil zien wat er zou gebeuren."""
    r = subprocess.run(["sh", str(SCRIPT), "--dry-run", "--dir",
                        str(tmp_path / "su"), "--yes"],
                       capture_output=True, text=True, timeout=60,
                       env={"PATH": _pad_zonder(tmp_path, "curl", "tar"),
                            "HOME": str(tmp_path)})
    assert r.returncode == 0, r.stderr
    assert "[droog]" in r.stdout
    assert "Bouwen en starten" in r.stdout
    assert not (tmp_path / "su").exists()


# ── Beheerdersaccount ────────────────────────────────────────────────────────

def test_wachtwoord_mag_niet_op_de_opdrachtregel():
    """Argumenten zijn voor elke gebruiker op de server zichtbaar met 'ps' en
    blijven in de shell-geschiedenis staan."""
    r = _draai("--admin-password", "geheimgeheim")
    assert r.returncode != 0
    assert "ps" in r.stderr


def test_create_admin_zonder_naam_stopt(tmp_path):
    r = _draai("--create-admin", "--dir", str(tmp_path), "--yes")
    assert r.returncode != 0
    assert "--admin" in r.stderr


def test_dry_run_toont_het_aanmaken_van_het_account(tmp_path):
    r = _draai("--dry-run", "--dir", str(tmp_path / "su"), "--yes", "--admin", "ramon")
    assert "/api/auth/setup" in r.stdout
    assert "ramon" in r.stdout


def test_wachtwoord_staat_niet_in_de_curl_argumenten():
    """Het wachtwoord gaat via stdin naar curl (--data-binary @-), niet als
    argument. Anders is het alsnog zichtbaar in 'ps'."""
    tekst = SCRIPT.read_text(encoding="utf-8")
    setup = tekst[tekst.index("maak_beheerder()"):]
    setup = setup[:setup.index("\nafsluiting()")]
    assert "--data-binary @-" in setup
    assert not re.search(r'-d\s+["\']?\{', setup), "JSON als curl-argument"


@pytest.mark.parametrize("adres", ["localhost", "0.0.0.0;rm -rf /", "mijn server"])
def test_ongeldig_bind_adres_stopt(adres):
    r = _draai("--bind", adres)
    assert r.returncode != 0
    assert "bind" in r.stderr.lower()


@pytest.mark.parametrize("adres", ["127.0.0.1", "0.0.0.0", "10.0.20.22", "::1"])
def test_geldig_bind_adres_wordt_geaccepteerd(adres, tmp_path):
    r = _draai("--dry-run", "--dir", str(tmp_path / "su"), "--yes", "--bind", adres)
    assert r.returncode == 0, r.stderr
    assert f"BIND={adres}" in r.stdout


# ── De vragen die het script stelt ───────────────────────────────────────────
# Interactief is de normale manier waarop iemand dit script draait, en juist
# daar bepaalt een verkeerd antwoord waar de server straks op luistert. Daarom
# worden de vragen hier echt gesteld, via een pseudo-terminal.

def _in_terminal(fragment, invoer, tmp_path, tijdslimiet=20):
    """Draait een stukje sh dat install.sh inlaadt, met een echte terminal."""
    import os
    import pty
    import re as _re
    import select
    import signal
    import time

    script = tmp_path / "fragment.sh"
    script.write_text(f"set --\nSU_INSTALL_ALLEEN_FUNCTIES=1 . {SCRIPT}\n{fragment}\n")

    pid, fd = pty.fork()
    if pid == 0:                                    # kindproces
        os.execvp("sh", ["sh", str(script)])
    os.write(fd, invoer.encode())
    uit, einde = b"", time.time() + tijdslimiet
    while time.time() < einde:
        klaar, _, _ = select.select([fd], [], [], 0.5)
        if klaar:
            try:
                blok = os.read(fd, 4096)
            except OSError:                         # terminal dicht: proces weg
                break
            if not blok:
                break
            uit += blok
        elif os.waitpid(pid, os.WNOHANG)[0]:
            break
    try:
        os.kill(pid, signal.SIGKILL)
    except ProcessLookupError:
        pass
    try:
        os.waitpid(pid, 0)
    except ChildProcessError:
        pass
    return _re.sub(r"\x1b\[[0-9;]*m", "", uit.decode(errors="replace").replace("\r", ""))


BIND_FRAGMENT = ('DROOG=0; JA=0; BIND_EXPLICIET=0; DOEL="$MAP"\n'
                 'kies_bind\n'
                 'printf "UITKOMST BIND=%s\\n" "$BIND"\n')


@pytest.mark.parametrize("invoer,verwacht", [
    ("1\n", "127.0.0.1"),
    ("2\n", "0.0.0.0"),
    ("3\n10.0.20.99\n", "10.0.20.99"),
    ("\n", "127.0.0.1"),          # enter zonder bestaande .env
])
def test_keuzemenu_voor_het_bind_adres(invoer, verwacht, tmp_path):
    leeg = tmp_path / "leeg"
    leeg.mkdir()
    uit = _in_terminal(f'MAP="{leeg}"\n' + BIND_FRAGMENT, invoer, tmp_path)
    assert f"UITKOMST BIND={verwacht}" in uit, uit


def test_enter_behoudt_de_bestaande_instelling(tmp_path):
    """Opnieuw installeren over een bestaande map mag je server niet ongemerkt
    terugzetten op loopback."""
    bestaand = tmp_path / "bestaand"
    bestaand.mkdir()
    (bestaand / ".env").write_text("BIND=10.0.20.5\nPORT=8080\n")
    uit = _in_terminal(f'MAP="{bestaand}"\n' + BIND_FRAGMENT, "\n", tmp_path)
    assert "UITKOMST BIND=10.0.20.5" in uit, uit
    assert "Nu ingesteld: 10.0.20.5" in uit


@pytest.mark.parametrize("invoer,melding", [
    ("9\n", "Onbekende keuze"),
    ("3\nlocalhost\n", "Ongeldig bind-adres"),
])
def test_menu_weigert_onzin(invoer, melding, tmp_path):
    leeg = tmp_path / "leeg"
    leeg.mkdir()
    uit = _in_terminal(f'MAP="{leeg}"\n' + BIND_FRAGMENT, invoer, tmp_path)
    assert melding in uit
    assert "UITKOMST" not in uit


WW_FRAGMENT = ('DROOG=0; JA=0; BEHEERDER=""\n'
               'beheerder_vooraf\n'
               'printf "UITKOMST naam=%s lengte=%s zelf=%s\\n" '
               '"$BEHEERDER" "${#BEHEERDER_WW}" "$WW_GEGENEREERD"\n')


def test_account_aanmaken_kan_geweigerd_worden(tmp_path):
    uit = _in_terminal(WW_FRAGMENT, "n\n", tmp_path)
    assert "UITKOMST naam= lengte=0 zelf=0" in uit


def test_wachtwoord_twee_keer_intypen(tmp_path):
    uit = _in_terminal(WW_FRAGMENT, "j\nramon\neenlangwachtwoord\neenlangwachtwoord\n", tmp_path)
    assert "UITKOMST naam=ramon lengte=17 zelf=0" in uit, uit


@pytest.mark.parametrize("invoer,melding", [
    ("j\nramon\neenlangwachtwoord\nietsanders\neenlangwachtwoord\neenlangwachtwoord\n",
     "verschillen"),
    ("j\nramon\nkort\nkort\neenlangwachtwoord\neenlangwachtwoord\n",
     "Minstens 10 tekens"),
])
def test_wachtwoord_mag_opnieuw_na_een_fout(invoer, melding, tmp_path):
    uit = _in_terminal(WW_FRAGMENT, invoer, tmp_path)
    assert melding in uit
    assert "UITKOMST naam=ramon lengte=17 zelf=0" in uit, uit


def test_na_drie_pogingen_maakt_het_script_er_zelf_een(tmp_path):
    """Blijven hangen op een prompt is erger dan een gegenereerd wachtwoord."""
    uit = _in_terminal(WW_FRAGMENT, "j\nramon\na\na\nb\nb\nc\nc\n", tmp_path)
    assert "UITKOMST naam=ramon lengte=24 zelf=1" in uit, uit


def test_lege_gebruikersnaam_slaat_over(tmp_path):
    uit = _in_terminal(WW_FRAGMENT, "j\n\n", tmp_path)
    assert "Geen naam opgegeven" in uit
    assert "UITKOMST naam= lengte=0 zelf=0" in uit


def test_proefdraai_vraagt_niets_en_gebruikt_geen_sudo(tmp_path):
    """Een proefdraai beloofde dat er niets gebeurt, maar riep wél 'sudo docker'
    aan. Op een server waar sudo een wachtwoord wil, liep hij daarop vast."""
    uit = _in_terminal(
        f'DROOG=1; SUDO="sudo"; DOEL="{tmp_path}"\ncontroleer_docker\nprintf "KLAAR\\n"\n',
        "", tmp_path, tijdslimiet=15)
    assert "KLAAR" in uit, f"controleer_docker liep vast:\n{uit}"
    assert "password" not in uit.lower()


# ── .env ─────────────────────────────────────────────────────────────────────

def test_bind_wordt_doorgevoerd_in_een_bestaande_env(tmp_path):
    """Eerder bleef een bestaande .env altijd ongemoeid, ook als je --bind
    meegaf. Dan leek de installatie te lukken maar bleef de server onbereikbaar
    op het oude adres."""
    doel = tmp_path / "su"
    doel.mkdir()
    (doel / ".env").write_text("BIND=127.0.0.1\nPORT=5000\nSU_TAG=0.7.30\n")
    r = _draai("--dry-run", "--dir", str(doel), "--yes", "--bind", "0.0.0.0")
    assert "BIND=0.0.0.0 in bestaande .env zetten" in r.stdout


def test_env_blijft_ongemoeid_zonder_expliciete_opties(tmp_path):
    """Anders zet een --update je instellingen elke keer terug."""
    doel = tmp_path / "su"
    doel.mkdir()
    (doel / ".env").write_text("BIND=0.0.0.0\nPORT=8080\n")
    r = _draai("--dry-run", "--dir", str(doel), "--yes")
    assert "ongemoeid" in r.stdout
    # En hij rekent verder met wat er écht staat, niet met de standaardwaarden.
    assert "BIND=0.0.0.0, PORT=8080" in r.stdout


def test_poort_uit_env_wordt_gebruikt_voor_het_slotadres(tmp_path):
    doel = tmp_path / "su"
    doel.mkdir()
    (doel / ".env").write_text("BIND=0.0.0.0\nPORT=8080\n")
    r = _draai("--dry-run", "--dir", str(doel), "--yes")
    assert ":8080" in r.stdout


# ── Tegen een echt draaiende server ──────────────────────────────────────────
# Deze tests starten de app zelf en laten het script er een account op
# aanmaken. Dat is het enige wat aantoont dat de JSON-opbouw, de poortdetectie
# en de foutafhandeling in samenhang kloppen.

geen_curl = pytest.mark.skipif(not shutil.which("curl"),
                               reason="curl ontbreekt (o.a. in python:3.12-slim)")


def _vrije_poort():
    import socket
    s = socket.socket()
    s.bind(("127.0.0.1", 0))
    poort = s.getsockname()[1]
    s.close()
    return poort


@pytest.fixture
def draaiende_app(tmp_path):
    """De echte app op een vrije poort, met een lege configuratie."""
    import os
    import sys
    import time
    import urllib.error
    import urllib.request

    poort = _vrije_poort()
    data = tmp_path / "data"
    data.mkdir()
    omgeving = dict(os.environ,
                    SU_CONFIG=str(data / "config.json"),
                    SU_AUDIT=str(data / "audit.db"),
                    SU_GIT_CACHE=str(data / "git"),
                    SU_SECRET=str(data / "secret.key"),
                    LIBRARY_DIR=str(data / "lib"),
                    DATA_DIR=str(tmp_path / "appdata"),
                    BACKUP_DIR=str(tmp_path / "backups"),
                    PORT=str(poort))
    proces = subprocess.Popen([sys.executable, str(WORTEL / "server-up" / "app.py")],
                              env=omgeving, stdout=subprocess.PIPE,
                              stderr=subprocess.STDOUT, text=True)
    try:
        for _ in range(80):
            if proces.poll() is not None:
                pytest.skip(f"app start niet: {proces.communicate()[0][-500:]}")
            try:
                urllib.request.urlopen(f"http://127.0.0.1:{poort}/healthz", timeout=1)
                break
            except (urllib.error.URLError, OSError):
                time.sleep(0.25)
        else:
            pytest.skip("app reageerde niet op tijd")
        yield poort
    finally:
        proces.terminate()
        try:
            proces.wait(timeout=10)
        except subprocess.TimeoutExpired:
            proces.kill()


def _inloggen(poort, naam, wachtwoord):
    import json
    import urllib.error
    import urllib.request
    verzoek = urllib.request.Request(
        f"http://127.0.0.1:{poort}/api/auth/login",
        data=json.dumps({"username": naam, "password": wachtwoord}).encode(),
        headers={"Content-Type": "application/json"})
    try:
        with urllib.request.urlopen(verzoek, timeout=5) as a:
            return a.status
    except urllib.error.HTTPError as e:
        return e.code


@geen_curl
def test_script_maakt_een_werkend_beheerdersaccount(draaiende_app, tmp_path):
    poort = draaiende_app
    inst = tmp_path / "inst"
    inst.mkdir()
    # De poort komt uit .env, niet uit een optie: zo werkt het ook als je
    # destijds met --port hebt geïnstalleerd.
    (inst / ".env").write_text(f"BIND=127.0.0.1\nPORT={poort}\n")
    ww = tmp_path / "ww"
    ww.write_text("een-heel-lang-wachtwoord\n")

    r = _draai("--create-admin", "--dir", str(inst), "--admin", "ramon",
               "--admin-password-file", str(ww), "--yes")
    assert r.returncode == 0, r.stderr
    assert "aangemaakt" in r.stdout
    assert _inloggen(poort, "ramon", "een-heel-lang-wachtwoord") == 200


@geen_curl
def test_wachtwoord_met_aanhalingstekens_en_backslashes(draaiende_app, tmp_path):
    """Het wachtwoord gaat als JSON over de lijn; zonder escapen levert dit een
    kapot verzoek op of — erger — een ander wachtwoord dan je denkt."""
    poort = draaiende_app
    inst = tmp_path / "inst"
    inst.mkdir()
    (inst / ".env").write_text(f"PORT={poort}\n")
    lastig = 'pa"ss\\wo/rd$`{}[]&|;<>#!\'x'
    ww = tmp_path / "ww"
    ww.write_text(lastig + "\n")

    r = _draai("--create-admin", "--dir", str(inst), "--admin", "lastig",
               "--admin-password-file", str(ww), "--yes")
    assert r.returncode == 0, r.stderr
    assert "aangemaakt" in r.stdout
    assert _inloggen(poort, "lastig", lastig) == 200


@geen_curl
def test_zelf_aangemaakt_wachtwoord_wordt_getoond_en_werkt(draaiende_app, tmp_path):
    """Zonder bestand, omgevingsvariabele of terminal maakt het script er zelf
    één aan. Die moet dan wel op het scherm komen, want hij staat nergens."""
    poort = draaiende_app
    inst = tmp_path / "inst"
    inst.mkdir()
    (inst / ".env").write_text(f"PORT={poort}\n")

    r = _draai("--create-admin", "--dir", str(inst), "--admin", "auto", "--yes")
    assert r.returncode == 0, r.stderr
    getoond = re.search(r"Wachtwoord: (\S+)", r.stdout)
    assert getoond, f"geen wachtwoord getoond in:\n{r.stdout}"
    assert _inloggen(poort, "auto", getoond.group(1)) == 200


@geen_curl
def test_tweede_account_wordt_netjes_geweigerd(draaiende_app, tmp_path):
    poort = draaiende_app
    inst = tmp_path / "inst"
    inst.mkdir()
    (inst / ".env").write_text(f"PORT={poort}\n")
    ww = tmp_path / "ww"
    ww.write_text("een-heel-lang-wachtwoord\n")
    _draai("--create-admin", "--dir", str(inst), "--admin", "eerste",
           "--admin-password-file", str(ww), "--yes")

    r = _draai("--create-admin", "--dir", str(inst), "--admin", "tweede",
               "--admin-password-file", str(ww), "--yes")
    assert r.returncode == 0, r.stderr
    assert "bestond al" in r.stdout
    assert _inloggen(poort, "tweede", "een-heel-lang-wachtwoord") != 200


@geen_curl
def test_onbereikbare_server_meldt_dat_zonder_te_ontsporen(tmp_path):
    inst = tmp_path / "inst"
    inst.mkdir()
    (inst / ".env").write_text(f"PORT={_vrije_poort()}\n")
    ww = tmp_path / "ww"
    ww.write_text("een-heel-lang-wachtwoord\n")

    r = _draai("--create-admin", "--dir", str(inst), "--admin", "ramon",
               "--admin-password-file", str(ww), "--yes")
    assert r.returncode == 0, r.stderr
    assert "niet bereikbaar" in r.stdout


def test_verwijst_naar_het_juiste_git_domein():
    tekst = SCRIPT.read_text(encoding="utf-8")
    assert 'GIT_HOST="git.ramonbesselink.nl"' in tekst


def test_compose_wordt_met_expliciet_bestand_aangeroepen():
    """`--project-directory` alleen laat compose het bestand nog in de huidige
    map zoeken; met `curl | sh` is dat je thuismap."""
    for regel in SCRIPT.read_text(encoding="utf-8").splitlines():
        if "docker compose" in regel and ("up -d" in regel or "down " in regel):
            assert "-f " in regel, f"geen expliciet compose-bestand: {regel.strip()}"


def test_readme_en_docs_noemen_dezelfde_installatieregel():
    regel = ("curl -fsSL https://git.ramonbesselink.nl/bes-r/server-up/"
             "raw/branch/main/install.sh")
    assert regel in (WORTEL / "README.md").read_text(encoding="utf-8")
    assert regel in (WORTEL / "docs" / "installeren.md").read_text(encoding="utf-8")