volumen/tests/test_installer.py

"""Tests for the first-run bootstrap installer (system and per-user)."""

from __future__ import annotations

import contextlib
import json
import re
import sys
from pathlib import Path
from unittest import mock

import pytest

from volumen import installer
from volumen.cli import main
from volumen.config import Config

SYSTEM_DIR = Path("/etc/volumen")
SYSTEM_USERS_FILE = Path("/var/lib/volumen/users.toml")


# --- 1. system_install produces the expected config substitutions -----------


class TestSystemInstall:
    def test_writes_config_with_expected_substitutions(
        self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
    ) -> None:
        config_path = tmp_path / "config.toml"
        content_dir = tmp_path / "posts"
        users_file = tmp_path / "users.toml"
        monkeypatch.setattr(installer, "_ensure_user", lambda name: False)
        result = installer.system_install(
            config_path=config_path,
            content_dir=content_dir,
            users_file=users_file,
            service_user="volumen",
            admin_password="TopSecret!",
        )
        assert result.config_path == config_path
        text = config_path.read_text()
        assert re.search(r'^host = "::1"', text, re.MULTILINE) is not None
        assert re.search(r'^env = "production"', text, re.MULTILINE) is not None
        assert re.search(r"^trust_proxy = true", text, re.MULTILINE) is not None
        assert re.search(r"^cookie_secure = true", text, re.MULTILINE) is not None
        assert f'content_dir = "{content_dir}"' in text
        assert f'users_file = "{users_file}"' in text
        assert content_dir.is_dir()
        assert (content_dir / "media").is_dir()

    def test_writes_valid_scrypt_password_hash(
        self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
    ) -> None:
        config_path = tmp_path / "config.toml"
        monkeypatch.setattr(installer, "_ensure_user", lambda name: False)
        result = installer.system_install(
            config_path=config_path,
            content_dir=tmp_path / "posts",
            users_file=tmp_path / "users.toml",
            admin_password="TopSecret!",
        )
        text = config_path.read_text()
        m = re.search(r'^password_hash = "(.+)"', text, re.MULTILINE)
        assert m is not None
        encoded = m.group(1)
        assert encoded == result.password_hash
        assert encoded.startswith("scrypt$16384$8$1$")
        # Verify against password.
        from volumen.password import verify_password

        assert verify_password("TopSecret!", encoded) is True

    def test_writes_long_hex_session_key(
        self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
    ) -> None:
        config_path = tmp_path / "config.toml"
        monkeypatch.setattr(installer, "_ensure_user", lambda name: False)
        result = installer.system_install(
            config_path=config_path,
            content_dir=tmp_path / "posts",
            users_file=tmp_path / "users.toml",
            admin_password="whatever",
        )
        text = config_path.read_text()
        m = re.search(r'^session_key = "([0-9a-f]+)"', text, re.MULTILINE)
        assert m is not None
        assert len(m.group(1)) >= 128  # 64 bytes hex-encoded
        assert m.group(1) == result.session_key

    def test_config_round_trips_through_config_load(
        self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
    ) -> None:
        config_path = tmp_path / "config.toml"
        posts = tmp_path / "posts"
        users_file = tmp_path / "users.toml"
        monkeypatch.setattr(installer, "_ensure_user", lambda name: False)
        installer.system_install(
            config_path=config_path,
            content_dir=posts,
            users_file=users_file,
            admin_password="TopSecret!",
        )
        loaded = Config.load(path=str(config_path))
        assert loaded.content_dir == str(posts)
        assert loaded.users_file == str(users_file)
        assert loaded.host == "::1"
        assert loaded.env == "production"
        assert loaded.trust_proxy is True
        assert loaded.cookie_secure is True
        assert loaded.admin["password_hash"].startswith("scrypt$")
        assert len(loaded.admin["session_key"]) >= 128

    def test_existing_config_is_not_overwritten_without_force(
        self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
    ) -> None:
        config_path = tmp_path / "config.toml"
        original = 'host = "1.2.3.4"\nport = 1234\n'
        config_path.write_text(original)
        monkeypatch.setattr(installer, "_ensure_user", lambda name: False)
        result = installer.system_install(
            config_path=config_path,
            content_dir=tmp_path / "posts",
            users_file=tmp_path / "users.toml",
            admin_password="TopSecret!",
        )
        # File unchanged.
        assert config_path.read_text() == original
        # Result carries a sentinel message and no fresh password_hash.
        assert any("leaving untouched" in m for m in result.messages)
        assert result.password_hash == ""

    def test_force_makes_backup_with_original_content(
        self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
    ) -> None:
        config_path = tmp_path / "config.toml"
        original = 'host = "1.2.3.4"\nport = 1234\n'
        config_path.write_text(original)
        monkeypatch.setattr(installer, "_ensure_user", lambda name: False)
        installer.system_install(
            config_path=config_path,
            content_dir=tmp_path / "posts",
            users_file=tmp_path / "users.toml",
            admin_password="TopSecret!",
            force=True,
        )
        backups = list(tmp_path.glob("config.toml.bak.*.toml"))
        assert len(backups) == 1
        assert backups[0].read_text() == original
        # New content must round-trip.
        loaded = Config.load(path=str(config_path))
        assert loaded.host == "::1"

    def test_ensure_user_only_runs_when_user_missing(
        self, monkeypatch: pytest.MonkeyPatch, tmp_path: Path
    ) -> None:
        getpwnam_log: list[str] = []
        run_log: list[list[str]] = []

        def fake_getpwnam(name: str) -> object:
            getpwnam_log.append(name)
            raise KeyError(name)

        def fake_run(cmd: list[str], *args: object, **kwargs: object) -> object:
            run_log.append(list(cmd))
            return object()

        # Redirect the home-dir creation away from /var/lib so the test
        # can run unprivileged without hitting real root.
        fake_home = tmp_path / "fakehome"
        monkeypatch.setattr(installer, "DEFAULT_SERVICE_HOME", fake_home)
        with monkeypatch.context() as m:
            m.setattr(installer.pwd, "getpwnam", fake_getpwnam)
            m.setattr(installer, "_is_root", lambda: True)
            m.setattr(installer.subprocess, "run", fake_run)
            created = installer._ensure_user("volumen")
        assert created is True
        assert getpwnam_log == ["volumen"]
        # useradd should fire exactly once.
        assert any(cmd[:1] == ["useradd"] for cmd in run_log)
        # Home dir was created at our redirected location.
        assert fake_home.is_dir()

        # If the user exists, no useradd is run.
        getpwnam_log.clear()
        run_log.clear()

        class _Pwnam:
            pw_uid = 1234

        with monkeypatch.context() as m:
            m.setattr(installer.pwd, "getpwnam", lambda name: _Pwnam())
            m.setattr(installer.subprocess, "run", fake_run)
            created = installer._ensure_user("volumen")
        assert created is False
        assert run_log == []

        # If not root, even a missing user doesn't trigger useradd.
        getpwnam_log.clear()
        run_log.clear()
        with monkeypatch.context() as m:
            m.setattr(installer.pwd, "getpwnam", fake_getpwnam)
            m.setattr(installer, "_is_root", lambda: False)
            m.setattr(installer.subprocess, "run", fake_run)
            created = installer._ensure_user("volumen")
        assert created is False
        assert run_log == []

    def test_chown_chmod_skipped_silently_when_not_root(
        self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
    ) -> None:
        config_path = tmp_path / "config.toml"
        posts = tmp_path / "posts"
        users_file = tmp_path / "users.toml"
        monkeypatch.setattr(installer, "_ensure_user", lambda name: False)
        # Pretend we're not root — the chown/chmod helpers must no-op rather
        # than raise, and no subprocess is spawned for useradd/systemctl.
        monkeypatch.setattr(installer, "_is_root", lambda: False)
        run_log: list[list[str]] = []
        monkeypatch.setattr(
            installer.subprocess,
            "run",
            lambda cmd, *a, **kw: run_log.append(list(cmd)) or object(),
        )
        result = installer.system_install(
            config_path=config_path,
            content_dir=posts,
            users_file=users_file,
            admin_password="TopSecret!",
        )
        assert config_path.is_file()
        assert posts.is_dir()
        # No subprocess calls touched the system at all.
        assert run_log == []
        # Chmod was applied as a no-op (mode unchanged).
        assert oct(config_path.stat().st_mode & 0o777) != oct(0o600)  # default
        # Result is a valid InstallResult.
        assert result.config_path == config_path

    def test_install_service_writes_systemd_unit_and_calls_systemctl(
        self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
    ) -> None:
        config_path = tmp_path / "config.toml"
        content_dir = tmp_path / "posts"
        users_file = tmp_path / "users.toml"
        systemd_unit = tmp_path / "volumen.service"
        cmd_log: list[list[str]] = []

        class _Pwnam:
            pw_uid = 1234

        class _Grnam:
            gr_gid = 1234

        def fake_getpwnam(name: str) -> object:
            return _Pwnam()

        def fake_getgrnam(name: str) -> object:
            return _Grnam()

        def fake_which(name: str) -> str | None:
            return "/usr/local/bin/volumen" if name == "volumen" else None

        def fake_run(cmd: list[str], *args: object, **kwargs: object) -> object:
            cmd_log.append(list(cmd))
            return object()

        monkeypatch.setattr(installer, "_ensure_user", lambda name: False)
        monkeypatch.setattr(installer, "_is_root", lambda: True)
        monkeypatch.setattr(installer.pwd, "getpwnam", fake_getpwnam)
        monkeypatch.setattr(installer.grp, "getgrnam", fake_getgrnam)
        monkeypatch.setattr(installer.shutil, "which", fake_which)
        monkeypatch.setattr(installer.subprocess, "run", fake_run)
        monkeypatch.setattr(installer, "SYSTEMD_UNIT_PATH", systemd_unit)
        result = installer.system_install(
            config_path=config_path,
            content_dir=content_dir,
            users_file=users_file,
            admin_password="TopSecret!",
            install_service=True,
            enable_service=True,
        )
        assert systemd_unit.is_file()
        unit_text = systemd_unit.read_text()
        assert "/usr/local/bin/volumen" in unit_text
        assert "serve --config" in unit_text
        assert f"--config {config_path}" in unit_text
        assert f"--content {content_dir}" in unit_text
        assert "User=volumen" in unit_text
        assert "NoNewPrivileges=true" in unit_text
        assert "ProtectSystem=strict" in unit_text
        assert result.systemd_unit_path == systemd_unit
        assert ["systemctl", "daemon-reload"] in cmd_log
        assert ["systemctl", "enable", "--now", "volumen"] in cmd_log


