chore: prepare release v0.4.0
Test / test (push) Successful in 1m7s
Release / release (push) Successful in 1m11s

This commit is contained in:
2026-07-26 18:15:29 +02:00
parent 19c6161ae0
commit 3b8ed31f67
61 changed files with 6614 additions and 2704 deletions
+9 -8
View File
@@ -17,14 +17,14 @@ from volumen.users import Users
@pytest.fixture(autouse=True)
def _reset_login_attempts() -> Generator[None, None, None]:
def _reset_login_attempts() -> Generator[None]:
"""Reset the global login-attempts tracker before every test."""
reset_login_attempts()
yield
@pytest.fixture
def tmp_dir() -> Generator[Path, None, None]:
def tmp_dir() -> Generator[Path]:
"""Create a temporary directory that is cleaned up after the test."""
with tempfile.TemporaryDirectory() as d:
yield Path(d)
@@ -43,11 +43,12 @@ def tmp_content_dir(tmp_path: Path) -> Path:
@pytest.fixture
def config(tmp_content_dir: Path) -> Config:
def config(tmp_content_dir: Path, tmp_path: Path) -> Config:
"""Test Config instance pointing at a temp content directory."""
users_file = str(tmp_path / "users.toml")
return Config.load(
path="/nonexistent/volumen.toml",
overrides={"content": str(tmp_content_dir)},
overrides={"content": str(tmp_content_dir), "users_file": users_file},
)
@@ -66,7 +67,7 @@ def users_path(tmp_path: Path) -> str:
@pytest.fixture
def bootstrap_hash() -> str:
"""A pre-hashed password for bootstrap tests."""
return hash_pw("secret")
return hash_pw("volumen-test-password")
@pytest.fixture
@@ -78,7 +79,7 @@ def users(users_path: str, bootstrap_hash: str) -> Users:
@pytest.fixture
def client(config: Config, store: Store) -> TestClient:
"""FastAPI TestClient for public API tests."""
from volumen.server import create_app
from volumen.app import create_app
app = create_app(config, store)
return TestClient(app)
@@ -91,7 +92,7 @@ def admin_config_dir(tmp_path: Path) -> Path:
posts_dir.mkdir()
users_path = tmp_path / "users.toml"
config_path = tmp_path / "config.toml"
pw_hash = hash_pw("secret")
pw_hash = hash_pw("volumen-test-password")
config_path.write_text(
f'content_dir = "{posts_dir}"\n'
f'users_file = "{users_path}"\n\n'
@@ -104,8 +105,8 @@ def admin_config_dir(tmp_path: Path) -> Path:
@pytest.fixture
def admin_client(admin_config_dir: Path) -> TestClient:
"""FastAPI TestClient for admin tests with bootstrap admin."""
from volumen.app import create_app
from volumen.config import Config
from volumen.server import create_app
from volumen.store import Store
config = Config.load(path=str(admin_config_dir / "config.toml"))
+24 -11
View File
@@ -32,7 +32,11 @@ class TestAdminLogin:
token = _csrf_from_body(resp.text)
resp = admin_client.post(
"/admin/login",
data={"username": "admin", "password": "secret", "_csrf": token},
data={
"username": "admin",
"password": "volumen-test-password",
"_csrf": token,
},
follow_redirects=False,
)
assert resp.status_code in (302, 307, 303)
@@ -62,7 +66,11 @@ def _login(client: TestClient) -> str:
token = _csrf_from_body(resp.text)
client.post(
"/admin/login",
data={"username": "admin", "password": "secret", "_csrf": token},
data={
"username": "admin",
"password": "volumen-test-password",
"_csrf": token,
},
follow_redirects=True,
)
return token
@@ -264,16 +272,16 @@ class TestSettings:
"/admin/settings/password",
data={
"_csrf": token,
"current_password": "secret",
"new_password": "brand-new",
"current_password": "volumen-test-password",
"new_password": "volumen-brand-new-pw",
},
)
assert "Password updated" in resp.text
users_path = str(admin_config_dir / "users.toml")
fresh = Users(path=users_path)
assert fresh.authenticate("admin", "secret") is None
assert fresh.authenticate("admin", "brand-new") is not None
assert fresh.authenticate("admin", "volumen-test-password") is None
assert fresh.authenticate("admin", "volumen-brand-new-pw") is not None
def test_change_password_wrong_current(self, admin_client: TestClient) -> None:
_login(admin_client)
@@ -339,7 +347,7 @@ class TestFediverseSettings:
"/admin/settings/fediverse",
data={"_csrf": token, "fediverse_creator": "@user@example.com"},
)
assert "updated" in resp.text.lower() or "updated" in resp.text.lower()
assert "updated" in resp.text.lower()
def test_change_fediverse_creator_invalid(self, admin_client: TestClient) -> None:
_login(admin_client)
@@ -359,9 +367,14 @@ class TestUserManagement:
token = _csrf_from_body(resp.text)
resp = admin_client.post(
"/admin/settings/users",
data={"_csrf": token, "username": "editor", "password": "pw12345", "role": "author"},
data={
"_csrf": token,
"username": "editor",
"password": "editor-test-pw",
"role": "author",
},
)
assert "User added" in resp.text or "added" in resp.text.lower()
assert "User added" in resp.text
users_path = str(admin_config_dir / "users.toml")
users = Users(path=users_path)
@@ -373,7 +386,7 @@ class TestUserManagement:
token = _csrf_from_body(resp.text)
admin_client.post(
"/admin/settings/users",
data={"_csrf": token, "username": "editor2", "password": "pw12345"},
data={"_csrf": token, "username": "editor2", "password": "editor-test-pw"},
)
# Logout
@@ -382,7 +395,7 @@ class TestUserManagement:
admin_client.post("/admin/logout", data={"_csrf": token})
# Login as new user
_login_as(admin_client, "editor2", "pw12345")
_login_as(admin_client, "editor2", "editor-test-pw")
resp = admin_client.get("/admin/")
assert resp.status_code == 200
+423
View File
@@ -2,14 +2,18 @@
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
@@ -87,3 +91,422 @@ class TestPrepareConfig:
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
+104
View File
@@ -0,0 +1,104 @@
"""Concurrency tests: store/users racy write scenarios.
These tests hammer the same files from many threads to confirm that
per-target locks, fsync, and unique tempfiles do not corrupt data.
"""
from __future__ import annotations
from concurrent.futures import ThreadPoolExecutor, as_completed
from pathlib import Path
from volumen.post import Post
from volumen.store import Store
from volumen.users import Users
class TestStoreRace:
def test_concurrent_saves_keep_all_data(self, tmp_content_dir: Path) -> None:
"""Many threads save distinct posts in parallel."""
store = Store(str(tmp_content_dir), default_lang="en")
def worker(idx: int) -> str:
slug = f"post-{idx}"
post = Post(metadata={"slug": slug, "title": f"Post {idx}"}, body=f"body {idx}")
store.save(post)
return slug
with ThreadPoolExecutor(max_workers=16) as pool:
futures = [pool.submit(worker, i) for i in range(50)]
for f in as_completed(futures):
assert f.result() is not None
posts = store.all()
slugs = {p.slug for p in posts}
for i in range(50):
assert f"post-{i}" in slugs
def test_concurrent_saves_to_same_target(self, tmp_content_dir: Path) -> None:
"""Many threads save to the same target path; no data lost."""
store = Store(str(tmp_content_dir), default_lang="en")
def worker(idx: int) -> None:
post = Post(
metadata={"slug": "shared", "title": f"v{idx}"},
body=f"iteration {idx}",
)
store.save(post)
with ThreadPoolExecutor(max_workers=8) as pool:
list(pool.map(worker, range(20)))
post = store.find("shared")
assert post is not None
# The file content must end with one of our iterations, no torn write.
assert post.body.startswith("iteration ")
class TestUsersRace:
def test_concurrent_adds_keep_all_users(self, users_path: str) -> None:
users = Users(path=users_path)
def worker(idx: int) -> str:
name = f"user-{idx}"
users.add(name, "alice-test-password")
return name
with ThreadPoolExecutor(max_workers=8) as pool:
futures = [pool.submit(worker, i) for i in range(30)]
for f in as_completed(futures):
assert f.result() is not None
assert len(users.all) == 30
class TestSnapshotCache:
def test_snapshot_changes_invalidate_cache(self, tmp_content_dir: Path) -> None:
store = Store(str(tmp_content_dir), default_lang="en")
store.all() # prime cache
(tmp_content_dir / "extra.md").write_text('+++\ntitle = "Extra"\n+++\n\nNew post.\n')
# Even without touching mtime directly, stat() picks up the new file.
posts = store.all()
slugs = {p.slug for p in posts}
assert "extra" in slugs
def test_size_only_change_invalidates(self, tmp_content_dir: Path) -> None:
"""The cache key includes size; in-place rewrites invalidate."""
import os
store = Store(str(tmp_content_dir), default_lang="en")
# Force initial cache.
store.all()
# Overwrite an existing file with the same mtime but different size.
path = tmp_content_dir / "hello.md"
st = os.stat(path)
os.utime(path, (st.st_atime, st.st_mtime))
path.write_text(
'+++\ntitle = "Hello (resized)"\n+++\n\nNew body that is much longer than before.\n' * 5
)
# Touch back to original mtime so only size differs.
os.utime(path, (st.st_atime, st.st_mtime))
post = store.find("hello")
assert post is not None
assert "resized" in (post.title or "")
+177
View File
@@ -0,0 +1,177 @@
"""Coverage tests for feed helpers — RSS, JSON Feed, sitemap."""
from __future__ import annotations
from datetime import date
from typing import Any
from volumen.feed_helpers import (
_iso_date,
_rfc822_date,
render_json_feed,
render_rss_feed,
render_sitemap,
)
from volumen.post import Post
SITE: dict[str, Any] = {
"title": "My Blog",
"description": "A blog.",
"language": "en",
}
def _post(
*,
title: str = "Hello",
slug: str = "hello",
post_date: str = "2026-01-15",
tags: list[str] | None = None,
excerpt: str = "",
fediverse_creator: str = "",
body: str = "Body.",
) -> Post:
"""Build a Post with optional fields, simulating parse() without frontmatter I/O."""
metadata: dict[str, Any] = {"title": title, "slug": slug}
if post_date:
metadata["date"] = post_date
if tags is not None:
metadata["tags"] = tags
if excerpt:
metadata["excerpt"] = excerpt
if fediverse_creator:
metadata["fediverse_creator"] = fediverse_creator
return Post(metadata=metadata, body=body)
class TestRenderRSSFeed:
def test_basic_two_posts(self) -> None:
posts = [
_post(slug="a", title="Alpha"),
_post(slug="b", title="Beta"),
]
xml = render_rss_feed(posts, SITE, "https://example.com")
assert '<?xml version="1.0"' in xml
assert "<rss version=" in xml
assert "<title>My Blog</title>" in xml
assert "<link>https://example.com</link>" in xml
assert "<description>A blog.</description>" in xml
assert "<language>en</language>" in xml
assert "<title>Alpha</title>" in xml
assert "<title>Beta</title>" in xml
assert "<link>https://example.com/a</link>" in xml
assert "<link>https://example.com/b</link>" in xml
assert "pubDate" in xml
def test_truncates_to_20_items(self) -> None:
posts = [_post(slug=f"p{i:02d}") for i in range(25)]
xml = render_rss_feed(posts, SITE, "https://x")
assert xml.count("<item>") == 20
def test_escapes_special_characters_in_title(self) -> None:
xml = render_rss_feed(
[_post(title="A & B <C>", slug="x")],
SITE,
"https://x",
)
assert "A &amp; B &lt;C&gt;" in xml
def test_no_pub_date_when_date_missing(self) -> None:
# No `date` in metadata → date_string is None → no <pubDate> in the item.
xml = render_rss_feed([_post(post_date="")], SITE, "https://x")
item_block = xml.split("<item>")[1].split("</item>")[0]
assert "<pubDate>" not in item_block
def test_dc_creator_present_when_fediverse_set(self) -> None:
xml = render_rss_feed(
[_post(fediverse_creator="@alice@example.com")],
SITE,
"https://x",
)
assert "<dc:creator>@alice@example.com</dc:creator>" in xml
def test_no_dc_creator_when_fediverse_empty(self) -> None:
xml = render_rss_feed([_post()], SITE, "https://x")
assert "<dc:creator>" not in xml
class TestRenderJSONFeed:
def test_basic_shape(self) -> None:
feed = render_json_feed([_post(slug="a", tags=["intro"])], SITE, "https://x")
assert feed["version"] == "https://jsonfeed.org/version/1.1"
assert feed["title"] == "My Blog"
assert feed["home_page_url"] == "https://x"
assert feed["feed_url"] == "https://x/api/volumen/feed.json"
assert feed["language"] == "en"
assert feed["items"][0]["id"] == "https://x/a"
assert feed["items"][0]["url"] == "https://x/a"
assert feed["items"][0]["title"] == "Hello"
assert feed["items"][0]["tags"] == ["intro"]
def test_filters_empty_authors_and_tags(self) -> None:
# Default _post has no fediverse handle, no tags. ``summary`` is also
# present (it falls back to the derived excerpt from ``body``).
feed = render_json_feed([_post()], SITE, "https://x")
item = feed["items"][0]
assert "authors" not in item
assert "tags" not in item
def test_author_from_site_wide_fediverse(self) -> None:
site_with_creator = {**SITE, "fediverse_creator": "@site@example.com"}
feed = render_json_feed([_post()], site_with_creator, "https://x")
assert feed["items"][0]["authors"] == [{"name": "@site@example.com"}]
def test_author_from_post_overrides_site(self) -> None:
site_with_creator = {**SITE, "fediverse_creator": "@site@example.com"}
feed = render_json_feed(
[_post(fediverse_creator="@post@example.com")],
site_with_creator,
"https://x",
)
assert feed["items"][0]["authors"] == [{"name": "@post@example.com"}]
class TestRenderSitemap:
def test_basic_urlset(self) -> None:
xml = render_sitemap([_post(slug="x"), _post(slug="y")], "https://x")
assert "<urlset" in xml
assert "<loc>https://x/x</loc>" in xml
assert "<loc>https://x/y</loc>" in xml
assert "</urlset>" in xml
def test_lastmod_when_date_present(self) -> None:
xml = render_sitemap([_post()], "https://x")
assert "<lastmod>2026-01-15</lastmod>" in xml
def test_no_lastmod_when_date_empty(self) -> None:
# When post_date="" is passed, _post() omits "date" → date_string is None
# → no <lastmod> rendered.
xml = render_sitemap([_post(post_date="")], "https://x")
url_block = xml.split("<url>")[1].split("</url>")[0]
assert "<lastmod>" not in url_block
class TestDateHelpers:
def test_rfc822_from_date_object(self) -> None:
out = _rfc822_date(date(2026, 1, 15))
assert "2026" in out and "GMT" in out
def test_rfc822_from_iso_string(self) -> None:
out = _rfc822_date("2026-01-15")
assert "2026" in out
assert "GMT" in out
def test_rfc822_invalid_string_returns_as_is(self) -> None:
assert _rfc822_date("not-a-date") == "not-a-date"
def test_iso_none(self) -> None:
assert _iso_date(None) is None
def test_iso_date_object(self) -> None:
assert _iso_date(date(2026, 1, 15)) == "2026-01-15"
def test_iso_valid_string_passes_through(self) -> None:
assert _iso_date("2026-01-15") == "2026-01-15"
def test_iso_invalid_string_returns_as_is(self) -> None:
assert _iso_date("not-a-date") == "not-a-date"
+461
View File
@@ -0,0 +1,461 @@
"""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
+23
View File
@@ -0,0 +1,23 @@
"""Import smoke tests for the split application modules."""
from __future__ import annotations
from volumen.admin_auth_routes import router as auth_router
from volumen.admin_media_routes import router as media_router
from volumen.admin_post_routes import router as post_router
from volumen.admin_settings_routes import router as settings_router
from volumen.app import create_app
from volumen.server import create_app as legacy_create_app
def test_split_modules_import_and_construct_app(config, store) -> None:
assert legacy_create_app is create_app
assert auth_router.routes
assert post_router.routes
assert settings_router.routes
assert media_router.routes
app = create_app(config, store)
assert app.state.config is config
assert app.state.store is store
+4 -2
View File
@@ -34,8 +34,10 @@ class TestVerify:
assert verify_password("nope", encoded) is False
def test_verify_empty_string(self) -> None:
encoded = hash_password("")
assert verify_password("", encoded) is True
# ``hash_password`` rejects empty input; verify_password with an
# empty candidate returns False for any well-formed hash.
encoded = hash_password("some-pw")
assert verify_password("", encoded) is False
def test_verify_malformed_input(self) -> None:
assert verify_password("x", "not-a-hash") is False
+150
View File
@@ -0,0 +1,150 @@
"""Coverage tests for ``Post`` properties — non-happy-path branches."""
from __future__ import annotations
from datetime import date, timedelta
from volumen.post import Post
def _post(metadata: dict[str, object] | None = None, body: str = "") -> Post:
return Post(metadata=metadata, body=body)
class TestTagsProperty:
def test_tags_default_is_empty(self) -> None:
assert _post().tags == []
def test_tags_list_of_strings(self) -> None:
assert _post({"tags": ["python", "blog"]}).tags == ["python", "blog"]
def test_tags_non_list_coerces_to_empty(self) -> None:
# When ``tags`` is set to something that isn't a list, the property
# returns an empty list rather than raising.
assert _post({"tags": "python"}).tags == []
assert _post({"tags": 42}).tags == []
assert _post({"tags": None}).tags == []
class TestTranslationsProperty:
def test_default_is_empty(self) -> None:
assert _post().translations == {}
def test_dict_coerces_values_to_strings(self) -> None:
out = _post({"translations": {"cs": "cs-slug", "de": "de-slug"}}).translations
assert out == {"cs": "cs-slug", "de": "de-slug"}
def test_non_dict_returns_empty(self) -> None:
assert _post({"translations": ["cs", "de"]}).translations == {}
assert _post({"translations": "cs"}).translations == {}
class TestFediverseCreator:
def test_none(self) -> None:
assert _post().fediverse_creator is None
def test_empty_string_is_none(self) -> None:
assert _post({"fediverse_creator": ""}).fediverse_creator is None
assert _post({"fediverse_creator": " "}).fediverse_creator is None
def test_valid_fediverse_creator_regex(self) -> None:
p = _post({"fediverse_creator": "@alice@example.com"})
assert p.fediverse_creator == "@alice@example.com"
assert p.valid_fediverse_creator is True
def test_invalid_fediverse_creator(self) -> None:
# No `@host` part → regex doesn't match.
p = _post({"fediverse_creator": "alice"})
assert p.fediverse_creator == "alice"
assert p.valid_fediverse_creator is False
class TestPublishAt:
def test_none(self) -> None:
assert _post().publish_at is None
def test_date_object(self) -> None:
assert _post({"publish_at": date(2026, 6, 1)}).publish_at == date(2026, 6, 1)
def test_iso_string(self) -> None:
assert _post({"publish_at": "2026-06-01"}).publish_at == date(2026, 6, 1)
def test_invalid_string_returns_none(self) -> None:
assert _post({"publish_at": "not-a-date"}).publish_at is None
assert _post({"publish_at": 12345}).publish_at is None
class TestScheduled:
def test_past_date_is_not_scheduled(self) -> None:
past = (date.today() - timedelta(days=1)).isoformat()
assert _post({"publish_at": past}).scheduled is False
def test_future_date_is_scheduled(self) -> None:
future = (date.today() + timedelta(days=1)).isoformat()
assert _post({"publish_at": future}).scheduled is True
def test_no_publish_at(self) -> None:
assert _post().scheduled is False
class TestReadingTime:
def test_empty_body_minimum_one_minute(self) -> None:
assert _post(body="").reading_time == 1
def test_short_body(self) -> None:
# 5 words → ceil(5/200) = 1.
assert _post(body="one two three four five").reading_time == 1
def test_long_body(self) -> None:
body = " ".join(f"word{i}" for i in range(500))
# 500 words → ceil(500/200) = 3.
assert _post(body=body).reading_time == 3
class TestExcerpt:
def test_explicit_excerpt(self) -> None:
assert _post({"excerpt": "Short summary"}).excerpt == "Short summary"
def test_derived_from_body(self) -> None:
# No explicit excerpt → derived from first non-heading paragraph.
p = _post(body="Hello **world**.")
assert "Hello" in p.excerpt
assert "<" not in p.excerpt # HTML stripped
assert "*" not in p.excerpt # Markdown stripped
def test_derived_empty_when_only_headings(self) -> None:
assert _post(body="# Heading\n\n## Subheading").excerpt == ""
class TestSummaryAndDetail:
def test_summary_includes_all_known_fields(self) -> None:
p = _post(
{
"title": "T",
"slug": "s",
"date": "2026-01-15",
"lang": "en",
"tags": ["x"],
"author": "A",
"fediverse_creator": "@a@example.com",
"cover": "media/c.webp",
},
body="Body.",
)
s = p.summary()
assert s["slug"] == "s"
assert s["title"] == "T"
assert s["date"] == "2026-01-15"
assert s["lang"] == "en"
assert s["tags"] == ["x"]
assert s["author"] == "A"
assert s["fediverse_creator"] == "@a@example.com"
assert s["cover"] == "media/c.webp"
assert s["url"] == "/api/volumen/posts/s"
def test_detail_includes_body_and_html(self) -> None:
p = _post({"title": "T", "slug": "s"}, body="Hello.")
d = p.detail()
assert d["body"] == "Hello."
assert "html" in d
assert "<p>" in d["html"]
+422
View File
@@ -0,0 +1,422 @@
"""Security-focused tests: signature validation, sanitisation, cookies, headers."""
from __future__ import annotations
import io
from fastapi.testclient import TestClient
from volumen.markdown import render
from volumen.password import (
BLOCK_R,
COST_N,
KEY_LENGTH,
PARALLEL_P,
hash_password,
needs_rehash,
verify_password,
)
from volumen.store import (
ALLOWED_IMAGE_EXTENSIONS,
detect_image_extension,
validate_image_signature,
)
# ---------------------------------------------------------------------------
# Image signature validation
# ---------------------------------------------------------------------------
class TestImageSignature:
def test_webp_magic_accepted(self) -> None:
# Minimal RIFF/WEBP header (12 bytes).
good = b"RIFF" + b"\x10\x00\x00\x00" + b"WEBP" + b"VP8 "
assert validate_image_signature(good, ".webp") is True
def test_webp_wrong_magic_rejected(self) -> None:
# RIFF but not WEBP at offset 8.
bad = b"RIFF" + b"\x10\x00\x00\x00" + b"WAVE" + b"VP8 "
assert validate_image_signature(bad, ".webp") is False
def test_webp_too_short_rejected(self) -> None:
assert validate_image_signature(b"RIFF", ".webp") is False
def test_avif_magic_accepted(self) -> None:
# ftyp box with size=0x18, brand=avif.
avif = b"\x00\x00\x00\x18ftyp" + b"avif" + b"\x00\x00\x00\x00" + b"avif" + b"mif1"
assert validate_image_signature(avif, ".avif") is True
def test_avif_alt_brand_accepted(self) -> None:
avif = b"\x00\x00\x00\x18ftyp" + b"avis" + b"\x00\x00\x00\x00" + b"avis" + b"mif1"
assert validate_image_signature(avif, ".avif") is True
def test_avif_wrong_brand_rejected(self) -> None:
avif = b"\x00\x00\x00\x18ftyp" + b"mp42" + b"\x00\x00\x00\x00" + b"mp42" + b"isom"
assert validate_image_signature(avif, ".avif") is False
def test_avif_too_short_rejected(self) -> None:
assert validate_image_signature(b"\x00\x00\x00\x18ftyp", ".avif") is False
def test_unknown_extension_rejected(self) -> None:
assert validate_image_signature(b"anything", ".png") is False
def test_detect_extension_webp(self) -> None:
data = b"RIFF" + b"\x00\x00\x00\x00" + b"WEBP"
assert detect_image_extension(data) == ".webp"
def test_detect_extension_avif(self) -> None:
data = b"\x00\x00\x00\x18ftypavif" + b"\x00" * 8
assert detect_image_extension(data) == ".avif"
def test_detect_extension_unknown(self) -> None:
assert detect_image_extension(b"random") is None
def test_allowed_extensions_match_policy(self) -> None:
assert ALLOWED_IMAGE_EXTENSIONS == (".webp", ".avif")
# ---------------------------------------------------------------------------
# Markdown XSS sanitisation
# ---------------------------------------------------------------------------
class TestMarkdownSanitisation:
def test_script_tag_stripped(self) -> None:
rendered = render("Hello <script>alert(1)</script>world")
assert "<script" not in rendered.lower()
assert "alert(1)" not in rendered
def test_inline_event_handler_stripped(self) -> None:
rendered = render('<img src="x" onerror="alert(1)">')
assert "onerror" not in rendered.lower()
assert "alert" not in rendered.lower()
def test_javascript_url_sanitised(self) -> None:
rendered = render("[click me](javascript:alert(1))")
assert "javascript:" not in rendered.lower()
def test_data_image_url_kept(self) -> None:
rendered = render("![alt](data:image/webp;base64,UklGRg==)")
assert "<img" in rendered
def test_figure_wrapping_still_works(self) -> None:
rendered = render('![alt](url "caption")')
assert "<figure>" in rendered
assert "caption" in rendered
# ---------------------------------------------------------------------------
# Password hashing policy
# ---------------------------------------------------------------------------
class TestPasswordPolicy:
def test_strong_hash_passes(self) -> None:
encoded = hash_password("a-strong-password")
assert verify_password("a-strong-password", encoded) is True
def test_weak_params_rejected(self) -> None:
encoded = f"scrypt$8$1$1${'A' * 24}${'B' * 32}"
assert verify_password("anything", encoded) is False
def test_needs_rehash_for_strong_hash(self) -> None:
encoded = hash_password("a-strong-password")
assert needs_rehash(encoded) is False
def test_needs_rehash_for_weak_hash(self) -> None:
encoded = f"scrypt$8$1$1${'A' * 24}${'B' * 32}"
assert needs_rehash(encoded) is True
def test_needs_rehash_for_malformed(self) -> None:
assert needs_rehash("not-a-hash") is True
def test_module_constants_match_policy(self) -> None:
# Policy floor equals current implementation.
assert COST_N >= 16384
assert BLOCK_R >= 8
assert PARALLEL_P >= 1
assert KEY_LENGTH >= 32
# ---------------------------------------------------------------------------
# HTTP security headers and CSP
# ---------------------------------------------------------------------------
class TestSecurityHeaders:
def test_global_security_headers_present(self, client: TestClient) -> None:
response = client.get("/api/volumen/site")
assert response.headers.get("x-content-type-options") == "nosniff"
assert response.headers.get("x-frame-options") == "DENY"
assert response.headers.get("referrer-policy") == "strict-origin-when-cross-origin"
assert "permissions-policy" in response.headers
def test_csp_on_public_routes(self, client: TestClient) -> None:
response = client.get("/api/volumen/site")
csp = response.headers.get("content-security-policy", "")
assert "default-src 'self'" in csp
# Public routes have script-src but without nonce (nonce is admin-only)
assert "script-src 'self'" in csp
assert "nonce-" not in csp
def test_admin_route_has_csp_with_nonce(self, admin_client: TestClient) -> None:
response = admin_client.get("/admin/login")
csp = response.headers.get("content-security-policy", "")
assert "default-src 'self'" in csp
assert "'unsafe-inline'" not in csp
assert "nonce-" in csp
def test_https_only_cookie_unset_when_untrusted(self, client: TestClient) -> None:
# Default trust_proxy=false — X-Forwarded-Proto is ignored.
response = client.get("/admin/login")
# No cookie issued, but the middleware must mark session_secure=False.
assert response.status_code == 200
def test_https_only_cookie_with_trusted_proxy(self, admin_config_dir) -> None:
# Build a fresh app with trust_proxy=True and confirm it boots.
from volumen.app import create_app
from volumen.config import Config
from volumen.store import Store
config = Config.load(path=str(admin_config_dir / "config.toml"))
config._data["server"]["trust_proxy"] = True # type: ignore[index]
store = Store(config.content_dir)
app = create_app(config, store)
client = TestClient(app)
resp = client.get("/admin/login")
assert resp.status_code == 200
def test_cookie_secure_false_without_proxy_or_flag(self, admin_client: TestClient) -> None:
# When trust_proxy=false and cookie_secure=false, the SessionMiddleware
# is configured with https_only=False.
resp = admin_client.get("/admin/login")
assert resp.status_code == 200
# ---------------------------------------------------------------------------
# Config validation
# ---------------------------------------------------------------------------
class TestConfigValidation:
def test_default_config_is_valid(self, config) -> None:
# Should not raise.
config.validate()
def test_invalid_port_rejected(self, tmp_path) -> None:
from volumen.config import Config
(tmp_path / "posts").mkdir()
cfg = Config.load(
path="/nonexistent.toml",
overrides={
"users_file": str(tmp_path / "users.toml"),
"content": str(tmp_path / "posts"),
},
)
cfg._data["server"]["port"] = 999999 # type: ignore[index]
try:
cfg.validate()
except Exception as exc:
assert "port" in str(exc).lower()
else:
raise AssertionError("expected ConfigError")
def test_invalid_env_rejected(self, tmp_path) -> None:
from volumen.config import Config, ConfigError
cfg = Config.load(
path="/nonexistent.toml",
overrides={"users_file": str(tmp_path / "users.toml")},
)
cfg._data["server"]["env"] = "staging" # type: ignore[index]
try:
cfg.validate()
except ConfigError as exc:
assert "env" in str(exc).lower()
else:
raise AssertionError("expected ConfigError")
def test_invalid_base_url_rejected(self, tmp_path) -> None:
from volumen.config import Config, ConfigError
cfg = Config.load(
path="/nonexistent.toml",
overrides={"users_file": str(tmp_path / "users.toml")},
)
cfg._data["site"]["base_url"] = "not-a-url" # type: ignore[index]
try:
cfg.validate()
except ConfigError as exc:
assert "base_url" in str(exc).lower()
else:
raise AssertionError("expected ConfigError")
def test_production_requires_session_key(self, tmp_path) -> None:
from volumen.config import Config
(tmp_path / "posts").mkdir()
cfg = Config.load(
path="/nonexistent.toml",
overrides={
"users_file": str(tmp_path / "users.toml"),
"content": str(tmp_path / "posts"),
},
)
cfg._data["server"]["env"] = "production" # type: ignore[index]
cfg._data["admin"]["session_key"] = "" # type: ignore[index]
# Validation succeeds (file is allowed) but create_app will refuse.
cfg.validate()
# ---------------------------------------------------------------------------
# Session secret validation
# ---------------------------------------------------------------------------
class TestSessionSecret:
def test_production_without_session_key_fails(self, config) -> None:
from volumen.app import create_app
from volumen.store import Store
config._data["server"]["env"] = "production" # type: ignore[index]
config._data["admin"]["session_key"] = "" # type: ignore[index]
store = Store(config.content_dir)
try:
create_app(config, store)
except RuntimeError as exc:
assert "session_key" in str(exc)
else:
raise AssertionError("expected RuntimeError")
def test_production_with_short_key_fails(self, config) -> None:
from volumen.app import create_app
from volumen.store import Store
config._data["server"]["env"] = "production" # type: ignore[index]
config._data["admin"]["session_key"] = "tooshort" # type: ignore[index]
store = Store(config.content_dir)
try:
create_app(config, store)
except RuntimeError as exc:
assert "64 bytes" in str(exc)
else:
raise AssertionError("expected RuntimeError")
# ---------------------------------------------------------------------------
# Password policy helpers
# ---------------------------------------------------------------------------
class TestPasswordPolicyHelpers:
def test_password_error_min_length(self) -> None:
from volumen.auth import password_error
assert password_error("short", min_length=10) is not None
def test_password_error_max_length(self) -> None:
from volumen.auth import password_error
assert password_error("x" * 2000, min_length=10, max_length=1024) is not None
def test_password_error_empty(self) -> None:
from volumen.auth import password_error
assert password_error("", min_length=10) is not None
def test_password_error_common_rejected(self) -> None:
from volumen.auth import password_error
assert password_error("password", min_length=10) is not None
def test_password_error_accepts_strong(self) -> None:
from volumen.auth import password_error
assert password_error("volumen-strong-pw", min_length=10) is None
# ---------------------------------------------------------------------------
# Store hardening
# ---------------------------------------------------------------------------
class _Upload:
"""Fake UploadFile for testing validate_upload."""
content_type: str
filename: str
file: io.BytesIO
def __init__(
self, data: bytes, content_type: str = "image/webp", filename: str = "x.webp"
) -> None:
self._data = data
self.content_type = content_type
self.filename = filename
self.file = io.BytesIO(data)
self._pos = 0
async def read(self, size: int = -1) -> bytes:
if size == -1:
result = self._data[self._pos :]
self._pos = len(self._data)
return result
result = self._data[self._pos : self._pos + size]
self._pos += size
return result
class TestStoreHardening:
def test_validate_upload_accepts_webp(self, tmp_content_dir) -> None:
# Build a fake UploadFile-shaped object.
good_webp = b"RIFF" + b"\x10\x00\x00\x00" + b"WEBP" + b"\x00" * 100
import asyncio
from volumen.upload_validation import validate_upload
async def go() -> tuple[bytes, str]:
return await validate_upload(_Upload(good_webp), max_bytes=10_000)
data, ext = asyncio.run(go())
assert ext == ".webp"
assert data == good_webp
def test_validate_upload_rejects_mime_mismatch(self, tmp_content_dir) -> None:
import asyncio
from volumen.upload_validation import validate_upload
async def go() -> None:
try:
await validate_upload(
_Upload(b"nope", content_type="image/png", filename="x.png"),
max_bytes=10_000,
)
except Exception as exc:
# The detail is JSON-encoded text.
msg = str(exc)
assert "unsupported_format" in msg or "signature" in msg.lower()
else:
raise AssertionError("expected HTTPException")
asyncio.run(go())
def test_store_upload_uses_signature_extension(self, tmp_content_dir) -> None:
from volumen.store import Store
store = Store(str(tmp_content_dir))
avif_bytes = b"\x00\x00\x00\x18ftyp" + b"avif" + b"\x00" * 100
url = store.store_upload("evil.exe", None, bytes_data=avif_bytes)
assert url.endswith(".avif")
assert "evil" not in url
def test_store_upload_resolves_whitelisted_extension(self, tmp_content_dir) -> None:
from volumen.store import Store
store = Store(str(tmp_content_dir))
webp_bytes = b"RIFF" + b"\x10\x00\x00\x00" + b"WEBP" + b"\x00" * 100
# File name has an unrelated extension; signature wins.
url = store.store_upload("cover.png", None, bytes_data=webp_bytes)
assert url.endswith(".webp")
+27 -2
View File
@@ -190,9 +190,34 @@ class TestJSONFeed:
assert item["authors"] == [{"name": "@jf@example.io"}]
def test_json_feed_falls_back_to_site_default_fediverse_creator(
self, client: TestClient, tmp_content_dir: Path
self, tmp_path: Path, tmp_content_dir: Path
) -> None:
pass # Requires a custom config with site.fediverse_creator set
"""When a post has no fediverse_creator, fall back to site.fediverse_creator."""
from fastapi.testclient import TestClient
from volumen.app import create_app
from volumen.config import Config
from volumen.store import Store
config_path = tmp_path / "custom.toml"
users_path = tmp_path / "users.toml"
users_path.write_text("")
config_path.write_text(
f'content_dir = "{tmp_content_dir}"\n'
f'users_file = "{users_path}"\n'
"\n[site]\n"
'fediverse_creator = "@siteowner@example.social"\n'
)
custom_config = Config.load(path=str(config_path))
custom_store = Store(str(tmp_content_dir), default_lang="en")
custom_app = create_app(custom_config, custom_store)
custom_client = TestClient(custom_app)
response = custom_client.get("/api/volumen/feed.json")
assert response.status_code == 200
item = response.json()["items"][0]
assert "authors" in item
assert item["authors"] == [{"name": "@siteowner@example.social"}]
class TestSitemap:
+5 -5
View File
@@ -11,7 +11,7 @@ class TestBootstrap:
def test_bootstrap_user_from_hash(self, users_path: str, bootstrap_hash: str) -> None:
users = Users(path=users_path, bootstrap_hash=bootstrap_hash)
assert len(users.all) == 1
assert users.authenticate("admin", "secret") is not None
assert users.authenticate("admin", "volumen-test-password") is not None
def test_bootstrap_user_is_admin(self, users_path: str, bootstrap_hash: str) -> None:
users = Users(path=users_path, bootstrap_hash=bootstrap_hash)
@@ -34,13 +34,13 @@ class TestFind:
class TestAuthenticate:
def test_authenticate_correct_password(self, users: Users) -> None:
assert users.authenticate("admin", "secret") is not None
assert users.authenticate("admin", "volumen-test-password") is not None
def test_authenticate_wrong_password(self, users: Users) -> None:
assert users.authenticate("admin", "nope") is None
def test_authenticate_nonexistent_user(self, users: Users) -> None:
assert users.authenticate("nobody", "secret") is None
assert users.authenticate("nobody", "volumen-test-password") is None
class TestAdd:
@@ -209,9 +209,9 @@ class TestPersistence:
def test_password_survives_persistence(self, users_path: str) -> None:
users = Users(path=users_path)
users.add("alice", "secret123")
users.add("alice", "alice-test-password")
fresh = Users(path=users_path)
assert fresh.authenticate("alice", "secret123") is not None
assert fresh.authenticate("alice", "alice-test-password") is not None
assert fresh.authenticate("alice", "wrong") is None
def test_mtime_cache_invalidation(self, users_path: str) -> None: