Add full test suite with pytest
This commit is contained in:
@@ -0,0 +1,55 @@
|
||||
from __future__ import annotations
|
||||
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
|
||||
from chguard.cli import (
|
||||
_common_snapshot_root,
|
||||
_extract_paths_from_command,
|
||||
_parse_prune_states_value,
|
||||
)
|
||||
|
||||
|
||||
def test_parse_prune_states_value_accepts_days_and_all(monkeypatch) -> None:
|
||||
assert _parse_prune_states_value("14") == 14
|
||||
assert _parse_prune_states_value(" all ") == "all"
|
||||
|
||||
monkeypatch.setenv("CHGUARD_STATES_LIFE", "30")
|
||||
assert _parse_prune_states_value(None) == 30
|
||||
|
||||
|
||||
@pytest.mark.parametrize("value", ["", "abc", "-1"])
|
||||
def test_parse_prune_states_value_rejects_invalid_values(
|
||||
value: str, monkeypatch
|
||||
) -> None:
|
||||
monkeypatch.delenv("CHGUARD_STATES_LIFE", raising=False)
|
||||
|
||||
with pytest.raises(SystemExit):
|
||||
_parse_prune_states_value(value)
|
||||
|
||||
|
||||
def test_extract_paths_from_command_returns_existing_non_options(
|
||||
tmp_path: Path, monkeypatch
|
||||
) -> None:
|
||||
monkeypatch.chdir(tmp_path)
|
||||
target = tmp_path / "file.txt"
|
||||
target.write_text("data", encoding="utf-8")
|
||||
|
||||
paths = _extract_paths_from_command(
|
||||
["chmod", "-R", "644", "file.txt", "missing.txt"]
|
||||
)
|
||||
|
||||
assert paths == [target.resolve()]
|
||||
|
||||
|
||||
def test_common_snapshot_root_uses_single_path_or_common_parent(
|
||||
tmp_path: Path,
|
||||
) -> None:
|
||||
one = tmp_path / "one"
|
||||
two = tmp_path / "two"
|
||||
one.mkdir()
|
||||
two.mkdir()
|
||||
|
||||
assert _common_snapshot_root([one]) == one.resolve()
|
||||
assert _common_snapshot_root([one, two]) == tmp_path.resolve()
|
||||
@@ -0,0 +1,111 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
import sys
|
||||
from pathlib import Path
|
||||
from types import SimpleNamespace
|
||||
|
||||
from chguard.cli import _restore_preview_rows, main
|
||||
from chguard.db import connect, create_state, init_db
|
||||
from chguard.restore import PlannedChange
|
||||
|
||||
|
||||
class FakePath:
|
||||
def relative_to(self, root: Path) -> Path:
|
||||
return Path("link")
|
||||
|
||||
def lstat(self):
|
||||
return SimpleNamespace(st_uid=123)
|
||||
|
||||
def stat(self):
|
||||
raise AssertionError("restore preview must not follow symlinks")
|
||||
|
||||
|
||||
def test_missing_and_type_changes_are_reported_as_skipped() -> None:
|
||||
root = Path("/snapshot")
|
||||
changes = [
|
||||
PlannedChange(
|
||||
root / "deleted.conf",
|
||||
"missing",
|
||||
"path does not exist",
|
||||
False,
|
||||
),
|
||||
PlannedChange(root / "logs", "type", "file -> dir", False),
|
||||
]
|
||||
|
||||
rows, counts, needs_root = _restore_preview_rows(
|
||||
changes, root, current_uid=999
|
||||
)
|
||||
|
||||
assert not needs_root
|
||||
assert counts["skipped"] == 2
|
||||
assert rows[Path("deleted.conf")]["skipped"] == "missing path"
|
||||
assert rows[Path("logs")]["skipped"] == "found file, expected dir"
|
||||
|
||||
|
||||
def test_unselected_scope_changes_are_not_reported() -> None:
|
||||
root = Path("/snapshot")
|
||||
changes = [
|
||||
PlannedChange(root / "file.txt", "mode", "0o644 -> 0o600", False)
|
||||
]
|
||||
|
||||
rows, counts, needs_root = _restore_preview_rows(
|
||||
changes, root, current_uid=999
|
||||
)
|
||||
|
||||
assert rows == {}
|
||||
assert counts["mode"] == 0
|
||||
assert not needs_root
|
||||
|
||||
|
||||
def test_applicable_changes_use_lstat_for_privilege_check() -> None:
|
||||
changes = [PlannedChange(FakePath(), "mode", "0o644 -> 0o600", True)]
|
||||
|
||||
rows, counts, needs_root = _restore_preview_rows(
|
||||
changes, Path("/snapshot"), current_uid=999
|
||||
)
|
||||
|
||||
assert needs_root
|
||||
assert counts["mode"] == 1
|
||||
assert "rw-r--r--" in rows[Path("link")]["mode"]
|
||||
assert "rw-------" in rows[Path("link")]["mode"]
|
||||
|
||||
|
||||
def test_restore_with_only_skipped_items_does_not_prompt(
|
||||
tmp_path: Path, monkeypatch, capsys
|
||||
) -> None:
|
||||
root = tmp_path / "root"
|
||||
root.mkdir()
|
||||
|
||||
conn = connect(tmp_path / "states.db")
|
||||
init_db(conn)
|
||||
with conn:
|
||||
state_id = create_state(
|
||||
conn, "baseline", str(root), os.getuid(), commit=False
|
||||
)
|
||||
conn.execute(
|
||||
"""
|
||||
INSERT INTO entries (state_id, path, type, mode, uid, gid)
|
||||
VALUES (?, ?, ?, ?, ?, ?)
|
||||
""",
|
||||
(state_id, "deleted.conf", "file", 0o644, os.getuid(), 0),
|
||||
)
|
||||
conn.close()
|
||||
|
||||
monkeypatch.setattr(
|
||||
sys,
|
||||
"argv",
|
||||
[
|
||||
"chguard",
|
||||
"--db",
|
||||
str(tmp_path / "states.db"),
|
||||
"--restore",
|
||||
"baseline",
|
||||
],
|
||||
)
|
||||
|
||||
main()
|
||||
|
||||
output = capsys.readouterr().out
|
||||
assert "missing path" in output
|
||||
assert "No applicable changes" in output
|
||||
@@ -0,0 +1,76 @@
|
||||
from __future__ import annotations
|
||||
|
||||
from pathlib import Path
|
||||
|
||||
from chguard.db import (
|
||||
connect,
|
||||
create_state,
|
||||
delete_state,
|
||||
get_state,
|
||||
init_db,
|
||||
prune_all_states,
|
||||
prune_states_before,
|
||||
state_exists,
|
||||
)
|
||||
|
||||
|
||||
def test_state_crud_and_entry_cascade_delete(tmp_path: Path) -> None:
|
||||
conn = connect(tmp_path / "states.db")
|
||||
init_db(conn)
|
||||
|
||||
state_id = create_state(conn, "baseline", "/srv/app", 1000)
|
||||
conn.execute(
|
||||
"""
|
||||
INSERT INTO entries (state_id, path, type, mode, uid, gid)
|
||||
VALUES (?, ?, ?, ?, ?, ?)
|
||||
""",
|
||||
(state_id, "config.txt", "file", 0o644, 1000, 1000),
|
||||
)
|
||||
conn.commit()
|
||||
|
||||
assert state_exists(conn, "baseline")
|
||||
state = get_state(conn, "baseline")
|
||||
assert state is not None
|
||||
assert state.id == state_id
|
||||
assert state.root_path == "/srv/app"
|
||||
|
||||
assert delete_state(conn, "baseline") == 1
|
||||
assert not state_exists(conn, "baseline")
|
||||
remaining_entries = conn.execute("SELECT COUNT(*) FROM entries").fetchone()
|
||||
assert remaining_entries[0] == 0
|
||||
|
||||
|
||||
def test_prune_states_before_deletes_only_old_states(tmp_path: Path) -> None:
|
||||
conn = connect(tmp_path / "states.db")
|
||||
init_db(conn)
|
||||
conn.execute(
|
||||
"""
|
||||
INSERT INTO states (name, root_path, created_at, created_by_uid)
|
||||
VALUES (?, ?, ?, ?)
|
||||
""",
|
||||
("old", "/old", "2024-01-01T00:00:00+00:00", 1000),
|
||||
)
|
||||
conn.execute(
|
||||
"""
|
||||
INSERT INTO states (name, root_path, created_at, created_by_uid)
|
||||
VALUES (?, ?, ?, ?)
|
||||
""",
|
||||
("new", "/new", "2024-02-01T00:00:00+00:00", 1000),
|
||||
)
|
||||
conn.commit()
|
||||
|
||||
deleted = prune_states_before(conn, "2024-01-15T00:00:00+00:00")
|
||||
|
||||
assert deleted == 1
|
||||
assert not state_exists(conn, "old")
|
||||
assert state_exists(conn, "new")
|
||||
|
||||
|
||||
def test_prune_all_states_deletes_every_state(tmp_path: Path) -> None:
|
||||
conn = connect(tmp_path / "states.db")
|
||||
init_db(conn)
|
||||
create_state(conn, "one", "/one", 1000)
|
||||
create_state(conn, "two", "/two", 1000)
|
||||
|
||||
assert prune_all_states(conn) == 2
|
||||
assert conn.execute("SELECT COUNT(*) FROM states").fetchone()[0] == 0
|
||||
@@ -0,0 +1,105 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
import stat
|
||||
from pathlib import Path
|
||||
|
||||
from chguard.restore import apply_restore, plan_restore
|
||||
|
||||
|
||||
def test_plan_restore_reports_mode_owner_missing_and_type_changes(
|
||||
tmp_path: Path,
|
||||
) -> None:
|
||||
root = tmp_path / "root"
|
||||
root.mkdir()
|
||||
config = root / "config.txt"
|
||||
config.write_text("config", encoding="utf-8")
|
||||
config.chmod(0o644)
|
||||
logs = root / "logs"
|
||||
logs.write_text("not a directory", encoding="utf-8")
|
||||
|
||||
current = config.lstat()
|
||||
rows = [
|
||||
("config.txt", "file", 0o600, current.st_uid + 1, current.st_gid),
|
||||
("missing.txt", "file", 0o644, current.st_uid, current.st_gid),
|
||||
("logs", "dir", 0o755, current.st_uid, current.st_gid),
|
||||
]
|
||||
|
||||
changes = plan_restore(
|
||||
root=root,
|
||||
rows=rows,
|
||||
restore_permissions=True,
|
||||
restore_owner=True,
|
||||
)
|
||||
|
||||
kinds_by_path = {(change.path.name, change.kind) for change in changes}
|
||||
assert ("config.txt", "mode") in kinds_by_path
|
||||
assert ("config.txt", "owner") in kinds_by_path
|
||||
assert ("missing.txt", "missing") in kinds_by_path
|
||||
assert ("logs", "type") in kinds_by_path
|
||||
|
||||
|
||||
def test_plan_restore_marks_unselected_scope_as_not_applicable(
|
||||
tmp_path: Path,
|
||||
) -> None:
|
||||
root = tmp_path / "root"
|
||||
root.mkdir()
|
||||
config = root / "config.txt"
|
||||
config.write_text("config", encoding="utf-8")
|
||||
config.chmod(0o644)
|
||||
st = config.lstat()
|
||||
|
||||
changes = plan_restore(
|
||||
root=root,
|
||||
rows=[("config.txt", "file", 0o600, st.st_uid, st.st_gid)],
|
||||
restore_permissions=False,
|
||||
restore_owner=True,
|
||||
)
|
||||
|
||||
assert len(changes) == 1
|
||||
assert changes[0].kind == "mode"
|
||||
assert not changes[0].will_apply
|
||||
|
||||
|
||||
def test_apply_restore_changes_mode_without_changing_contents(
|
||||
tmp_path: Path,
|
||||
) -> None:
|
||||
root = tmp_path / "root"
|
||||
root.mkdir()
|
||||
config = root / "config.txt"
|
||||
config.write_text("config", encoding="utf-8")
|
||||
config.chmod(0o644)
|
||||
st = config.lstat()
|
||||
|
||||
apply_restore(
|
||||
root=root,
|
||||
rows=[("config.txt", "file", 0o600, st.st_uid, st.st_gid)],
|
||||
restore_permissions=True,
|
||||
restore_owner=False,
|
||||
)
|
||||
|
||||
assert stat.S_IMODE(config.lstat().st_mode) == 0o600
|
||||
assert config.read_text(encoding="utf-8") == "config"
|
||||
|
||||
|
||||
def test_apply_restore_skips_missing_and_type_mismatch(tmp_path: Path) -> None:
|
||||
root = tmp_path / "root"
|
||||
root.mkdir()
|
||||
config = root / "config.txt"
|
||||
config.write_text("config", encoding="utf-8")
|
||||
config.chmod(0o644)
|
||||
st = config.lstat()
|
||||
|
||||
apply_restore(
|
||||
root=root,
|
||||
rows=[
|
||||
("missing.txt", "file", 0o600, st.st_uid, st.st_gid),
|
||||
("config.txt", "dir", 0o600, st.st_uid, st.st_gid),
|
||||
],
|
||||
restore_permissions=True,
|
||||
restore_owner=False,
|
||||
)
|
||||
|
||||
assert not (root / "missing.txt").exists()
|
||||
assert stat.S_IMODE(config.lstat().st_mode) == 0o644
|
||||
assert os.listdir(root) == ["config.txt"]
|
||||
@@ -0,0 +1,67 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import stat
|
||||
from pathlib import Path
|
||||
|
||||
from chguard.scan import scan_tree
|
||||
|
||||
|
||||
def entries_by_path(root: Path, excludes: tuple[str, ...] = ()):
|
||||
return {entry.path: entry for entry in scan_tree(root, excludes=excludes)}
|
||||
|
||||
|
||||
def test_scan_tree_records_root_dirs_files_and_symlinks(
|
||||
tmp_path: Path,
|
||||
) -> None:
|
||||
root = tmp_path / "root"
|
||||
root.mkdir()
|
||||
data_dir = root / "data"
|
||||
data_dir.mkdir()
|
||||
file_path = data_dir / "config.txt"
|
||||
file_path.write_text("config", encoding="utf-8")
|
||||
file_path.chmod(0o640)
|
||||
(root / "config-link").symlink_to(file_path)
|
||||
|
||||
entries = entries_by_path(root)
|
||||
|
||||
assert entries[""].type == "dir"
|
||||
assert entries["data"].type == "dir"
|
||||
assert entries["data/config.txt"].type == "file"
|
||||
assert entries["data/config.txt"].mode == 0o640
|
||||
assert entries["config-link"].type == "symlink"
|
||||
assert entries["config-link"].mode == stat.S_IMODE(
|
||||
(root / "config-link").lstat().st_mode
|
||||
)
|
||||
|
||||
|
||||
def test_scan_tree_excludes_path_prefixes(tmp_path: Path) -> None:
|
||||
root = tmp_path / "root"
|
||||
root.mkdir()
|
||||
(root / "keep").mkdir()
|
||||
(root / "keep" / "file.txt").write_text("keep", encoding="utf-8")
|
||||
(root / "cache").mkdir()
|
||||
(root / "cache" / "file.txt").write_text("cache", encoding="utf-8")
|
||||
(root / "var").mkdir()
|
||||
(root / "var" / "tmp").mkdir()
|
||||
(root / "var" / "tmp" / "file.txt").write_text("tmp", encoding="utf-8")
|
||||
|
||||
entries = entries_by_path(root, excludes=("cache", "var/tmp"))
|
||||
|
||||
assert "keep" in entries
|
||||
assert "keep/file.txt" in entries
|
||||
assert "cache" not in entries
|
||||
assert "cache/file.txt" not in entries
|
||||
assert "var" in entries
|
||||
assert "var/tmp" not in entries
|
||||
assert "var/tmp/file.txt" not in entries
|
||||
|
||||
|
||||
def test_scan_tree_file_root_records_only_root_entry(tmp_path: Path) -> None:
|
||||
root_file = tmp_path / "single.txt"
|
||||
root_file.write_text("single", encoding="utf-8")
|
||||
|
||||
entries = list(scan_tree(root_file))
|
||||
|
||||
assert len(entries) == 1
|
||||
assert entries[0].path == ""
|
||||
assert entries[0].type == "file"
|
||||
Reference in New Issue
Block a user