server-up/tests/test_paden.py

"""De mappen van Server Up verplaatsen.

Dit is het enige onderdeel dat gegevens van de gebruiker weggooit, dus de
volgorde is de hele zaak: kopiëren, controleren, en pas daarna opruimen. Een
verhuizing die halverwege strandt met het origineel al verwijderd is het ergste
dat hier kan gebeuren.
"""
import json
import os
import sys
from pathlib import Path

import pytest

WORTEL = Path(__file__).resolve().parent.parent
sys.path.insert(0, str(WORTEL / "server-up"))

from conftest import login


def _vul(d: Path, bestanden: dict[str, str]) -> Path:
    for naam, inhoud in bestanden.items():
        p = d / naam
        p.parent.mkdir(parents=True, exist_ok=True)
        p.write_text(inhoud, encoding="utf-8")
    return d


# ── Hoofdmap en afgeleide paden ──────────────────────────────────────────────

def test_afgeleide_paden(env):
    import core
    assert core.afgeleide_paden("/srv/serverup") == {
        "LIBRARY_DIR": "/srv/serverup/stacks",
        "DATA_DIR": "/srv/serverup/appdata",
        "BACKUP_DIR": "/srv/serverup/backups",
    }
    # Een afsluitende schuine streep hoort niet te verdubbelen.
    assert core.afgeleide_paden("/srv/serverup/")["DATA_DIR"] == "/srv/serverup/appdata"
    # Een relatief pad levert niets op in plaats van iets kapots.
    assert core.afgeleide_paden("srv/data") == {}
    assert core.afgeleide_paden("") == {}


def test_base_dir_stuurt_de_standaarden(monkeypatch, tmp_path):
    """De drie paden stonden hard op /opt/serverup/*, ook als je BASE_DIR
    verzette — precies waarom de interface /opt bleef voorstellen."""
    import importlib
    monkeypatch.setenv("BASE_DIR", "/srv/eigen")
    monkeypatch.setenv("SU_CONFIG", str(tmp_path / "c.json"))
    for naam in [n for n in sys.modules if n == "core" or n.startswith("core.")]:
        del sys.modules[naam]
    core = importlib.import_module("core")
    assert core.DEFAULTS["LIBRARY_DIR"] == "/srv/eigen/stacks"
    assert core.DEFAULTS["DATA_DIR"] == "/srv/eigen/appdata"
    assert core.DEFAULTS["BACKUP_DIR"] == "/srv/eigen/backups"


# ── Meten en verplaatsen ─────────────────────────────────────────────────────

def test_meet_telt_bestanden_en_bytes(env, tmp_path):
    from core import paden
    d = _vul(tmp_path / "bron", {"a.txt": "12345", "sub/b.txt": "678"})
    aantal, bytes_ = paden.meet(d)
    assert (aantal, bytes_) == (2, 8)
    assert paden.meet(tmp_path / "bestaat-niet") == (0, 0)


def test_verhuizen_kopieert_en_ruimt_daarna_op(env, tmp_path):
    from core import paden
    bron = _vul(tmp_path / "oud", {"a.txt": "hallo", "sub/b.txt": "wereld"})
    doel = tmp_path / "nieuw"

    stappen = []
    n = paden.verhuis(bron, doel, voortgang_fn=lambda g, t, l: stappen.append((g, t)))

    assert n == 2
    assert (doel / "a.txt").read_text() == "hallo"
    assert (doel / "sub" / "b.txt").read_text() == "wereld"
    assert not bron.exists(), "de bron hoort na een geslaagde verhuizing weg te zijn"
    assert stappen and stappen[-1][0] == stappen[-1][1], "voortgang loopt niet vol"


def test_de_bron_blijft_staan_als_de_controle_faalt(env, tmp_path, monkeypatch):
    """Als er onderweg iets misgaat mag het origineel niet verdwijnen."""
    from core import paden
    bron = _vul(tmp_path / "oud", {"a.txt": "hallo"})
    doel = tmp_path / "nieuw"

    # Doe alsof er minder is aangekomen dan verstuurd.
    echt = paden.meet

    def nep(p):
        return (0, 0) if Path(p) == doel.resolve() else echt(p)

    monkeypatch.setattr(paden, "meet", nep)
    with pytest.raises(paden.PadFout, match="Controle mislukt"):
        paden.verhuis(bron, doel)
    assert (bron / "a.txt").exists(), "de bron is weg terwijl de controle faalde"