# --- 2. user_install ---------------------------------------------------------


class TestUserInstall:
    def test_writes_config_under_user_path(
        self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
    ) -> None:
        config_path = tmp_path / "config.toml"
        content_dir = tmp_path / "posts"
        users_file = tmp_path / "users.toml"
        monkeypatch.setattr(installer, "_ensure_user", lambda name: False)
        result = installer.user_install(
            config_path=config_path,
            content_dir=content_dir,
            users_file=users_file,
            admin_password="TopSecret!",
        )
        assert result.service_user is None
        assert result.systemd_unit_path is None
        text = config_path.read_text()
        assert re.search(r'^env = "development"', text, re.MULTILINE) is not None
        assert re.search(r"^cookie_secure = false", text, re.MULTILINE) is not None
        assert config_path.exists()
        assert content_dir.is_dir()

    def test_rejects_install_service(self, tmp_path: Path) -> None:
        with pytest.raises(installer.InstallerError):
            installer.user_install(
                config_path=tmp_path / "config.toml",
                content_dir=tmp_path / "posts",
                users_file=tmp_path / "users.toml",
                admin_password="x",
                install_service=True,
            )


# --- 3. _mutate_config_text --------------------------------------------------


class TestMutateConfigText:
    def test_substitutes_six_keys_without_disturbing_comments(self) -> None:
        text = (
            'host = "::"\n'
            'env = "development"\n'
            "trust_proxy = false\n"
            "cookie_secure = false\n"
            'content_dir = "/old"\n'
            'users_file = "/old/users.toml"\n'
            'password_hash = ""\n'
            'session_key = ""\n'
        )
        out = installer._mutate_config_text(
            text,
            host="::1",
            env="production",
            trust_proxy=True,
            cookie_secure=True,
            content_dir="/new",
            users_file="/new/users.toml",
        )
        assert 'host = "::1"' in out
        assert 'env = "production"' in out
        assert "trust_proxy = true" in out
        assert "cookie_secure = true" in out
        assert 'content_dir = "/new"' in out
        assert 'users_file = "/new/users.toml"' in out

    def test_injects_password_hash_and_session_key(self) -> None:
        text = 'host = "::"\npassword_hash = ""\nsession_key = ""\n'
        out = installer._mutate_config_text(
            text,
            password_hash="scrypt$abc",
            session_key="deadbeef" * 16,
        )
        assert 'password_hash = "scrypt$abc"' in out
        assert 'session_key = "deadbeef"' * 4 not in out  # sentinel
        assert 'session_key = "' in out


