"""Tests for the Phase-2 self-update logic. These tests exercise the pure logic in furtka/updater.py: version compare, sha256 verify, tarball extract, symlink swap, state writes. The service- restart + health-check paths are stubbed so tests don't talk to systemd or hit the network. FURTKA_ROOT + FURTKA_STATE_DIR + FURTKA_LOCK_PATH all override to tmp_path so each test gets a clean filesystem. """ import hashlib import io import tarfile from pathlib import Path import pytest @pytest.fixture def updater(tmp_path, monkeypatch): monkeypatch.setenv("FURTKA_ROOT", str(tmp_path / "opt_furtka")) monkeypatch.setenv("FURTKA_STATE_DIR", str(tmp_path / "var_lib_furtka")) monkeypatch.setenv("FURTKA_LOCK_PATH", str(tmp_path / "update.lock")) # Reload the module so the path constants pick up the env vars. import importlib import furtka.updater as u importlib.reload(u) return u def _make_tarball(path: Path, version: str): """Build a minimal valid Furtka release tarball at `path`.""" buf = io.BytesIO() with tarfile.open(fileobj=buf, mode="w:gz") as tf: for name, content in [ ("VERSION", f"{version}\n"), ("furtka/__init__.py", ""), ("apps/fileshare/manifest.json", "{}"), ]: data = content.encode() info = tarfile.TarInfo(name=name) info.size = len(data) tf.addfile(info, io.BytesIO(data)) path.write_bytes(buf.getvalue()) def test_version_tuple_orders_prereleases_before_stable(updater): vt = updater._version_tuple assert vt("26.0-alpha") < vt("26.0-beta") assert vt("26.0-beta") < vt("26.0-rc1") assert vt("26.0-rc1") < vt("26.0") assert vt("26.0") < vt("26.1-alpha") assert vt("26.1-alpha") < vt("27.0-alpha") def test_verify_tarball_accepts_matching_hash(tmp_path, updater): tar = tmp_path / "t.tar.gz" tar.write_bytes(b"hello world") sha = hashlib.sha256(b"hello world").hexdigest() updater.verify_tarball(tar, sha) # no raise def test_verify_tarball_rejects_mismatch(tmp_path, updater): tar = tmp_path / "t.tar.gz" tar.write_bytes(b"hello world") with pytest.raises(updater.UpdateError, match="sha256 mismatch"): updater.verify_tarball(tar, "0" * 64) def test_parse_sha256_sidecar_strips_filename(updater): line = "abc123 furtka-26.1-alpha.tar.gz\n" assert updater._parse_sha256_sidecar(line) == "abc123" def test_parse_sha256_sidecar_rejects_empty(updater): with pytest.raises(updater.UpdateError): updater._parse_sha256_sidecar("") def test_extract_tarball_returns_version_and_refuses_unsafe_paths(tmp_path, updater): tar = tmp_path / "t.tar.gz" _make_tarball(tar, "26.2-alpha") dest = tmp_path / "dest" assert updater._extract_tarball(tar, dest) == "26.2-alpha" assert (dest / "VERSION").read_text().strip() == "26.2-alpha" # Build a malicious tarball with a traversal path; must refuse. evil = tmp_path / "evil.tar.gz" buf = io.BytesIO() with tarfile.open(fileobj=buf, mode="w:gz") as tf: info = tarfile.TarInfo(name="../escape") info.size = 0 tf.addfile(info, io.BytesIO(b"")) evil.write_bytes(buf.getvalue()) with pytest.raises(updater.UpdateError, match="refusing"): updater._extract_tarball(evil, tmp_path / "dest2") def test_write_and_read_state_round_trip(updater): updater.write_state("downloading", latest="26.2-alpha") s = updater.read_state() assert s["stage"] == "downloading" assert s["latest"] == "26.2-alpha" assert "updated_at" in s def test_apply_update_happy_path(tmp_path, updater, monkeypatch): # Set up an existing "26.0-alpha" current symlink so apply_update has # something to swap out. versions = updater.versions_dir() versions.mkdir(parents=True) (versions / "26.0-alpha").mkdir() (versions / "26.0-alpha" / "VERSION").write_text("26.0-alpha\n") current = updater.current_symlink() current.symlink_to(versions / "26.0-alpha") tar = tmp_path / "t.tar.gz" _make_tarball(tar, "26.1-alpha") # Stub the shell-out + health check — both succeed. monkeypatch.setattr(updater, "_run", lambda cmd: None) monkeypatch.setattr(updater, "_health_check", lambda url, deadline_s=30.0: True) updater.apply_update(tar, "26.1-alpha") assert current.resolve() == (versions / "26.1-alpha").resolve() assert (versions / "26.1-alpha" / "VERSION").read_text().strip() == "26.1-alpha" state = updater.read_state() assert state["stage"] == "done" assert state["version"] == "26.1-alpha" def test_apply_update_rolls_back_on_health_check_failure(tmp_path, updater, monkeypatch): versions = updater.versions_dir() versions.mkdir(parents=True) (versions / "26.0-alpha").mkdir() (versions / "26.0-alpha" / "VERSION").write_text("26.0-alpha\n") current = updater.current_symlink() current.symlink_to(versions / "26.0-alpha") tar = tmp_path / "t.tar.gz" _make_tarball(tar, "26.1-alpha") # _run succeeds (restart "works"), but the API never comes back healthy. monkeypatch.setattr(updater, "_run", lambda cmd: None) monkeypatch.setattr(updater, "_health_check", lambda url, deadline_s=30.0: False) # The rollback path calls subprocess.run directly — stub that too so we # don't actually try to restart a real service. import subprocess monkeypatch.setattr(subprocess, "run", lambda *a, **kw: None) with pytest.raises(updater.UpdateError, match="rolled back"): updater.apply_update(tar, "26.1-alpha") # Symlink should point back at 26.0-alpha. assert current.resolve() == (versions / "26.0-alpha").resolve() state = updater.read_state() assert state["stage"] == "rolled_back" assert state["failed_version"] == "26.1-alpha" def test_apply_update_rejects_version_mismatch(tmp_path, updater, monkeypatch): versions = updater.versions_dir() versions.mkdir(parents=True) tar = tmp_path / "t.tar.gz" _make_tarball(tar, "26.1-alpha") with pytest.raises(updater.UpdateError, match="doesn't match expected"): updater.apply_update(tar, "26.2-alpha") def test_acquire_lock_prevents_concurrent_runs(tmp_path, updater): first = updater.acquire_lock() try: with pytest.raises(updater.UpdateError, match="already in progress"): updater.acquire_lock() finally: first.close() def test_read_current_version_falls_back_to_dev(updater): # No symlink, no VERSION — should be "dev" not raise. assert updater.read_current_version() == "dev" def test_rollback_flips_to_previous_slot(tmp_path, updater, monkeypatch): versions = updater.versions_dir() versions.mkdir(parents=True) for v in ("26.0-alpha", "26.1-alpha"): (versions / v).mkdir() (versions / v / "VERSION").write_text(f"{v}\n") current = updater.current_symlink() current.symlink_to(versions / "26.1-alpha") import subprocess monkeypatch.setattr(subprocess, "run", lambda *a, **kw: None) restored = updater.rollback() assert restored == "26.0-alpha" assert current.resolve() == (versions / "26.0-alpha").resolve() def test_check_update_queries_forgejo_and_compares(updater, monkeypatch): # Stub the API and the current-version read. monkeypatch.setattr(updater, "read_current_version", lambda: "26.0-alpha") monkeypatch.setattr( updater, "_forgejo_api", lambda path: { "tag_name": "26.1-alpha", "assets": [ { "name": "furtka-26.1-alpha.tar.gz", "browser_download_url": "https://x/t.tar.gz", }, { "name": "furtka-26.1-alpha.tar.gz.sha256", "browser_download_url": "https://x/t.tar.gz.sha256", }, ], }, ) check = updater.check_update() assert check.current == "26.0-alpha" assert check.latest == "26.1-alpha" assert check.update_available is True assert check.tarball_url == "https://x/t.tar.gz" assert check.sha256_url == "https://x/t.tar.gz.sha256" def test_check_update_reports_up_to_date_when_same_version(updater, monkeypatch): monkeypatch.setattr(updater, "read_current_version", lambda: "26.1-alpha") monkeypatch.setattr( updater, "_forgejo_api", lambda path: {"tag_name": "26.1-alpha", "assets": []}, ) check = updater.check_update() assert check.update_available is False