def test_doel_binnen_de_bron_wordt_geweigerd(env, tmp_path):
    from core import paden
    bron = _vul(tmp_path / "oud", {"a.txt": "x"})
    with pytest.raises(paden.PadFout, match="binnen"):
        paden.verhuis(bron, bron / "erin")


def test_zelfde_map_doet_niets(env, tmp_path):
    from core import paden
    bron = _vul(tmp_path / "oud", {"a.txt": "x"})
    assert paden.verhuis(bron, bron) == 0
    assert (bron / "a.txt").exists()


def test_eigenaar_en_rechten_gaan_mee(env, tmp_path, monkeypatch):
    """`shutil.copy2` neemt de rechten mee maar niet de eigenaar.

    Server Up draait als root, dus zonder chown werd verhuisde appdata root:root
    en kwam een container die als PUID 1000 draait er niet meer in. De test kan
    zelf niet chownen (dat mag alleen root), dus we leggen vast wát er gevraagd
    wordt.
    """
    from core import paden
    bron = _vul(tmp_path / "oud", {"a.txt": "hallo", "sub/b.txt": "wereld"})
    os.chmod(bron / "a.txt", 0o640)
    doel = tmp_path / "nieuw"

    gevraagd = {}
    monkeypatch.setattr(paden.os, "chown",
                        lambda p, u, g, **kw: gevraagd.__setitem__(str(p), (u, g)))

    paden.verhuis(bron, doel)

    eigen = (os.getuid(), os.getgid())
    assert gevraagd.get(str(doel)) == eigen, "de hoofdmap zelf blijft van root"
    assert gevraagd.get(str(doel / "a.txt")) == eigen
    assert gevraagd.get(str(doel / "sub")) == eigen
    assert gevraagd.get(str(doel / "sub" / "b.txt")) == eigen
    # De rechten komen van copy2/copystat en horen ook te kloppen.
    assert os.stat(doel / "a.txt").st_mode & 0o777 == 0o640


def test_de_eigenaar_komt_van_de_bron(env, tmp_path, monkeypatch):
    """De uid/gid moet van het origineel komen, niet van wie er toevallig
    kopieert. Chownen kan alleen root, dus we leggen de vraag vast."""
    from core import paden
    bron = _vul(tmp_path, {"a.txt": "x"}) / "a.txt"
    doel = tmp_path / "b.txt"
    doel.write_text("x")

    st = os.stat(bron)
    # Een echte stat_result met een andere eigenaar, zodat copystat blijft werken.
    nep = os.stat_result((st.st_mode, st.st_ino, st.st_dev, st.st_nlink,
                          1000, 1000, st.st_size,
                          int(st.st_atime), int(st.st_mtime), int(st.st_ctime)))
    monkeypatch.setattr(paden.os, "stat", lambda p, **kw: nep)
    gevraagd = []
    monkeypatch.setattr(paden.os, "chown",
                        lambda p, u, g, **kw: gevraagd.append((u, g)))

    paden._neem_eigenaar_over(bron, doel, [])
    assert gevraagd == [(1000, 1000)]


def test_chown_die_niet_mag_stopt_de_verhuizing_niet(env, tmp_path, monkeypatch):
    """Draai je niet als root, dan is alles toch al van jou. Dat mag geen fout
    zijn — wel een melding, want dan kán het misgaan."""
    from core import paden
    bron = _vul(tmp_path / "oud", {"a.txt": "hallo"})
    doel = tmp_path / "nieuw"
    monkeypatch.setattr(paden.os, "chown",
                        lambda *a, **k: (_ for _ in ()).throw(PermissionError("mag niet")))

    regels = []
    n = paden.verhuis(bron, doel, log_fn=regels.append)

    assert n == 1 and (doel / "a.txt").read_text() == "hallo"
    assert any("eigenaar" in r for r in regels), "geen waarschuwing over de rechten"


def test_symlinks_blijven_symlinks(env, tmp_path):
    """Een link volgen zou de inhoud dupliceren of buiten de boom kunnen wijzen."""
    from core import paden
    bron = _vul(tmp_path / "oud", {"echt.txt": "inhoud"})
    os.symlink("echt.txt", bron / "link.txt")
    doel = tmp_path / "nieuw"
    paden.verhuis(bron, doel)
    assert (doel / "link.txt").is_symlink()
    assert os.readlink(doel / "link.txt") == "echt.txt"