# --- 4. CLI status -----------------------------------------------------------


class TestStatus:
    def test_status_json_reports_all_fields(
        self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str]
    ) -> None:
        from volumen import installer as inst_mod

        # Pre-create a config so the inspector finds something.
        posts = tmp_path / "posts"
        users_file = tmp_path / "users.toml"
        posts.mkdir()
        users_file.touch()
        config_path = tmp_path / "config.toml"
        monkeypatch.setattr(inst_mod, "_ensure_user", lambda name: False)
        installer.system_install(
            config_path=config_path,
            content_dir=posts,
            users_file=users_file,
            admin_password="TopSecret!",
        )
        monkeypatch.setattr(inst_mod, "SYSTEMD_UNIT_PATH", tmp_path / "volumen.service")
        # Force systemctl to a known "unknown" branch.
        monkeypatch.setattr(inst_mod.shutil, "which", lambda name: None)
        # Inject a users.toml entry for "admin".
        users_file.write_text(
            '[[users]]\nusername = "admin"\npassword_hash = "scrypt$x"\nrole = "admin"\n'
        )
        with (
            mock.patch.object(
                sys,
                "argv",
                [
                    "volumen",
                    "status",
                    "--config",
                    str(config_path),
                    "--data",
                    str(posts),
                    "--users-file",
                    str(users_file),
                    "--json",
                ],
            ),
            contextlib.suppress(SystemExit),
        ):
            main()
        captured = capsys.readouterr()
        assert captured.out.strip().startswith("{")
        info = json.loads(captured.out)
        for key in (
            "config_path",
            "data_dir",
            "users_file",
            "config_exists",
            "data_dir_exists",
            "users_file_exists",
            "password_hash_set",
            "session_key_ok",
            "systemd_unit_installed",
            "service_active",
            "admin_user_present",
            "issues",
        ):
            assert key in info
        assert info["config_exists"] is True
        assert info["password_hash_set"] is True
        assert info["session_key_ok"] is True
        assert info["admin_user_present"] is True


class TestCLISystemdLocalRefuse:
    def test_local_with_systemd_exits_nonzero(self, capsys: pytest.CaptureFixture[str]) -> None:
        exit_code = 0
        with mock.patch.object(
            sys,
            "argv",
            ["volumen", "init", "--local", "--systemd", "--admin-password", "x"],
        ):
            try:
                main()
            except SystemExit as exc:
                exit_code = exc.code if isinstance(exc.code, int) else 1
        assert exit_code == 2
        captured = capsys.readouterr()
        assert "--systemd" in captured.err