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
def tagged():
    """A release tag on HEAD, with notes in the annotated tag object.

    Made and removed around the test rather than assumed: this repository has no
    tags until the release, and a test that needed one to exist would pass once
    and then start failing for the wrong reason.
    """
    import subprocess

    root = updates.checkout_dir()
    subprocess.run(
        ["git", "tag", "-a", "v9.9.9", "-m", "Ninth release.\n\n- one thing"],
        cwd=root, check=True, capture_output=True,
    )
    try:
        yield "v9.9.9"
    finally:
        subprocess.run(
            ["git", "tag", "-d", "v9.9.9"], cwd=root, check=False, capture_output=True
        )


@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(channel: str = ""):
    (settings.data_dir / updates.MARKER_NAME).write_text(channel)


# --- 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
    # `git describe`, so a tagged checkout says "1.0.0" and an untagged one says
    # a short sha rather than failing outright -- which is what `--always` is
    # there for, and is every repository before its first release.
    assert state.running


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


# --- Channels ---------------------------------------------------------------------
def test_only_a_release_shaped_tag_counts_as_one(db):
    """`v1.1.0-rc1` sorts *above* `v1.1.0` under git's version sort, so accepting
    a suffix would step a stable instance onto a release candidate on the
    strength of a hyphen."""
    for good in ("v1.0.0", "1.0.0", "v10.2.30"):
        assert updates.RELEASE_TAG.match(good), good
    for bad in ("v1.1.0-rc1", "v1.0", "nightly", "v1.0.0+build", "release-1"):
        assert not updates.RELEASE_TAG.match(bad), bad


def test_stable_picks_the_newest_release_and_reads_its_notes(db, tagged):
    """Notes travel inside the annotated tag object, so there is no forge API
    anywhere in this -- which matters, because the API this was checked against
    returns a 500 from a server-side panic."""
    root = updates.checkout_dir()

    assert updates.release_tags(root)[0] == "v9.9.9"
    target = updates.resolve_target(root, updates.CHANNEL_STABLE, "main")

    assert target.label == "9.9.9"
    assert "one thing" in target.notes


def test_a_signed_tag_shows_notes_and_not_base64(db):
    """`%(contents)` carries the signature block, and which header it uses
    depends on `gpg.format` -- PGP for `openpgp`, SSH for `ssh`. Stripping only
    the first would have rendered forty lines of base64 as the release notes on
    a repository that signs with an SSH key, which is this one."""
    import subprocess

    root = updates.checkout_dir()
    subprocess.run(
        ["git", "tag", "-a", "v9.9.8", "-m", "Signed release.\n\n- a note"],
        cwd=root, check=True, capture_output=True,
    )
    try:
        code, raw = updates._git(
            ["for-each-ref", "--format=%(contents)", "refs/tags/v9.9.8"], cwd=root
        )
        notes = updates._notes_for(root, "v9.9.8")
    finally:
        subprocess.run(
            ["git", "tag", "-d", "v9.9.8"], cwd=root, check=False, capture_output=True
        )

    assert code == 0
    # Only meaningful while this repository actually signs its tags; when it
    # does, the raw contents carry a block and the notes must not.
    if any(header in raw for header in updates._SIGNATURE_HEADERS):
        assert not any(header in notes for header in updates._SIGNATURE_HEADERS)
    assert notes == "Signed release.\n\n- a note"


def test_a_version_sort_is_not_a_lexical_one(db, tagged):
    """`v1.10.0` above `v1.9.0`, which a lexical sort gets wrong -- and gets
    wrong silently the first time a project reaches ten of anything."""
    import subprocess

    root = updates.checkout_dir()
    for tag in ("v1.9.0", "v1.10.0"):
        # `-m` rather than a lightweight tag: this repository sets
        # `tag.gpgSign`, which makes a bare `git tag <name>` behave as `-s` and
        # wait for an editor that a test does not have.
        subprocess.run(
            ["git", "tag", "-m", tag, tag], cwd=root, check=True, capture_output=True
        )
    try:
        tags = updates.release_tags(root)
        assert tags.index("v1.10.0") < tags.index("v1.9.0")
    finally:
        subprocess.run(
            ["git", "tag", "-d", "v1.9.0", "v1.10.0"], cwd=root, check=False,
            capture_output=True,
        )


def test_edge_follows_the_branch_tip(db):
    root = updates.checkout_dir()
    target = updates.resolve_target(root, updates.CHANNEL_EDGE, updates.branch_name())

    # A short sha rather than a version: the two channels answer "what version
    # is this?" with different kinds of thing, which is why the label is stored
    # rather than derived.
    assert target is None or len(target.label) == 7


def test_stable_with_nothing_tagged_says_so(db, monkeypatch):
    """"Nothing has been released" is not "up to date" -- the second would be
    true of an empty repository and useless."""
    monkeypatch.setattr(updates, "release_tags", lambda root: [])
    monkeypatch.setattr(updates, "channel_name", lambda: updates.CHANNEL_STABLE)

    state = updates.read()

    assert state.no_releases is True
    assert state.up_to_date is False


def test_a_tag_that_disagrees_with_the_version_is_flagged(db, tagged):
    """A tag cut before the version bump names a release nobody can identify
    afterwards. Costs no subprocess: both facts are already in hand."""
    state = updates.read()

    assert state.version_mismatch == "9.9.9"
    assert state.version != "9.9.9"


# --- 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_channel_and_branch_come_from_the_environment(db):
    """Deployment configuration rather than instance settings: they decide what
    code runs here, and a value a web administrator could edit would turn "you
    may deploy the channel" into "you may deploy anything"."""
    assert updates.branch_name() == settings.update_branch
    assert updates.channel_name() in updates.CHANNELS


def test_an_unknown_channel_falls_back_to_stable(db, monkeypatch):
    """A value stored by an older version, or a typo in lembas.env. Falling back
    to edge would silently start deploying unreleased work."""
    monkeypatch.setattr(settings, "update_channel", "nightly")

    assert updates.channel_name() == updates.CHANNEL_STABLE


def test_a_helper_installed_for_another_channel_is_named(db, client, registered):
    """The channel is declared in two places -- lembas.env and the systemd unit --
    and only the installer writes both. Edit one by hand and the button deploys
    something other than what the page has been naming all the way down."""
    _install_helper("stable" if updates.channel_name() == "edge" else "edge")

    state = updates.read()
    assert state.channel_mismatch is True
    assert "would deploy" in client.get("/admin/updates").text


def test_a_helper_that_did_not_say_claims_no_disagreement(db):
    """Every host installed before the marker carried a channel. Unknown is not
    a mismatch, and claiming one would be a red alert on every existing host."""
    _install_helper("")

    assert updates.helper_channel() == ""
    assert updates.read().channel_mismatch is False


def test_a_matching_helper_says_nothing(db, client, registered):
    _install_helper(updates.channel_name())

    assert updates.read().channel_mismatch is False
    assert "would deploy" not in client.get("/admin/updates").text


# --- 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_CHANNEL__" 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 . .")