# ── De stacks meeverhuizen ───────────────────────────────────────────────────

def test_paden_in_stacks_worden_omgeschreven(env, tmp_path):
    """Alleen bestanden verplaatsen laat elke stack naar zijn oude appdata
    wijzen — die dan leeg is."""
    from core import paden
    lib = tmp_path / "stacks"
    stack = lib / "sonarr"
    stack.mkdir(parents=True)
    (stack / "docker-compose.yml").write_text(
        "services:\n  sonarr:\n    volumes:\n"
        "      - /opt/serverup/appdata/sonarr:/config\n", encoding="utf-8")
    (stack / ".env").write_text("PAD=/opt/serverup/appdata/sonarr\n", encoding="utf-8")
    (stack / ".serverup.json").write_text(json.dumps(
        {"source": "sonarr", "values": {"appdata_dir": "/opt/serverup/appdata"}}))

    n = paden.schrijf_paden_om(lib, {"/opt/serverup/appdata": "/srv/serverup/appdata"})
    assert n == 3
    assert "/srv/serverup/appdata/sonarr:/config" in (stack / "docker-compose.yml").read_text()
    assert "/srv/serverup/appdata/sonarr" in (stack / ".env").read_text()
    meta = json.loads((stack / ".serverup.json").read_text())
    assert meta["values"]["appdata_dir"] == "/srv/serverup/appdata"


def test_een_pad_dat_er_op_lijkt_blijft_met_rust(env, tmp_path):
    """`/opt/serverup` vervangen mag `/opt/serverup-oud` niet aanraken."""
    from core import paden
    lib = tmp_path / "stacks"
    stack = lib / "app"
    stack.mkdir(parents=True)
    (stack / "docker-compose.yml").write_text(
        "services:\n  a:\n    volumes:\n"
        "      - /opt/serverup-oud/data:/config\n"
        "      - /opt/serverup/data:/data\n", encoding="utf-8")

    paden.schrijf_paden_om(lib, {"/opt/serverup": "/srv/serverup"})
    tekst = (stack / "docker-compose.yml").read_text()
    assert "/opt/serverup-oud/data:/config" in tekst, "de buurmap is meegenomen"
    assert "/srv/serverup/data:/data" in tekst


# ── Bereikbaarheid ───────────────────────────────────────────────────────────

def test_relatief_pad_wordt_geweigerd(env):
    from core import paden
    uit = paden.bereikbaar("relatief/pad")
    assert not uit["ok"] and uit["reden"] == "geen_absoluut_pad"


def test_onschrijfbaar_pad_wordt_gemeld(env, tmp_path, monkeypatch):
    """Buiten een container komt de mountcontrole er niet aan te pas.

    Dit expliciet zetten in plaats van op de omgeving vertrouwen: de CI draait
    zelf in een container, en daar sloeg deze test aan op `buiten_mount` zonder
    ooit bij de schrijftest te komen.
    """
    from core import paden
    monkeypatch.setattr(paden, "_in_container", lambda: False)
    monkeypatch.setattr(Path, "mkdir",
                        lambda *a, **k: (_ for _ in ()).throw(OSError("alleen-lezen")))
    uit = paden.bereikbaar(str(tmp_path / "kan-niet"))
    assert not uit["ok"] and uit["reden"] == "niet_schrijfbaar"
    assert "alleen-lezen" in uit["detail"]


def test_pad_buiten_de_mount_wordt_herkend(env, monkeypatch):
    """In de container bestaat /srv wel als pad, maar wat je erheen schrijft
    belandt in de overlay en niet op de host. Dat moet je vóóraf horen."""
    from core import paden
    monkeypatch.setattr(paden, "_in_container", lambda: True)
    monkeypatch.setattr(paden, "_mountpunten", lambda: ["/opt/serverup", "/data"])

    uit = paden.bereikbaar("/srv/serverup/appdata")
    assert not uit["ok"] and uit["reden"] == "buiten_mount"
    # De melding moet zeggen wát je in .env zet, niet alleen dát het misging.
    assert uit["voorstel_base"] == "/srv/serverup"

    # Wat wél gekoppeld is, komt gewoon door de mountcontrole heen.
    assert paden.bereikbaar("/opt/serverup/appdata").get("reden") != "buiten_mount"


