513 lines
19 KiB
Python
513 lines
19 KiB
Python
"""Tests for CLI entry point."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import contextlib
|
|
import json
|
|
import sys
|
|
import urllib.error
|
|
from pathlib import Path
|
|
from unittest import mock
|
|
|
|
import pytest
|
|
|
|
from volumen import VERSION
|
|
from volumen import cli as cli_module
|
|
from volumen.cli import main
|
|
|
|
|
|
class TestVersionCommand:
|
|
def test_version_output(self, capsys: pytest.CaptureFixture[str]) -> None:
|
|
with (
|
|
mock.patch.object(sys, "argv", ["volumen", "version"]),
|
|
contextlib.suppress(SystemExit),
|
|
):
|
|
main()
|
|
captured = capsys.readouterr()
|
|
assert VERSION in captured.out
|
|
|
|
|
|
class TestHelpCommand:
|
|
def test_help_output(self, capsys: pytest.CaptureFixture[str]) -> None:
|
|
with mock.patch.object(sys, "argv", ["volumen", "--help"]), contextlib.suppress(SystemExit):
|
|
main()
|
|
captured = capsys.readouterr()
|
|
assert "volumen" in captured.out or "usage" in captured.out.lower()
|
|
|
|
|
|
class TestUnknownCommand:
|
|
def test_unknown_command_exits(self) -> None:
|
|
with mock.patch.object(sys, "argv", ["volumen", "bogus"]), pytest.raises(SystemExit):
|
|
main()
|
|
|
|
|
|
class TestParseServeOptions:
|
|
def test_parse_serve_defaults(self) -> None:
|
|
with mock.patch.object(sys, "argv", ["volumen", "serve"]):
|
|
import argparse
|
|
|
|
parser = argparse.ArgumentParser()
|
|
sp = parser.add_subparsers(dest="command")
|
|
sp_serve = sp.add_parser("serve")
|
|
sp_serve.add_argument("--config", default="/etc/volumen/config.toml")
|
|
sp_serve.add_argument("--host", default=None)
|
|
sp_serve.add_argument("--port", type=int, default=None)
|
|
sp_serve.add_argument("--content", default=None)
|
|
args = parser.parse_args(["serve"])
|
|
assert args.config == "/etc/volumen/config.toml"
|
|
|
|
def test_parse_serve_overrides(self) -> None:
|
|
import argparse
|
|
|
|
parser = argparse.ArgumentParser()
|
|
sp = parser.add_subparsers(dest="command")
|
|
sp_serve = sp.add_parser("serve")
|
|
sp_serve.add_argument("--config", default="/etc/volumen/config.toml")
|
|
sp_serve.add_argument("--host", default=None)
|
|
sp_serve.add_argument("--port", type=int, default=None)
|
|
sp_serve.add_argument("--content", default=None)
|
|
args = parser.parse_args(
|
|
["serve", "--config", "/t.toml", "--host", "127.0.0.1", "--port", "9000"]
|
|
)
|
|
assert args.config == "/t.toml"
|
|
assert args.host == "127.0.0.1"
|
|
assert args.port == 9000
|
|
|
|
|
|
class TestPrepareConfig:
|
|
def test_prepare_config_writes_template(self, tmp_path: Path) -> None:
|
|
from volumen.config import Config
|
|
|
|
path = tmp_path / "config.toml"
|
|
result = Config.generate_default(str(path))
|
|
assert result == str(path)
|
|
assert path.exists()
|
|
|
|
def test_prepare_config_noops_when_present(self, tmp_path: Path) -> None:
|
|
from volumen.config import Config
|
|
|
|
path = tmp_path / "config.toml"
|
|
path.write_text("[server]\nport = 1\n")
|
|
result = Config.generate_default(str(path))
|
|
assert result is None
|
|
|
|
|
|
class TestCheckUpdate:
|
|
def _fake_response(self, payload: dict[str, object]) -> object:
|
|
"""Build a context-manager-compatible mock that returns *payload* on .read()."""
|
|
response = mock.MagicMock()
|
|
response.read.return_value = json.dumps(payload).encode("utf-8")
|
|
response.__enter__.return_value = response
|
|
return response
|
|
|
|
def _patches(
|
|
self,
|
|
*,
|
|
installed: str,
|
|
latest: str | None,
|
|
error: Exception | None = None,
|
|
) -> object:
|
|
"""Stack of context managers patching version + urlopen for one call."""
|
|
from contextlib import ExitStack
|
|
|
|
stack = ExitStack()
|
|
stack.enter_context(
|
|
mock.patch.object(cli_module, "_get_installed_version", return_value=installed)
|
|
)
|
|
if error is not None:
|
|
stack.enter_context(
|
|
mock.patch.object(cli_module.urllib.request, "urlopen", side_effect=error)
|
|
)
|
|
else:
|
|
stack.enter_context(
|
|
mock.patch.object(
|
|
cli_module.urllib.request,
|
|
"urlopen",
|
|
return_value=self._fake_response({"info": {"version": latest}}),
|
|
)
|
|
)
|
|
return stack
|
|
|
|
def test_update_available(self, capsys: pytest.CaptureFixture[str]) -> None:
|
|
args = argparse.Namespace(json=False)
|
|
with self._patches(installed="0.4.0", latest="0.5.0"):
|
|
rc = cli_module._check_update(args)
|
|
captured = capsys.readouterr()
|
|
assert rc == 1
|
|
assert "0.4.0" in captured.out
|
|
assert "0.5.0" in captured.out
|
|
assert "uv tool upgrade volumen" in captured.out
|
|
|
|
def test_up_to_date(self, capsys: pytest.CaptureFixture[str]) -> None:
|
|
args = argparse.Namespace(json=False)
|
|
with self._patches(installed="0.5.0", latest="0.5.0"):
|
|
rc = cli_module._check_update(args)
|
|
captured = capsys.readouterr()
|
|
assert rc == 0
|
|
assert "up to date" in captured.out
|
|
|
|
def test_local_ahead_of_remote(self, capsys: pytest.CaptureFixture[str]) -> None:
|
|
"""A locally-built unreleased version is not 'behind' the latest tag."""
|
|
args = argparse.Namespace(json=False)
|
|
with self._patches(installed="1.0.0a1", latest="0.9.0"):
|
|
rc = cli_module._check_update(args)
|
|
captured = capsys.readouterr()
|
|
assert rc == 0
|
|
assert "up to date" in captured.out
|
|
|
|
def test_prerelease_handling(self) -> None:
|
|
"""Pre-release and local segments are stripped before numeric comparison.
|
|
|
|
Implementation is intentionally conservative — ``0.5.0a1`` and ``0.5.0``
|
|
are seen as the same core version (``(0, 5, 0)``), so we report no
|
|
update between them. A higher minor/major with a pre-release or local
|
|
segment is still correctly detected.
|
|
"""
|
|
assert cli_module._compare_versions("0.5.0", "0.5.1a1+build") is True
|
|
assert cli_module._compare_versions("0.5.0+a", "0.5.0+a") is False
|
|
assert cli_module._compare_versions("0.5.0+a", "0.5.1") is True
|
|
|
|
def test_offline_returns_exit_code_2(self, capsys: pytest.CaptureFixture[str]) -> None:
|
|
args = argparse.Namespace(json=False)
|
|
with self._patches(
|
|
installed="0.4.0",
|
|
latest=None,
|
|
error=urllib.error.URLError("no internet"),
|
|
):
|
|
rc = cli_module._check_update(args)
|
|
captured = capsys.readouterr()
|
|
assert rc == 2
|
|
assert "could not reach" in captured.err
|
|
assert "0.4.0" in captured.err
|
|
|
|
def test_json_output_with_update_available(self, capsys: pytest.CaptureFixture[str]) -> None:
|
|
args = argparse.Namespace(json=True)
|
|
with self._patches(installed="0.4.0", latest="0.5.0"):
|
|
rc = cli_module._check_update(args)
|
|
captured = capsys.readouterr()
|
|
assert rc == 1
|
|
doc = json.loads(captured.out)
|
|
assert doc["current_version"] == "0.4.0"
|
|
assert doc["latest_version"] == "0.5.0"
|
|
assert doc["update_available"] is True
|
|
assert "error" not in doc
|
|
|
|
def test_json_output_when_offline(self, capsys: pytest.CaptureFixture[str]) -> None:
|
|
args = argparse.Namespace(json=True)
|
|
with self._patches(
|
|
installed="0.4.0",
|
|
latest=None,
|
|
error=urllib.error.URLError("dns failure"),
|
|
):
|
|
rc = cli_module._check_update(args)
|
|
captured = capsys.readouterr()
|
|
assert rc == 2
|
|
doc = json.loads(captured.out)
|
|
assert doc["current_version"] == "0.4.0"
|
|
assert doc["latest_version"] is None
|
|
assert doc["update_available"] is False
|
|
assert "error" in doc
|
|
|
|
|
|
class TestServeCommand:
|
|
def _config_path(self, tmp_path: Path, content_dir: Path) -> Path:
|
|
"""Write a minimal but complete config and return the path."""
|
|
config_path = tmp_path / "config.toml"
|
|
users_file = tmp_path / "users.toml"
|
|
config_path.write_text(
|
|
f'content_dir = "{content_dir}"\n'
|
|
f'users_file = "{users_file}"\n'
|
|
"[admin]\n"
|
|
'password_hash = ""\n'
|
|
)
|
|
return config_path
|
|
|
|
def test_serve_loads_config_and_invokes_uvicorn(
|
|
self, monkeypatch: pytest.MonkeyPatch, tmp_path: Path
|
|
) -> None:
|
|
content_dir = tmp_path / "posts"
|
|
content_dir.mkdir()
|
|
config_path = self._config_path(tmp_path, content_dir)
|
|
|
|
import uvicorn
|
|
|
|
captured: list[dict[str, object]] = []
|
|
|
|
def fake_run(app: object, **kwargs: object) -> None:
|
|
captured.append({"app": app, **kwargs})
|
|
|
|
monkeypatch.setattr(uvicorn, "run", fake_run)
|
|
|
|
args = argparse.Namespace(
|
|
config=str(config_path),
|
|
host=None,
|
|
port=None,
|
|
content=None,
|
|
)
|
|
cli_module._serve(args)
|
|
|
|
assert len(captured) == 1
|
|
call = captured[0]
|
|
assert call["host"] == "::"
|
|
assert call["port"] == 9090
|
|
assert call["log_level"] == "info"
|
|
from fastapi import FastAPI
|
|
|
|
assert isinstance(call["app"], FastAPI)
|
|
|
|
def test_serve_applies_host_and_port_overrides(
|
|
self, monkeypatch: pytest.MonkeyPatch, tmp_path: Path
|
|
) -> None:
|
|
content_dir = tmp_path / "posts"
|
|
content_dir.mkdir()
|
|
config_path = self._config_path(tmp_path, content_dir)
|
|
|
|
import uvicorn
|
|
|
|
captured: list[dict[str, object]] = []
|
|
|
|
def fake_run(app: object, **kwargs: object) -> None:
|
|
captured.append(kwargs)
|
|
|
|
monkeypatch.setattr(uvicorn, "run", fake_run)
|
|
|
|
args = argparse.Namespace(
|
|
config=str(config_path),
|
|
host="127.0.0.1",
|
|
port=9000,
|
|
content=None,
|
|
)
|
|
cli_module._serve(args)
|
|
|
|
assert captured[0]["host"] == "127.0.0.1"
|
|
assert captured[0]["port"] == 9000
|
|
|
|
|
|
class TestStatusCommand:
|
|
def test_status_human_output_with_issues(
|
|
self, monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str]
|
|
) -> None:
|
|
info = {
|
|
"config_path": "/etc/volumen/config.toml",
|
|
"data_dir": "/var/lib/volumen/posts",
|
|
"users_file": "/var/lib/volumen/users.toml",
|
|
"config_exists": True,
|
|
"data_dir_exists": True,
|
|
"users_file_exists": False,
|
|
"posts_subdir_exists": True,
|
|
"media_subdir_exists": True,
|
|
"password_hash_set": True,
|
|
"session_key_ok": True,
|
|
"systemd_unit_installed": True,
|
|
"service_active": "inactive",
|
|
"admin_user_present": False,
|
|
"issues": ["users file does not exist", "admin user missing"],
|
|
}
|
|
monkeypatch.setattr(cli_module, "_status", _FakeStatus(info))
|
|
with (
|
|
mock.patch.object(sys, "argv", ["volumen", "status"]),
|
|
pytest.raises(SystemExit) as exc,
|
|
):
|
|
main()
|
|
captured = capsys.readouterr()
|
|
assert exc.value.code == 1
|
|
assert "==>" in captured.out
|
|
assert "volumen status" in captured.out
|
|
assert "users file does not exist" in captured.out
|
|
assert "admin user missing" in captured.out
|
|
assert "Issues:" in captured.out
|
|
|
|
def test_status_returns_zero_when_no_issues(
|
|
self, monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str]
|
|
) -> None:
|
|
info = {
|
|
"config_path": "/etc/volumen/config.toml",
|
|
"data_dir": "/var/lib/volumen/posts",
|
|
"users_file": "/var/lib/volumen/users.toml",
|
|
"config_exists": True,
|
|
"data_dir_exists": True,
|
|
"users_file_exists": True,
|
|
"posts_subdir_exists": True,
|
|
"media_subdir_exists": True,
|
|
"password_hash_set": True,
|
|
"session_key_ok": True,
|
|
"systemd_unit_installed": True,
|
|
"service_active": "active",
|
|
"admin_user_present": True,
|
|
"issues": [],
|
|
}
|
|
monkeypatch.setattr(cli_module, "_status", _FakeStatus(info))
|
|
with (
|
|
mock.patch.object(sys, "argv", ["volumen", "status"]),
|
|
pytest.raises(SystemExit) as exc,
|
|
):
|
|
main()
|
|
captured = capsys.readouterr()
|
|
assert exc.value.code == 0
|
|
assert "All checks passed." in captured.out
|
|
|
|
|
|
class _FakeStatus:
|
|
"""Mock returning a static info dict from cli._status."""
|
|
|
|
def __init__(self, info: dict[str, object]) -> None:
|
|
self._info = info
|
|
|
|
def __call__(self, args: argparse.Namespace) -> int:
|
|
# Mirror the shape of the real _status(). For human mode print plain
|
|
# text; for --json print the info as JSON. Exit code 0 if no issues.
|
|
import json as _json
|
|
|
|
if args.json:
|
|
print(_json.dumps(self._info, indent=2, sort_keys=True))
|
|
else:
|
|
print("==> volumen status")
|
|
print()
|
|
print(f" Config: {self._info['config_path']}")
|
|
print(f" Data dir: {self._info['data_dir']}")
|
|
print(f" Users file: {self._info['users_file']}")
|
|
unit = "installed" if self._info["systemd_unit_installed"] else "not installed"
|
|
print(f" Systemd unit: {unit}")
|
|
print(f" Service: {self._info['service_active']}")
|
|
presence = "present" if self._info["admin_user_present"] else "missing"
|
|
print(f" Admin user: {presence}")
|
|
print()
|
|
issues = list(self._info["issues"])
|
|
if not issues:
|
|
print("All checks passed.")
|
|
else:
|
|
print("Issues:")
|
|
for issue in issues:
|
|
print(f" - {issue}")
|
|
return 0 if not self._info["issues"] else 1
|
|
|
|
|
|
class TestInitCommand:
|
|
def test_init_rejects_systemd_under_local(
|
|
self, monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str]
|
|
) -> None:
|
|
# Force ``use_user=True`` then ``--systemd`` → exit 2 with explanation.
|
|
monkeypatch.setattr(cli_module, "_is_root", lambda: False)
|
|
with (
|
|
mock.patch.object(
|
|
sys,
|
|
"argv",
|
|
["volumen", "init", "--local", "--systemd"],
|
|
),
|
|
pytest.raises(SystemExit) as exc,
|
|
):
|
|
main()
|
|
assert exc.value.code == 2
|
|
captured = capsys.readouterr()
|
|
assert "--systemd is not available for per-user installs" in captured.err
|
|
|
|
def test_init_propagates_installer_error(
|
|
self, monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str]
|
|
) -> None:
|
|
# ``system_install`` is mocked to raise InstallerError → exit 1 + stderr msg.
|
|
monkeypatch.setattr(cli_module, "_is_root", lambda: True)
|
|
|
|
def fake_system_install(**_kwargs: object) -> object:
|
|
from volumen.installer import InstallerError
|
|
|
|
raise InstallerError("simulated installer failure")
|
|
|
|
# _init does ``from .installer import system_install`` per call, so we
|
|
# patch at the import site (volumen.installer.system_install) rather
|
|
# than the local reference inside cli_module.
|
|
monkeypatch.setattr("volumen.installer.system_install", fake_system_install)
|
|
|
|
args = argparse.Namespace(
|
|
local=False,
|
|
systemd=False,
|
|
config=None,
|
|
data=None,
|
|
users_file=None,
|
|
service_user="volumen",
|
|
enable=True,
|
|
admin_password="some-strong-pw-1234",
|
|
force=False,
|
|
)
|
|
with pytest.raises(SystemExit) as exc:
|
|
cli_module._init(args)
|
|
captured = capsys.readouterr()
|
|
assert exc.value.code == 1
|
|
assert "simulated installer failure" in captured.err
|
|
|
|
def test_init_user_install_branch(
|
|
self, monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str]
|
|
) -> None:
|
|
# ``--local`` → user_install() path. Mock user_install to skip side effects.
|
|
monkeypatch.setattr(cli_module, "_is_root", lambda: False)
|
|
|
|
from volumen.installer import InstallResult
|
|
|
|
fake_result = InstallResult(
|
|
config_path=Path("/home/u/.config/volumen/config.toml"),
|
|
content_dir=Path("/home/u/.local/share/volumen/posts"),
|
|
users_file=Path("/home/u/.local/share/volumen/users.toml"),
|
|
service_user=None,
|
|
password_hash="scrypt$...",
|
|
session_key="x" * 128,
|
|
systemd_unit_path=None,
|
|
messages=[],
|
|
)
|
|
monkeypatch.setattr(
|
|
"volumen.installer.user_install",
|
|
lambda **_kw: fake_result,
|
|
)
|
|
|
|
args = argparse.Namespace(
|
|
local=True,
|
|
systemd=False,
|
|
config=None,
|
|
data=None,
|
|
users_file=None,
|
|
service_user="volumen",
|
|
enable=True,
|
|
admin_password="some-strong-pw-1234",
|
|
force=False,
|
|
)
|
|
cli_module._init(args) # Should not raise.
|
|
captured = capsys.readouterr()
|
|
assert "==> volumen installed." in captured.out
|
|
assert "Service: not installed" in captured.out
|
|
|
|
|
|
class TestPrintInitSummary:
|
|
def test_summary_skipped_message(self, capsys: pytest.CaptureFixture[str]) -> None:
|
|
from volumen.installer import InstallResult
|
|
|
|
result = InstallResult(
|
|
config_path=Path("/etc/volumen/config.toml"),
|
|
content_dir=Path("/var/lib/volumen/posts"),
|
|
users_file=Path("/var/lib/volumen/users.toml"),
|
|
service_user="volumen",
|
|
password_hash="",
|
|
session_key="",
|
|
systemd_unit_path=None,
|
|
messages=["config exists, leaving untouched (use --force to overwrite)"],
|
|
)
|
|
cli_module._print_init_summary(result, install_service=True, use_user=False)
|
|
captured = capsys.readouterr()
|
|
assert "left untouched" in captured.out
|
|
assert "Nothing to do" in captured.out
|
|
|
|
def test_summary_user_install(self, capsys: pytest.CaptureFixture[str]) -> None:
|
|
from volumen.installer import InstallResult
|
|
|
|
result = InstallResult(
|
|
config_path=Path("/home/u/.config/volumen/config.toml"),
|
|
content_dir=Path("/home/u/.local/share/volumen/posts"),
|
|
users_file=Path("/home/u/.local/share/volumen/users.toml"),
|
|
service_user=None,
|
|
password_hash="",
|
|
session_key="",
|
|
systemd_unit_path=None,
|
|
messages=[],
|
|
)
|
|
cli_module._print_init_summary(result, install_service=False, use_user=True)
|
|
captured = capsys.readouterr()
|
|
assert "Service: not installed" in captured.out
|