Files
DECNET/tests/cli/test_init.py
anti 39a298f685 feat(init): persist DECNET-service api-user/api-group to decnet.ini
DEBT-035 step 1. The composer needs to know which uid/gid to inject
into each compose fragment's `user:` directive at deploy time. Today
the resolved `--user` / `--group` values reach systemd unit
rendering (init.py:349–354) but are not persisted anywhere the
composer can read them.

Persist as **names** (not numeric ids) under `[decnet] api-user` /
`api-group` in the rendered decnet.ini placeholder. Resolution to
uid/gid happens at deploy time on whichever host runs the deploy,
via `pwd.getpwnam(...)` / `grp.getgrnam(...)` — so the same user
name can have different uids on master vs agents (heterogeneous
/etc/passwd) without breaking artifact ownership. The existing
config_ini auto-translates kebab→DECNET_API_USER / DECNET_API_GROUP
at load time; no domain-map changes needed.

Two new tests: one asserting the rendered ini carries the
`api-user` / `api-group` keys for the values passed to `--user` /
`--group`; one round-tripping through `load_ini_config` to confirm
the env vars land in `os.environ` for the composer to pick up.
2026-05-02 19:33:53 -04:00

546 lines
20 KiB
Python

"""Orchestration tests for ``decnet init``.
The command is a thin orchestrator over privileged system calls. We
exercise every branch by monkeypatching subprocess + identity lookups
and using the hidden ``--prefix`` option to redirect filesystem writes
into a pytest ``tmp_path``.
"""
from __future__ import annotations
import os
from pathlib import Path
from typing import Any, List
import pytest
from typer.testing import CliRunner
from decnet.cli import app
from decnet.cli import init as _init
runner = CliRunner()
@pytest.fixture
def subprocess_calls(monkeypatch: Any) -> List[List[str]]:
calls: List[List[str]] = []
def _fake_run(argv: List[str], *a: Any, **kw: Any) -> Any:
calls.append(list(argv))
class _Ok:
returncode = 0
return _Ok()
monkeypatch.setattr(_init.subprocess, "run", _fake_run)
return calls
@pytest.fixture
def no_missing_tools(monkeypatch: Any) -> None:
monkeypatch.setattr(_init.shutil, "which", lambda _: "/usr/bin/fake")
@pytest.fixture
def present_user_and_group(monkeypatch: Any) -> None:
class _Stub:
pw_uid = 1000
gr_gid = 1000
monkeypatch.setattr(_init.pwd, "getpwnam", lambda _: _Stub())
monkeypatch.setattr(_init.grp, "getgrnam", lambda _: _Stub())
@pytest.fixture
def missing_user_and_group(monkeypatch: Any) -> None:
def _raise(_: str) -> None:
raise KeyError
monkeypatch.setattr(_init.pwd, "getpwnam", _raise)
monkeypatch.setattr(_init.grp, "getgrnam", _raise)
def _seed_deploy(monkeypatch: Any, tmp_path: Path) -> Path:
"""Point `_deploy_root()` at a faked deploy tree under tmp_path.
Services are Jinja2 templates keyed on ``{{ install_dir }}`` —
matching production layout since the refactor that made install
path configurable.
"""
deploy = tmp_path / "deploy"
(deploy / "polkit").mkdir(parents=True)
(deploy / "tmpfiles.d").mkdir()
(deploy / "decnet-bus.service.j2").write_text(
"[Service]\nExecStart={{ install_dir }}/venv/bin/decnet bus\n"
)
(deploy / "decnet-api.service.j2").write_text(
"[Service]\nWorkingDirectory={{ install_dir }}\n"
"ExecStart={{ install_dir }}/venv/bin/decnet api\n"
)
(deploy / "decnet.target").write_text("# target\n")
(deploy / "polkit" / "50-decnet-workers.rules.j2").write_text(
'// rule for {{ group }}\n'
)
(deploy / "tmpfiles.d" / "decnet.conf").write_text("d /run/decnet\n")
(deploy / "logrotate.d").mkdir()
(deploy / "logrotate.d" / "decnet").write_text(
"/var/log/decnet/*.log {\n daily\n rotate 7\n copytruncate\n}\n"
)
monkeypatch.setattr(_init, "_deploy_root", lambda: deploy)
return deploy
def test_non_root_exits_one(monkeypatch: Any) -> None:
monkeypatch.setattr(_init.os, "geteuid", lambda: 1000)
result = runner.invoke(app, ["init"])
assert result.exit_code == 1
assert "must run as root" in result.output
def test_dry_run_issues_no_subprocess_calls(
monkeypatch: Any, tmp_path: Path, subprocess_calls: List[List[str]],
no_missing_tools: None, missing_user_and_group: None,
) -> None:
_seed_deploy(monkeypatch, tmp_path)
result = runner.invoke(
app,
["init", "--dry-run", "--prefix", str(tmp_path / "root")],
)
assert result.exit_code == 0, result.output
assert subprocess_calls == [], (
f"dry-run must not exec anything, got {subprocess_calls}"
)
assert "would run:" in result.output
# No real files created either.
assert not (tmp_path / "root" / "etc/systemd/system").exists()
def test_missing_user_and_group_triggers_useradd_groupadd(
monkeypatch: Any, tmp_path: Path, subprocess_calls: List[List[str]],
no_missing_tools: None, missing_user_and_group: None,
) -> None:
_seed_deploy(monkeypatch, tmp_path)
result = runner.invoke(
app,
["init", "--no-start", "--prefix", str(tmp_path / "root")],
)
assert result.exit_code == 0, result.output
groupadds = [c for c in subprocess_calls if c[:1] == ["groupadd"]]
useradds = [c for c in subprocess_calls if c[:1] == ["useradd"]]
assert groupadds == [["groupadd", "--system", "decnet"]]
assert useradds and useradds[0][:6] == [
"useradd", "--system", "--gid", "decnet", "--home-dir", "/opt/decnet",
]
assert useradds[0][-1] == "decnet"
def test_present_user_and_group_skipped(
monkeypatch: Any, tmp_path: Path, subprocess_calls: List[List[str]],
no_missing_tools: None, present_user_and_group: None,
) -> None:
_seed_deploy(monkeypatch, tmp_path)
result = runner.invoke(
app,
["init", "--no-start", "--prefix", str(tmp_path / "root")],
)
assert result.exit_code == 0, result.output
assert all(c[0] not in ("groupadd", "useradd") for c in subprocess_calls)
assert "[SKIP]" in result.output
def test_unit_files_are_installed_then_idempotent(
monkeypatch: Any, tmp_path: Path, subprocess_calls: List[List[str]],
no_missing_tools: None, present_user_and_group: None,
) -> None:
_seed_deploy(monkeypatch, tmp_path)
prefix = tmp_path / "root"
# First run: installs.
r1 = runner.invoke(
app, ["init", "--no-start", "--prefix", str(prefix)],
)
assert r1.exit_code == 0, r1.output
target = prefix / "etc/systemd/system" / "decnet.target"
assert target.is_file()
assert target.read_text() == "# target\n"
# Second run: every copy should SKIP.
subprocess_calls.clear()
r2 = runner.invoke(
app, ["init", "--no-start", "--prefix", str(prefix)],
)
assert r2.exit_code == 0, r2.output
assert "unit files up to date" in r2.output
def test_init_installs_logrotate_config(
monkeypatch: Any, tmp_path: Path, subprocess_calls: List[List[str]],
no_missing_tools: None, present_user_and_group: None,
) -> None:
"""`decnet init` must drop /etc/logrotate.d/decnet so /var/log/decnet/
can't fill the disk on a noisy honeypot day."""
_seed_deploy(monkeypatch, tmp_path)
prefix = tmp_path / "root"
r = runner.invoke(app, ["init", "--no-start", "--prefix", str(prefix)])
assert r.exit_code == 0, r.output
cfg = prefix / "etc/logrotate.d/decnet"
assert cfg.is_file(), "logrotate config should be installed"
body = cfg.read_text()
assert "copytruncate" in body, (
"must use copytruncate; ingester holds the file open and won't "
"reopen on a rename rotation"
)
def test_init_writes_decnet_ini_not_config_ini(
monkeypatch: Any, tmp_path: Path, subprocess_calls: List[List[str]],
no_missing_tools: None, missing_user_and_group: None,
) -> None:
"""Placeholder target is /etc/decnet/decnet.ini (new name) — matches
what decnet.config_ini.load_ini_config() actually reads. Guards
against regressing to the old `config.ini` name."""
_seed_deploy(monkeypatch, tmp_path)
prefix = tmp_path / "root"
r = runner.invoke(app, ["init", "--no-start", "--prefix", str(prefix)])
assert r.exit_code == 0, r.output
ini = prefix / "etc/decnet/decnet.ini"
legacy = prefix / "etc/decnet/config.ini"
assert ini.is_file(), "decnet.ini should be written"
assert not legacy.exists(), "legacy config.ini must not be written"
body = ini.read_text()
# Admin-facing sections are documented as commented examples so
# the placeholder teaches the file shape.
for header in ("[decnet]", "[api]", "[web]", "[database]",
"[bus]", "[swarm]", "[logging]", "[ingester]",
"[tracing]", "[agent]"):
assert header in body, f"placeholder missing {header} example"
def test_init_persists_api_user_group_to_decnet_ini(
monkeypatch: Any, tmp_path: Path, subprocess_calls: List[List[str]],
no_missing_tools: None, missing_user_and_group: None,
) -> None:
"""DEBT-035: `decnet init --user X --group Y` must persist the
DECNET-service user/group **as names** to decnet.ini under
`[decnet] api-user` / `api-group`. Resolution to numeric uid/gid
happens at deploy time on whichever host runs the deploy — names
survive across master/agent uid namespaces while raw numbers
would mismatch."""
_seed_deploy(monkeypatch, tmp_path)
prefix = tmp_path / "root"
r = runner.invoke(app, [
"init", "--no-start", "--prefix", str(prefix),
"--user", "decoyman", "--group", "decoygrp",
])
assert r.exit_code == 0, r.output
body = (prefix / "etc/decnet/decnet.ini").read_text()
assert "api-user = decoyman" in body
assert "api-group = decoygrp" in body
def test_init_decnet_ini_loads_via_config_ini(
monkeypatch: Any, tmp_path: Path, subprocess_calls: List[List[str]],
no_missing_tools: None, missing_user_and_group: None,
) -> None:
"""Defence-in-depth: round-trip the rendered ini through
`decnet.config_ini.load_ini_config` and confirm that
DECNET_API_USER / DECNET_API_GROUP appear in os.environ. The
composer reads those env vars at deploy time."""
from decnet.config_ini import load_ini_config
_seed_deploy(monkeypatch, tmp_path)
prefix = tmp_path / "root"
r = runner.invoke(app, [
"init", "--no-start", "--prefix", str(prefix),
"--user", "deckyowner", "--group", "deckygroup",
])
assert r.exit_code == 0, r.output
ini = prefix / "etc/decnet/decnet.ini"
monkeypatch.delenv("DECNET_API_USER", raising=False)
monkeypatch.delenv("DECNET_API_GROUP", raising=False)
load_ini_config(ini)
assert os.environ.get("DECNET_API_USER") == "deckyowner"
assert os.environ.get("DECNET_API_GROUP") == "deckygroup"
def test_install_dir_renders_into_service_units(
monkeypatch: Any, tmp_path: Path, subprocess_calls: List[List[str]],
no_missing_tools: None, missing_user_and_group: None,
) -> None:
"""`--install-dir /srv/decnet` must land in the rendered service
files. Regression guard for the Jinja2 templating refactor."""
_seed_deploy(monkeypatch, tmp_path)
prefix = tmp_path / "root"
r = runner.invoke(
app,
[
"init", "--no-start",
"--prefix", str(prefix),
"--install-dir", "/srv/decnet",
],
)
assert r.exit_code == 0, r.output
api_unit = prefix / "etc/systemd/system" / "decnet-api.service"
bus_unit = prefix / "etc/systemd/system" / "decnet-bus.service"
assert api_unit.is_file()
api_text = api_unit.read_text()
assert "/srv/decnet" in api_text
assert "/opt/decnet" not in api_text
assert "{{" not in api_text, "unrendered Jinja tag leaked through"
assert "/srv/decnet" in bus_unit.read_text()
# useradd --home-dir must match the install_dir override too.
useradds = [c for c in subprocess_calls if c and c[0] == "useradd"]
assert useradds, "expected useradd call"
assert "/srv/decnet" in useradds[0]
assert "/opt/decnet" not in useradds[0]
# And /srv/decnet on disk should be the dir we created.
assert (prefix / "srv/decnet").is_dir()
assert not (prefix / "opt/decnet").exists()
def test_install_dir_defaults_to_opt(
monkeypatch: Any, tmp_path: Path, subprocess_calls: List[List[str]],
no_missing_tools: None, present_user_and_group: None,
) -> None:
"""Default --install-dir is /opt/decnet — existing installs remain
byte-identical with no explicit flag."""
_seed_deploy(monkeypatch, tmp_path)
prefix = tmp_path / "root"
r = runner.invoke(app, ["init", "--no-start", "--prefix", str(prefix)])
assert r.exit_code == 0, r.output
api_unit = prefix / "etc/systemd/system" / "decnet-api.service"
assert "/opt/decnet" in api_unit.read_text()
def test_install_dir_rejects_relative_path(
monkeypatch: Any, tmp_path: Path,
no_missing_tools: None, missing_user_and_group: None,
) -> None:
"""Relative install_dir would break every absolute path in a
rendered service. Reject at the CLI boundary with a clear message."""
_seed_deploy(monkeypatch, tmp_path)
r = runner.invoke(
app,
[
"init", "--no-start",
"--prefix", str(tmp_path / "root"),
"--install-dir", "relative/path",
],
)
assert r.exit_code == 1
assert "must be absolute" in r.output
def test_install_dir_custom_idempotent_second_run(
monkeypatch: Any, tmp_path: Path, subprocess_calls: List[List[str]],
no_missing_tools: None, present_user_and_group: None,
) -> None:
"""Rendering the same templates twice with the same context must
produce byte-identical output — second run SKIPs, no churn."""
_seed_deploy(monkeypatch, tmp_path)
prefix = tmp_path / "root"
runner.invoke(
app,
[
"init", "--no-start",
"--prefix", str(prefix),
"--install-dir", "/srv/decnet",
],
)
r2 = runner.invoke(
app,
[
"init", "--no-start",
"--prefix", str(prefix),
"--install-dir", "/srv/decnet",
],
)
assert r2.exit_code == 0, r2.output
assert "unit files up to date" in r2.output
def test_force_overwrites_existing_units(
monkeypatch: Any, tmp_path: Path, subprocess_calls: List[List[str]],
no_missing_tools: None, present_user_and_group: None,
) -> None:
deploy = _seed_deploy(monkeypatch, tmp_path)
prefix = tmp_path / "root"
runner.invoke(app, ["init", "--no-start", "--prefix", str(prefix)])
# Mutate the installed copy so SHA-256 matches source, but we ask
# for --force anyway: source wins.
target = prefix / "etc/systemd/system" / "decnet.target"
target.write_text("# tampered\n")
r = runner.invoke(
app,
["init", "--no-start", "--force", "--prefix", str(prefix)],
)
assert r.exit_code == 0, r.output
assert target.read_text() == (deploy / "decnet.target").read_text()
def test_no_start_suppresses_target_start(
monkeypatch: Any, tmp_path: Path, subprocess_calls: List[List[str]],
no_missing_tools: None, present_user_and_group: None,
) -> None:
_seed_deploy(monkeypatch, tmp_path)
runner.invoke(
app,
["init", "--no-start", "--prefix", str(tmp_path / "root")],
)
enables = [
c for c in subprocess_calls
if c[:2] == ["systemctl", "enable"]
]
assert enables == []
def test_default_invokes_target_start(
monkeypatch: Any, tmp_path: Path, subprocess_calls: List[List[str]],
no_missing_tools: None, present_user_and_group: None,
) -> None:
_seed_deploy(monkeypatch, tmp_path)
result = runner.invoke(
app, ["init", "--prefix", str(tmp_path / "root")],
)
assert result.exit_code == 0, result.output
assert ["systemctl", "enable", "--now", "decnet.target"] in subprocess_calls
assert ["systemctl", "daemon-reload"] in subprocess_calls
def _seed_installed_state(prefix: Path) -> None:
"""Create the files a prior `decnet init` would have installed."""
systemd = prefix / "etc/systemd/system"
systemd.mkdir(parents=True)
(systemd / "decnet-bus.service").write_text("# bus\n")
(systemd / "decnet-api.service").write_text("# api\n")
(systemd / "decnet.target").write_text("# target\n")
polkit = prefix / "etc/polkit-1/rules.d"
polkit.mkdir(parents=True)
(polkit / "50-decnet-workers.rules").write_text("// rule\n")
tmpfiles = prefix / "etc/tmpfiles.d"
tmpfiles.mkdir(parents=True)
(tmpfiles / "decnet.conf").write_text("d /run/decnet\n")
logrotate = prefix / "etc/logrotate.d"
logrotate.mkdir(parents=True)
(logrotate / "decnet").write_text("/var/log/decnet/*.log {}\n")
etc_decnet = prefix / "etc/decnet"
etc_decnet.mkdir(parents=True)
(etc_decnet / "decnet.ini").write_text("[decnet]\n")
# Also seed the legacy config.ini so we cover the legacy-cleanup path.
(etc_decnet / "config.ini").write_text("[decnet]\n")
(prefix / "opt/decnet").mkdir(parents=True)
(prefix / "run/decnet").mkdir(parents=True)
(prefix / "var/lib/decnet").mkdir(parents=True)
(prefix / "var/log/decnet").mkdir(parents=True)
(prefix / "var/log/decnet/events.jsonl").write_text("{}\n")
def test_deinit_removes_units_polkit_tmpfiles_and_preserves_data(
tmp_path: Path, subprocess_calls: List[List[str]],
no_missing_tools: None, present_user_and_group: None,
) -> None:
prefix = tmp_path / "root"
_seed_installed_state(prefix)
result = runner.invoke(
app, ["init", "--deinit", "--prefix", str(prefix)],
)
assert result.exit_code == 0, result.output
# Units + polkit + tmpfiles.d gone.
assert not (prefix / "etc/systemd/system/decnet-bus.service").exists()
assert not (prefix / "etc/systemd/system/decnet.target").exists()
assert not (prefix / "etc/polkit-1/rules.d/50-decnet-workers.rules").exists()
assert not (prefix / "etc/tmpfiles.d/decnet.conf").exists()
assert not (prefix / "etc/logrotate.d/decnet").exists()
assert not (prefix / "etc/decnet").exists()
assert not (prefix / "opt/decnet").exists()
# Data dirs preserved.
assert (prefix / "var/lib/decnet").exists()
assert (prefix / "var/log/decnet/events.jsonl").read_text() == "{}\n"
# systemctl disable + daemon-reload invoked.
assert ["systemctl", "disable", "--now", "decnet.target"] in subprocess_calls
assert ["systemctl", "daemon-reload"] in subprocess_calls
# User / group are PRESERVED without --purge — an operator who
# passed --user $USER during dev must not lose their login account.
assert ["userdel", "decnet"] not in subprocess_calls
assert ["groupdel", "decnet"] not in subprocess_calls
def test_deinit_purge_wipes_data_dirs(
tmp_path: Path, subprocess_calls: List[List[str]],
no_missing_tools: None, present_user_and_group: None,
) -> None:
prefix = tmp_path / "root"
_seed_installed_state(prefix)
result = runner.invoke(
app, ["init", "--deinit", "--purge", "--prefix", str(prefix)],
)
assert result.exit_code == 0, result.output
assert not (prefix / "var/lib/decnet").exists()
assert not (prefix / "var/log/decnet").exists()
# --purge also removes the service user/group.
assert ["userdel", "decnet"] in subprocess_calls
assert ["groupdel", "decnet"] in subprocess_calls
def test_deinit_is_idempotent_on_clean_host(
tmp_path: Path, subprocess_calls: List[List[str]],
no_missing_tools: None, missing_user_and_group: None,
) -> None:
prefix = tmp_path / "root"
# Nothing seeded — everything should SKIP.
result = runner.invoke(
app, ["init", "--deinit", "--prefix", str(prefix)],
)
assert result.exit_code == 0, result.output
assert result.output.count("[SKIP]") >= 5
# userdel / groupdel never invoked because user/group are absent.
assert ["userdel", "decnet"] not in subprocess_calls
assert ["groupdel", "decnet"] not in subprocess_calls
def test_deinit_dry_run_touches_nothing(
tmp_path: Path, subprocess_calls: List[List[str]],
no_missing_tools: None, present_user_and_group: None,
) -> None:
prefix = tmp_path / "root"
_seed_installed_state(prefix)
result = runner.invoke(
app,
["init", "--deinit", "--purge", "--dry-run", "--prefix", str(prefix)],
)
assert result.exit_code == 0, result.output
assert subprocess_calls == []
assert (prefix / "etc/systemd/system/decnet.target").exists()
assert (prefix / "var/lib/decnet").exists()
def test_purge_without_deinit_errors(tmp_path: Path) -> None:
result = runner.invoke(
app, ["init", "--purge", "--prefix", str(tmp_path / "root")],
)
assert result.exit_code == 1
assert "--purge only applies with --deinit" in result.output
def test_missing_deploy_dir_errors_clearly(monkeypatch: Any, tmp_path: Path) -> None:
def _boom() -> Path:
raise RuntimeError("cannot locate deploy/ directory (looked at /nope)")
monkeypatch.setattr(_init, "_deploy_root", _boom)
monkeypatch.setattr(_init.shutil, "which", lambda _: "/bin/x")
result = runner.invoke(
app, ["init", "--prefix", str(tmp_path / "root")],
)
assert result.exit_code == 1
assert "cannot locate deploy/" in result.output