def test_de_mount_wordt_niet_op_naam_verward(env, monkeypatch):
    """`/opt/serverup` gekoppeld maakt `/opt/serverup-oud` niet bereikbaar."""
    from core import paden
    monkeypatch.setattr(paden, "_in_container", lambda: True)
    monkeypatch.setattr(paden, "_mountpunten", lambda: ["/opt/serverup"])
    assert paden.bereikbaar("/opt/serverup-oud")["reden"] == "buiten_mount"


# ── Voortgang ────────────────────────────────────────────────────────────────

def test_voortgang_komt_door_de_api(client, env):
    """Bij tientallen gigabytes zegt "bezig…" niets; de balk komt hiervandaan."""
    from core import jobs
    login(client)
    jid, _q = jobs.create("test")
    jobs.voortgang(jid, 256, 1024, "bestand.txt")

    d = client.get(f"/api/jobs/{jid}").get_json()
    assert d["progress"] == {"done": 256, "total": 1024,
                             "label": "bestand.txt", "pct": 25}


def test_voortgang_deelt_niet_door_nul(env):
    from core import jobs
    jid, _q = jobs.create("test")
    jobs.voortgang(jid, 0, 0, "")
    assert jobs.stream(jid)["progress"]["pct"] == 0


# ── Het formulier stelt de ingestelde map voor ───────────────────────────────

def test_installatieformulier_gebruikt_de_ingestelde_appdata(client, env, monkeypatch):
    """266 sjablonen noemen /opt/serverup/appdata. Wat je ook instelde, dáár
    kwam je data terecht.

    De testomgeving zet DATA_DIR als omgevingsvariabele, en die wint van het
    configuratiebestand — dus die verzetten we hier, niet de config.
    """
    from core import git
    csrf = login(client)
    monkeypatch.setenv("DATA_DIR", "/srv/eigen/appdata")

    bron = git.cache_dir("server-up") / "apps" / "testapp"
    (bron / "files").mkdir(parents=True)
    (bron / "template.json").write_text(json.dumps({
        "kind": "compose",
        "metadata": {"name": "Test", "description": "d", "tags": ["t"]},
        "variables": [{"title": "Algemeen", "items": [
            {"name": "service_name", "type": "str", "title": "Naam",
             "default": "testapp", "required": True},
            {"name": "appdata_dir", "type": "str", "title": "Appdata-map",
             "default": "/opt/serverup/appdata", "required": True},
        ]}],
    }), encoding="utf-8")
    (bron / "files" / "compose.yaml").write_text(
        "services:\n  << service_name >>:\n    image: x\n"
        "    volumes:\n      - << appdata_dir >>/<< service_name >>:/config\n",
        encoding="utf-8")

    d = client.post("/api/store/preview",
                    json={"stack": "testapp", "repo_id": "server-up"},
                    headers={"X-CSRF-Token": csrf}).get_json()
    veld = next(f for f in d["fields"] if f["name"] == "appdata_dir")
    assert veld["default"] == "/srv/eigen/appdata"


def test_puid_volgt_het_account_van_server_up(client, env):
    """44 sjablonen zetten PUID/PGID blind op 1000. Draait Server Up onder een
    ander account, dan schrijft de app zijn data weg als een gebruiker die
    Server Up daarna niet kan inpakken voor een backup."""
    from core import git
    csrf = login(client)

    bron = git.cache_dir("server-up") / "apps" / "lsioapp"
    (bron / "files").mkdir(parents=True)
    (bron / "template.json").write_text(json.dumps({
        "kind": "compose",
        "metadata": {"name": "Lsio", "description": "d", "tags": ["t"]},
        "variables": [{"title": "Algemeen", "items": [
            {"name": "service_name", "type": "str", "title": "Naam",
             "default": "lsioapp", "required": True},
            {"name": "puid", "type": "int", "title": "PUID", "default": 1000},
            {"name": "pgid", "type": "int", "title": "PGID", "default": 1000},
        ]}],
    }), encoding="utf-8")
    (bron / "files" / "compose.yaml").write_text(
        "services:\n  << service_name >>:\n    image: x\n"
        "    environment:\n      - PUID=<< puid >>\n      - PGID=<< pgid >>\n",
        encoding="utf-8")

    d = client.post("/api/store/preview",
                    json={"stack": "lsioapp", "repo_id": "server-up"},
                    headers={"X-CSRF-Token": csrf}).get_json()
    velden = {f["name"]: f for f in d["fields"]}
    assert velden["puid"]["default"] == os.getuid()
    assert velden["pgid"]["default"] == os.getgid()