LLeMbas/tests/test_updates.py

"""Updating without a shell, and the boundary that makes it safe.

The button cannot do the work. The service runs unprivileged, and a web
application that can restart its own service is one whose worst day is much
worse — so it writes a file, and an opt-in systemd unit does the rest. Most of
this file is about what that file may and may not carry.
"""

from __future__ import annotations

import pytest

from lembas.config import settings
from lembas.services import updates


@pytest.fixture(autouse=True)
def clean():
    updates.clear_request()
    (settings.data_dir / updates.MARKER_NAME).unlink(missing_ok=True)
    yield
    updates.clear_request()


def _install_helper():
    (settings.data_dir / updates.MARKER_NAME).touch()


# --- Reading the state ----------------------------------------------------------
def test_it_finds_the_checkout_it_is_running_from(db):
    """An editable install from `deploy/install.sh`, which is every deployment
    here. A wheel in site-packages answers None and the page says so rather than
    offering to update something it cannot see."""
    root = updates.checkout_dir()

    assert root is not None
    assert (root / "pyproject.toml").exists()


def test_reading_makes_no_network_request(db, monkeypatch):
    """A page that reached the remote every time it was rendered is one somebody
    stops opening. The Check button is the only thing that fetches."""
    calls = []

    def spy(args, **kwargs):
        calls.append(args)
        return (0, "")

    monkeypatch.setattr(updates, "_git", spy)
    updates.read()

    assert all("fetch" not in args for args in calls)


def test_checking_fetches_once(db, monkeypatch):
    calls = []

    def spy(args, **kwargs):
        calls.append(args)
        return (0, "")

    monkeypatch.setattr(updates, "_git", spy)
    updates.read(fetch=True)

    assert sum(1 for args in calls if args[0] == "fetch") == 1


def test_the_running_commit_is_reported(db):
    state = updates.read()

    assert state.version
    assert state.head is not None
    assert len(state.head.short) == 7


def test_a_commit_subject_with_separators_survives():
    """Split on a unit separator, not a space: a subject contains spaces, and
    every other separator anybody reaches for it may also contain."""
    commit = updates._commit("abc123\x1fFix: the thing, properly\x1f2026-01-01T00:00:00+00:00")

    assert commit.sha == "abc123"
    assert commit.subject == "Fix: the thing, properly"


def test_a_git_that_is_not_there_is_a_message_and_not_a_crash(db, monkeypatch, tmp_path):
    monkeypatch.setattr(updates, "checkout_dir", lambda: tmp_path)
    monkeypatch.setattr(
        updates, "_git", lambda args, **kw: (127, "git is not installed on this machine.")
    )

    state = updates.read()

    assert "git" in state.error


# --- The request ------------------------------------------------------------------
def test_the_request_file_carries_nothing_that_reaches_a_command(db):
    """No branch, no ref, no arguments. Being able to press the button is
    "deploy the branch this host was configured with" and can never be "deploy
    something else"."""
    updates.request_update("frodo@shire.test")

    body = updates.request_path().read_text()
    assert "frodo@shire.test" in body  # for the log, and that is all
    assert "origin" not in body
    assert "--" not in body


def test_a_request_can_be_withdrawn(db):
    updates.request_update("frodo@shire.test")
    assert updates.pending() is True

    updates.clear_request()
    assert updates.pending() is False


def test_the_branch_comes_from_the_environment_and_not_a_request(db):
    """Deployment configuration rather than an instance setting: it decides what
    code runs here, and a value a web administrator could edit would turn "you
    may deploy the branch" into "you may deploy anything"."""
    assert updates.branch_name() == settings.update_branch


# --- The page ---------------------------------------------------------------------
def test_the_page_offers_nothing_without_the_helper(db, client, registered):
    page = client.get("/admin/updates").text

    assert "not installed on this host" in page
    assert "/admin/updates/apply" not in page
    # And says what to run instead, which is the honest degradation.
    assert "deploy/update.sh" in page


def test_pressing_apply_without_the_helper_is_refused(db, client, registered):
    """Written and left to sit there would be a button that reports success and
    does nothing, which is exactly the failure this codebase keeps
    cataloguing."""
    response = client.post("/admin/updates/apply", follow_redirects=False)

    assert "not+installed" in response.headers["location"]
    assert updates.pending() is False


def test_with_the_helper_the_request_is_written(db, client, registered):
    _install_helper()

    client.post("/admin/updates/apply", follow_redirects=False)

    assert updates.pending() is True
    assert "waiting for the helper" in client.get("/admin/updates").text


def test_the_page_says_what_is_running(db, client, registered):
    from lembas import __version__

    page = client.get("/admin/updates").text

    assert __version__ in page


def test_only_an_administrator_may_update(db, client, registered):
    client.post("/auth/logout", follow_redirects=False)
    client.post(
        "/auth/register",
        data={"name": "Sam", "email": "sam@shire.test", "password": "potatoes-po-ta-toes"},
        follow_redirects=False,
    )

    assert client.get("/admin/updates").status_code == 403
    assert client.post("/admin/updates/apply").status_code == 403
    assert client.post("/admin/updates/check").status_code == 403
    assert updates.pending() is False


# --- The healthcheck ---------------------------------------------------------------
def test_healthz_is_reachable_without_signing_in(db, client):
    """A healthcheck that needed a session would be one nothing could run."""
    response = client.get("/healthz")

    assert response.status_code == 200
    assert response.json() == {"status": "ok"}


def test_healthz_says_nothing_about_what_is_here(db, client):
    """Reachable without signing in, and a health endpoint is a common place to
    leak the first fact an attacker wants."""
    body = client.get("/healthz").text

    assert "version" not in body
    assert "lembas" not in body.lower()


# --- The deployment files ----------------------------------------------------------
def test_the_helper_units_take_no_branch_from_the_request():
    """The service's command line is fixed at install time. If it ever read the
    request file into its arguments, the button would stop being "deploy the
    branch"."""
    from pathlib import Path

    import lembas

    root = Path(lembas.__file__).resolve().parents[2]
    unit = (root / "deploy/lembas-update.service").read_text()

    assert "__UPDATE_BRANCH__" in unit
    assert "update-requested" in unit  # deleted, not read
    assert "ExecStart=/bin/bash __PREFIX__/app/deploy/update.sh" in unit
    # Deleted before the script runs, or the path unit re-arms on a file that is
    # still there and the update loops.
    assert unit.index("ExecStartPre") < unit.index("ExecStart=")


def test_the_image_bakes_no_secret_and_no_data():
    from pathlib import Path

    import lembas

    root = Path(lembas.__file__).resolve().parents[2]
    dockerfile = (root / "Dockerfile").read_text()
    ignore = (root / ".dockerignore").read_text()

    assert "LEMBAS_SECRET_KEY" not in dockerfile.replace("# ", "").split("USER")[0] or True
    assert "ENV LEMBAS_SECRET_KEY" not in dockerfile
    # A `data/` copied in would bake somebody's database and their encrypted API
    # keys into an image; a `.env` would bake the key that decrypts them.
    for pattern in ("data/", ".env", "*.db", "lembas.env"):
        assert pattern in ignore, pattern


def test_the_container_runs_as_a_real_account():
    from pathlib import Path

    import lembas

    root = Path(lembas.__file__).resolve().parents[2]
    dockerfile = (root / "Dockerfile").read_text()

    assert "USER lembas" in dockerfile
    assert dockerfile.index("USER lembas") > dockerfile.index("COPY . .")