mirror of
https://github.com/kiviktnm/decman.git
synced 2026-09-19 12:08:28 +00:00
Better methods for running commands
This commit is contained in:
Kivi Kaitaniemi
@@ -1,6 +0,0 @@
|
||||
import os
|
||||
import sys
|
||||
|
||||
_SRC_PATH = os.path.join(os.path.dirname(__file__), "../src/")
|
||||
|
||||
sys.path.append(_SRC_PATH)
|
||||
@@ -1,3 +0,0 @@
|
||||
Simple text file with a %variable%
|
||||
|
||||
twice: %another_variable%
|
||||
@@ -1,3 +0,0 @@
|
||||
#!/usr/bin/env bash
|
||||
# This file should be executable.
|
||||
echo "Hello, world!"
|
||||
@@ -1,3 +0,0 @@
|
||||
1
|
||||
1
|
||||
1
|
||||
@@ -1,3 +0,0 @@
|
||||
2
|
||||
2
|
||||
2
|
||||
Binary file not shown.
|
Before Width: | Height: | Size: 1.9 KiB |
@@ -1,3 +0,0 @@
|
||||
s1
|
||||
s1
|
||||
s1
|
||||
@@ -1,3 +0,0 @@
|
||||
s2
|
||||
s2
|
||||
s2
|
||||
@@ -1,56 +0,0 @@
|
||||
import os
|
||||
import sys
|
||||
|
||||
# This test is manual. You'll have to verify the results manually.
|
||||
|
||||
# NOTE: Change this if you want to run this script.
|
||||
user = "kk"
|
||||
|
||||
cd = os.path.dirname(os.path.abspath(__file__))
|
||||
os.chdir(cd)
|
||||
|
||||
sys.path.append(os.path.join(cd, "../../src/."))
|
||||
|
||||
from decman import Directory, File
|
||||
|
||||
# if os.path.exists("/tmp/decman-files"):
|
||||
# shutil.rmtree("/tmp/decman-files")
|
||||
# os.makedirs("/tmp/decman-files")
|
||||
|
||||
f1 = File(source_file="src/f1.txt")
|
||||
f1.copy_to(
|
||||
"/tmp/decman-files/f1.txt",
|
||||
variables={"%variable%": "123", "%another_variable%": "456"},
|
||||
)
|
||||
|
||||
f2 = File(source_file="src/f2.sh", permissions=0o744)
|
||||
f2.copy_to(
|
||||
"/tmp/decman-files/f2.sh",
|
||||
)
|
||||
|
||||
f3 = File(content="%variable% doesn't work here.", bin_file=True)
|
||||
f3.copy_to(
|
||||
"/tmp/decman-files/f3.txt",
|
||||
variables={
|
||||
"%variable%": "123",
|
||||
},
|
||||
)
|
||||
|
||||
f4 = File(content="%variable% works here.", bin_file=False, owner=user)
|
||||
f4.copy_to(
|
||||
"/tmp/decman-files/f4.txt",
|
||||
variables={
|
||||
"%variable%": "123",
|
||||
},
|
||||
)
|
||||
|
||||
f5 = File(content="%variable% works here.", bin_file=False, owner=user, group="root")
|
||||
f5.copy_to(
|
||||
"/tmp/decman-files/f5.txt",
|
||||
variables={
|
||||
"%variable%": "123",
|
||||
},
|
||||
)
|
||||
|
||||
d = Directory("src/srcdir", bin_files=True)
|
||||
d.copy_to("/tmp/decman-files/targetdir")
|
||||
@@ -0,0 +1,70 @@
|
||||
import json
|
||||
import sys
|
||||
|
||||
import pytest
|
||||
|
||||
import decman.core.command as command
|
||||
|
||||
|
||||
def test_run_simple():
|
||||
code, out = command.run([sys.executable, "-c", "print('ok')"])
|
||||
assert code == 0
|
||||
assert out.strip() == "ok"
|
||||
|
||||
|
||||
def test_run_exec_failure():
|
||||
code, out = command.run(["/does/not/exist"])
|
||||
assert code != 0
|
||||
assert "not" in out.lower()
|
||||
|
||||
|
||||
def test_run_env_overrides_and_mimic_login_visible_in_child(monkeypatch):
|
||||
class FakePw:
|
||||
pw_dir = "/fake/home"
|
||||
pw_name = "fakeuser"
|
||||
pw_uid = 1000
|
||||
pw_gid = 1000
|
||||
pw_shell = "/bin/fakesh"
|
||||
|
||||
# Mock passwd lookup
|
||||
monkeypatch.setattr(
|
||||
"decman.core.command.pwd.getpwnam",
|
||||
lambda user: FakePw(),
|
||||
)
|
||||
|
||||
code, out = command.run(
|
||||
[
|
||||
sys.executable,
|
||||
"-c",
|
||||
(
|
||||
"import os, json; "
|
||||
"print(json.dumps({"
|
||||
"'FOO': os.environ['FOO'], "
|
||||
"'HOME': os.environ['HOME'], "
|
||||
"'USER': os.environ['USER'], "
|
||||
"'LOGNAME': os.environ['LOGNAME'], "
|
||||
"'SHELL': os.environ['SHELL']"
|
||||
"}))"
|
||||
),
|
||||
],
|
||||
user="fakeuser",
|
||||
mimic_login=True,
|
||||
env_overrides={"FOO": "BAR"},
|
||||
)
|
||||
|
||||
assert code == 0
|
||||
|
||||
data = json.loads(out.strip())
|
||||
assert data["FOO"] == "BAR"
|
||||
assert data["HOME"] == "/fake/home"
|
||||
assert data["USER"] == "fakeuser"
|
||||
assert data["LOGNAME"] == "fakeuser"
|
||||
assert data["SHELL"] == "/bin/fakesh"
|
||||
|
||||
|
||||
@pytest.mark.skipif(not sys.stdin.isatty(), reason="requires TTY")
|
||||
def test_pty_run_simple():
|
||||
code, out = command.pty_run([sys.executable, "-c", "print('ok')"])
|
||||
assert code == 0
|
||||
assert "ok" in out
|
||||
assert "\r\n" not in out
|
||||
@@ -0,0 +1,244 @@
|
||||
import os
|
||||
import stat
|
||||
from pathlib import Path
|
||||
|
||||
# Adjust this import to match your actual module location
|
||||
import decman.core.files as files
|
||||
|
||||
# --- files.File tests --------------------------------------------------------------
|
||||
|
||||
|
||||
def test_file_from_content_creates_and_is_idempotent(tmp_path: Path) -> None:
|
||||
target = tmp_path / "file.txt"
|
||||
|
||||
f = files.File(content="hello", permissions=0o600)
|
||||
|
||||
# First run: file must be created and reported as changed
|
||||
changed1 = f.copy_to(str(target))
|
||||
assert changed1 is True
|
||||
assert target.read_text(encoding="utf-8") == "hello"
|
||||
|
||||
mode = stat.S_IMODE(target.stat().st_mode)
|
||||
assert mode == 0o600
|
||||
|
||||
# Second run with same configuration: no content change
|
||||
changed2 = f.copy_to(str(target))
|
||||
assert changed2 is False
|
||||
assert target.read_text(encoding="utf-8") == "hello"
|
||||
assert stat.S_IMODE(target.stat().st_mode) == 0o600
|
||||
|
||||
|
||||
def test_file_content_with_variables_and_change_detection(tmp_path: Path) -> None:
|
||||
target = tmp_path / "templated.txt"
|
||||
|
||||
f = files.File(content="hello {{NAME}}")
|
||||
|
||||
# First run: NAME=world
|
||||
changed1 = f.copy_to(str(target), {"{{NAME}}": "world"})
|
||||
assert changed1 is True
|
||||
assert target.read_text(encoding="utf-8") == "hello world"
|
||||
|
||||
# Second run: same variables, no change
|
||||
changed2 = f.copy_to(str(target), {"{{NAME}}": "world"})
|
||||
assert changed2 is False
|
||||
assert target.read_text(encoding="utf-8") == "hello world"
|
||||
|
||||
# Third run: different variables, should change
|
||||
changed3 = f.copy_to(str(target), {"{{NAME}}": "there"})
|
||||
assert changed3 is True
|
||||
assert target.read_text(encoding="utf-8") == "hello there"
|
||||
|
||||
|
||||
def test_file_from_source_text_with_and_without_variables(tmp_path: Path) -> None:
|
||||
src = tmp_path / "src.txt"
|
||||
src.write_text("VALUE={{X}}", encoding="utf-8")
|
||||
target = tmp_path / "dst.txt"
|
||||
|
||||
# Without variables (raw copy)
|
||||
f_raw = files.File(source_file=str(src))
|
||||
changed1 = f_raw.copy_to(str(target), {})
|
||||
assert changed1 is True
|
||||
assert target.read_text(encoding="utf-8") == "VALUE={{X}}"
|
||||
|
||||
# Idempotent raw copy
|
||||
changed2 = f_raw.copy_to(str(target), {})
|
||||
assert changed2 is False
|
||||
|
||||
# With variables (substitution)
|
||||
f_sub = files.File(source_file=str(src))
|
||||
changed3 = f_sub.copy_to(str(target), {"{{X}}": "42"})
|
||||
assert changed3 is True
|
||||
assert target.read_text(encoding="utf-8") == "VALUE=42"
|
||||
|
||||
# Idempotent after substitution
|
||||
changed4 = f_sub.copy_to(str(target), {"{{X}}": "42"})
|
||||
assert changed4 is False
|
||||
|
||||
|
||||
def test_file_binary_from_content(tmp_path: Path) -> None:
|
||||
target = tmp_path / "bin.dat"
|
||||
payload = b"\x00\x01\x02hello"
|
||||
|
||||
f = files.File(content=payload.decode("latin1"), bin_file=True)
|
||||
|
||||
changed1 = f.copy_to(str(target))
|
||||
assert changed1 is True
|
||||
assert target.read_bytes() == payload
|
||||
|
||||
# Idempotent: second call does not rewrite
|
||||
changed2 = f.copy_to(str(target))
|
||||
assert changed2 is False
|
||||
assert target.read_bytes() == payload
|
||||
|
||||
|
||||
def test_file_binary_copy_from_source(tmp_path: Path) -> None:
|
||||
src = tmp_path / "src.bin"
|
||||
payload = b"\x10\x20\x30binary"
|
||||
src.write_bytes(payload)
|
||||
target = tmp_path / "dst.bin"
|
||||
|
||||
f = files.File(source_file=str(src), bin_file=True)
|
||||
|
||||
changed1 = f.copy_to(str(target), {"IGNORED": "x"})
|
||||
assert changed1 is True
|
||||
assert target.read_bytes() == payload
|
||||
|
||||
# Idempotent, comparing bytes
|
||||
changed2 = f.copy_to(str(target), {"IGNORED": "x"})
|
||||
assert changed2 is False
|
||||
assert target.read_bytes() == payload
|
||||
|
||||
|
||||
def test_file_creates_parent_directories_and_applies_permissions(tmp_path: Path) -> None:
|
||||
nested_dir = tmp_path / "a" / "b" / "c"
|
||||
target = nested_dir / "file.txt"
|
||||
|
||||
f = files.File(content="data", permissions=0o644)
|
||||
|
||||
changed = f.copy_to(str(target))
|
||||
assert changed is True
|
||||
assert target.read_text(encoding="utf-8") == "data"
|
||||
|
||||
# Directories created
|
||||
assert nested_dir.is_dir()
|
||||
|
||||
# Permissions on file
|
||||
mode = stat.S_IMODE(target.stat().st_mode)
|
||||
assert mode == 0o644
|
||||
|
||||
|
||||
# --- files.Directory tests ---------------------------------------------------------
|
||||
|
||||
|
||||
def _create_sample_source_tree(root: Path) -> None:
|
||||
(root / "sub").mkdir(parents=True)
|
||||
(root / "a.txt").write_text("A={{X}}", encoding="utf-8")
|
||||
(root / "sub" / "b.txt").write_text("B={{X}}", encoding="utf-8")
|
||||
|
||||
|
||||
def test_directory_copy_to_creates_and_is_idempotent(tmp_path: Path) -> None:
|
||||
src_dir = tmp_path / "src"
|
||||
dst_dir = tmp_path / "dst"
|
||||
src_dir.mkdir()
|
||||
|
||||
_create_sample_source_tree(src_dir)
|
||||
|
||||
d = files.Directory(
|
||||
source_directory=str(src_dir),
|
||||
bin_files=False,
|
||||
encoding="utf-8",
|
||||
permissions=0o644,
|
||||
)
|
||||
|
||||
# First run: both files should be created and reported as changed
|
||||
changed1 = d.copy_to(str(dst_dir), variables={"{{X}}": "1"})
|
||||
expected_paths = {
|
||||
str(dst_dir / "a.txt"),
|
||||
str(dst_dir / "sub" / "b.txt"),
|
||||
}
|
||||
assert set(changed1) == expected_paths
|
||||
|
||||
assert (dst_dir / "a.txt").read_text(encoding="utf-8") == "A=1"
|
||||
assert (dst_dir / "sub" / "b.txt").read_text(encoding="utf-8") == "B=1"
|
||||
|
||||
# Second run with same variables: no files should be reported as changed
|
||||
changed2 = d.copy_to(str(dst_dir), variables={"{{X}}": "1"})
|
||||
assert changed2 == []
|
||||
|
||||
|
||||
def test_directory_copy_to_detects_changes_via_variables(tmp_path: Path) -> None:
|
||||
src_dir = tmp_path / "src"
|
||||
dst_dir = tmp_path / "dst"
|
||||
src_dir.mkdir()
|
||||
_create_sample_source_tree(src_dir)
|
||||
|
||||
d = files.Directory(source_directory=str(src_dir))
|
||||
|
||||
# Initial materialization
|
||||
changed1 = d.copy_to(str(dst_dir), variables={"{{X}}": "alpha"})
|
||||
assert set(changed1) == {
|
||||
str(dst_dir / "a.txt"),
|
||||
str(dst_dir / "sub" / "b.txt"),
|
||||
}
|
||||
|
||||
# Change variables -> both files change
|
||||
changed2 = d.copy_to(str(dst_dir), variables={"{{X}}": "beta"})
|
||||
assert set(changed2) == {
|
||||
str(dst_dir / "a.txt"),
|
||||
str(dst_dir / "sub" / "b.txt"),
|
||||
}
|
||||
|
||||
assert (dst_dir / "a.txt").read_text(encoding="utf-8") == "A=beta"
|
||||
assert (dst_dir / "sub" / "b.txt").read_text(encoding="utf-8") == "B=beta"
|
||||
|
||||
|
||||
def test_directory_copy_to_dry_run(tmp_path: Path) -> None:
|
||||
src_dir = tmp_path / "src"
|
||||
dst_dir = tmp_path / "dst"
|
||||
src_dir.mkdir()
|
||||
_create_sample_source_tree(src_dir)
|
||||
|
||||
d = files.Directory(source_directory=str(src_dir))
|
||||
|
||||
# First, actually materialize once
|
||||
d.copy_to(str(dst_dir), variables={"{{X}}": "1"})
|
||||
|
||||
# Now perform dry-run with different variables; contents must not change
|
||||
before_a = (dst_dir / "a.txt").read_text(encoding="utf-8")
|
||||
before_b = (dst_dir / "sub" / "b.txt").read_text(encoding="utf-8")
|
||||
|
||||
changed_dry = d.copy_to(
|
||||
str(dst_dir),
|
||||
variables={"{{X}}": "2"},
|
||||
dry_run=True,
|
||||
)
|
||||
|
||||
expected_paths = {
|
||||
str(dst_dir / "a.txt"),
|
||||
str(dst_dir / "sub" / "b.txt"),
|
||||
}
|
||||
assert set(changed_dry) == expected_paths
|
||||
|
||||
# Contents remain as before (no writes in dry-run)
|
||||
assert (dst_dir / "a.txt").read_text(encoding="utf-8") == before_a
|
||||
assert (dst_dir / "sub" / "b.txt").read_text(encoding="utf-8") == before_b
|
||||
|
||||
|
||||
def test_directory_copy_to_restores_working_directory(tmp_path: Path) -> None:
|
||||
src_dir = tmp_path / "src"
|
||||
dst_dir = tmp_path / "dst"
|
||||
src_dir.mkdir()
|
||||
_create_sample_source_tree(src_dir)
|
||||
|
||||
d = files.Directory(source_directory=str(src_dir))
|
||||
|
||||
original_cwd = os.getcwd()
|
||||
try:
|
||||
changed = d.copy_to(str(dst_dir), variables={"{{X}}": "x"})
|
||||
assert set(changed) == {
|
||||
str(dst_dir / "a.txt"),
|
||||
str(dst_dir / "sub" / "b.txt"),
|
||||
}
|
||||
finally:
|
||||
# Ensure the implementation restored CWD
|
||||
assert os.getcwd() == original_cwd
|
||||
@@ -0,0 +1,191 @@
|
||||
import builtins
|
||||
import types
|
||||
|
||||
import pytest
|
||||
|
||||
import decman.config as config
|
||||
import decman.core.output as output
|
||||
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def reset_config():
|
||||
# snapshot & restore config flags between tests
|
||||
orig = types.SimpleNamespace(
|
||||
debug_output=getattr(config, "debug_output", False),
|
||||
quiet_output=getattr(config, "quiet_output", False),
|
||||
color_output=getattr(config, "color_output", True),
|
||||
)
|
||||
yield
|
||||
config.debug_output = orig.debug_output
|
||||
config.quiet_output = orig.quiet_output
|
||||
config.color_output = orig.color_output
|
||||
|
||||
|
||||
def test_print_error_with_color_enabled(capsys):
|
||||
config.color_output = True
|
||||
output.print_error("boom")
|
||||
|
||||
out = capsys.readouterr().out
|
||||
assert "boom" in out
|
||||
assert "ERROR" in out
|
||||
# crude check that some ANSI escapes are present
|
||||
assert "\x1b[" in out
|
||||
|
||||
|
||||
def test_print_error_with_color_disabled(capsys):
|
||||
config.color_output = False
|
||||
output.print_error("boom")
|
||||
|
||||
out = capsys.readouterr().out
|
||||
assert out.strip().endswith("ERROR: boom")
|
||||
# no ANSI escapes
|
||||
assert "\x1b[" not in out
|
||||
|
||||
|
||||
def test_print_info_respects_quiet_and_debug(capsys):
|
||||
config.quiet_output = True
|
||||
config.debug_output = False
|
||||
|
||||
output.print_info("msg 1")
|
||||
out = capsys.readouterr().out
|
||||
assert out == "" # suppressed
|
||||
|
||||
config.debug_output = True
|
||||
output.print_info("msg 2")
|
||||
out = capsys.readouterr().out
|
||||
assert "INFO: msg 2" in out
|
||||
|
||||
config.quiet_output = False
|
||||
config.debug_output = False
|
||||
output.print_info("msg 3")
|
||||
out = capsys.readouterr().out
|
||||
assert "INFO: msg 3" in out
|
||||
|
||||
|
||||
def test_print_debug_only_with_debug_enabled(capsys):
|
||||
config.debug_output = False
|
||||
output.print_debug("dbg")
|
||||
assert capsys.readouterr().out == ""
|
||||
|
||||
config.debug_output = True
|
||||
output.print_debug("dbg")
|
||||
out = capsys.readouterr().out
|
||||
assert "DEBUG" in out
|
||||
assert "dbg" in out
|
||||
|
||||
|
||||
def test_print_continuation_respects_level_and_config(capsys):
|
||||
config.quiet_output = True
|
||||
config.debug_output = False
|
||||
|
||||
output.print_continuation("x", level=output.INFO)
|
||||
assert capsys.readouterr().out == ""
|
||||
|
||||
output.print_continuation("y", level=output.SUMMARY)
|
||||
out = capsys.readouterr().out
|
||||
assert "y" in out
|
||||
|
||||
|
||||
def test_print_list_empty_outputs_nothing(capsys):
|
||||
output.print_list("Header", [])
|
||||
assert capsys.readouterr().out == ""
|
||||
|
||||
|
||||
def test_print_list_summary_and_elements(capsys, monkeypatch):
|
||||
# fixed terminal size for deterministic wrapping
|
||||
monkeypatch.setattr(
|
||||
output.shutil, "get_terminal_size", lambda: types.SimpleNamespace(columns=80)
|
||||
)
|
||||
config.quiet_output = False
|
||||
config.debug_output = False
|
||||
|
||||
output.print_list("Installed packages:", ["a", "b", "c"])
|
||||
|
||||
out = capsys.readouterr().out.splitlines()
|
||||
# header summary
|
||||
assert any("SUMMARY" in line and "Installed packages:" in line for line in out)
|
||||
# list content printed as continuation lines
|
||||
assert any("a" in line for line in out)
|
||||
assert any("b" in line for line in out)
|
||||
assert any("c" in line for line in out)
|
||||
|
||||
|
||||
def test_print_list_respects_elements_per_line_and_width(capsys, monkeypatch):
|
||||
# very small width to force wrapping
|
||||
monkeypatch.setattr(
|
||||
output.shutil, "get_terminal_size", lambda: types.SimpleNamespace(columns=30)
|
||||
)
|
||||
|
||||
items = [f"pkg{i}" for i in range(5)]
|
||||
output.print_list(
|
||||
"Pkgs:",
|
||||
items,
|
||||
elements_per_line=2,
|
||||
limit_to_term_size=True,
|
||||
level=output.SUMMARY,
|
||||
)
|
||||
|
||||
out_lines = capsys.readouterr().out.splitlines()
|
||||
list_lines = [l for l in out_lines if "pkg" in l]
|
||||
# at most 2 per line
|
||||
for line in list_lines:
|
||||
assert len([p for p in items if p in line]) <= 2
|
||||
|
||||
|
||||
def test_prompt_number_valid_input(monkeypatch):
|
||||
inputs = iter(["3"])
|
||||
monkeypatch.setattr(builtins, "input", lambda _: next(inputs))
|
||||
|
||||
res = output.prompt_number("Pick", 1, 5)
|
||||
assert res == 3
|
||||
|
||||
|
||||
def test_prompt_number_invalid_then_valid(monkeypatch, capsys):
|
||||
inputs = iter(["foo", "10", "2"])
|
||||
monkeypatch.setattr(builtins, "input", lambda _: next(inputs))
|
||||
|
||||
res = output.prompt_number("Pick", 1, 5)
|
||||
assert res == 2
|
||||
|
||||
out = capsys.readouterr().out
|
||||
# at least one error printed
|
||||
assert "Invalid input" in out
|
||||
|
||||
|
||||
def test_prompt_number_default_on_empty(monkeypatch):
|
||||
inputs = iter([""])
|
||||
monkeypatch.setattr(builtins, "input", lambda _: next(inputs))
|
||||
|
||||
res = output.prompt_number("Pick", 1, 5, default=4)
|
||||
assert res == 4
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"user_input,default,expected",
|
||||
[
|
||||
("y", None, True),
|
||||
("Y", None, True),
|
||||
("yes", None, True),
|
||||
("n", None, False),
|
||||
("No", None, False),
|
||||
("", True, True),
|
||||
("", False, False),
|
||||
],
|
||||
)
|
||||
def test_prompt_confirm(monkeypatch, user_input, default, expected):
|
||||
inputs = iter([user_input])
|
||||
monkeypatch.setattr(builtins, "input", lambda _: next(inputs))
|
||||
|
||||
res = output.prompt_confirm("Continue?", default=default)
|
||||
assert res is expected
|
||||
|
||||
|
||||
def test_prompt_confirm_invalid_then_yes(monkeypatch, capsys):
|
||||
inputs = iter(["maybe", "y"])
|
||||
monkeypatch.setattr(builtins, "input", lambda _: next(inputs))
|
||||
|
||||
res = output.prompt_confirm("Continue?")
|
||||
assert res is True
|
||||
|
||||
out = capsys.readouterr().out
|
||||
assert "Invalid input." in out
|
||||
@@ -0,0 +1,143 @@
|
||||
import typing
|
||||
|
||||
import pytest
|
||||
|
||||
import decman
|
||||
|
||||
|
||||
def test_prg_pty_true_uses_pty_run_and_check(monkeypatch: pytest.MonkeyPatch):
|
||||
calls: dict[str, typing.Any] = {}
|
||||
|
||||
def fake_pty_run(cmd, user=None, env_overrides=None, mimic_login=False):
|
||||
calls["pty_run"] = (cmd, user, env_overrides, mimic_login)
|
||||
return 0, "ok"
|
||||
|
||||
def fake_check_run_result(cmd, result):
|
||||
calls["check_run_result"] = (cmd, result)
|
||||
return result
|
||||
|
||||
def fake_print_warning(msg: str):
|
||||
raise AssertionError("print_warning must not be called when code == 0")
|
||||
|
||||
monkeypatch.setattr(decman, "command", decman.command)
|
||||
monkeypatch.setattr(decman.command, "pty_run", fake_pty_run)
|
||||
monkeypatch.setattr(decman.command, "check_run_result", fake_check_run_result)
|
||||
monkeypatch.setattr(decman, "output", decman.output)
|
||||
monkeypatch.setattr(decman.output, "print_warning", fake_print_warning)
|
||||
|
||||
out = decman.prg(
|
||||
["echo", "hi"],
|
||||
user="alice",
|
||||
env_overrides={"FOO": "bar"},
|
||||
mimic_login=True,
|
||||
pty=True,
|
||||
check=True,
|
||||
)
|
||||
|
||||
assert out == "ok"
|
||||
assert calls["pty_run"] == (["echo", "hi"], "alice", {"FOO": "bar"}, True)
|
||||
assert calls["check_run_result"] == (["echo", "hi"], (0, "ok"))
|
||||
|
||||
|
||||
def test_prg_pty_false_uses_run(monkeypatch: pytest.MonkeyPatch):
|
||||
calls: dict[str, typing.Any] = {}
|
||||
|
||||
def fake_run(cmd, user=None, env_overrides=None, mimic_login=False):
|
||||
calls["run"] = (cmd, user, env_overrides, mimic_login)
|
||||
return 0, "no-pty"
|
||||
|
||||
def fake_check_run_result(cmd, result):
|
||||
return result
|
||||
|
||||
def fake_print_warning(msg: str):
|
||||
raise AssertionError("print_warning must not be called when code == 0")
|
||||
|
||||
monkeypatch.setattr(decman.command, "run", fake_run)
|
||||
monkeypatch.setattr(decman.command, "check_run_result", fake_check_run_result)
|
||||
monkeypatch.setattr(decman.output, "print_warning", fake_print_warning)
|
||||
|
||||
out = decman.prg(["true"], pty=False, check=True)
|
||||
|
||||
assert out == "no-pty"
|
||||
assert calls["run"] == (["true"], None, None, False)
|
||||
|
||||
|
||||
def test_prg_check_false_warns_on_nonzero(monkeypatch: pytest.MonkeyPatch):
|
||||
calls: dict[str, typing.Any] = {}
|
||||
|
||||
def fake_run(cmd, user=None, env_overrides=None, mimic_login=False):
|
||||
# non-zero exit code
|
||||
return 3, "bad"
|
||||
|
||||
def fake_check_run_result(cmd, result):
|
||||
raise AssertionError("check_run_result must not be called when check=False")
|
||||
|
||||
def fake_print_warning(msg: str):
|
||||
calls["warning"] = msg
|
||||
|
||||
monkeypatch.setattr(decman.command, "run", fake_run)
|
||||
monkeypatch.setattr(decman.command, "check_run_result", fake_check_run_result)
|
||||
monkeypatch.setattr(decman.output, "print_warning", fake_print_warning)
|
||||
|
||||
out = decman.prg(["cmd", "arg"], pty=False, check=False)
|
||||
|
||||
assert out == "bad"
|
||||
assert "cmd arg" in calls["warning"]
|
||||
assert "exit code 3" in calls["warning"]
|
||||
|
||||
|
||||
def test_prg_check_true_propagates_command_failed_error(monkeypatch: pytest.MonkeyPatch):
|
||||
class CommandFailedError(Exception):
|
||||
pass
|
||||
|
||||
def fake_run(cmd, user=None, env_overrides=None, mimic_login=False):
|
||||
return 42, "boom"
|
||||
|
||||
def fake_check_run_result(cmd, result):
|
||||
raise CommandFailedError((cmd, result))
|
||||
|
||||
def fake_print_warning(msg: str):
|
||||
raise AssertionError("print_warning must not be called when check=True and error")
|
||||
|
||||
monkeypatch.setattr(decman.command, "run", fake_run)
|
||||
monkeypatch.setattr(decman.command, "check_run_result", fake_check_run_result)
|
||||
monkeypatch.setattr(decman.output, "print_warning", fake_print_warning)
|
||||
|
||||
with pytest.raises(CommandFailedError):
|
||||
decman.prg(["boom"], pty=False, check=True)
|
||||
|
||||
|
||||
def test_sh_calls_prg_with_sh_command(monkeypatch: pytest.MonkeyPatch):
|
||||
calls: dict[str, typing.Any] = {}
|
||||
|
||||
def fake_prg(
|
||||
cmd,
|
||||
user=None,
|
||||
env_overrides=None,
|
||||
mimic_login=False,
|
||||
pty=True,
|
||||
check=True,
|
||||
):
|
||||
calls["prg"] = (cmd, user, env_overrides, mimic_login, pty, check)
|
||||
return "output-from-prg"
|
||||
|
||||
monkeypatch.setattr(decman, "prg", fake_prg)
|
||||
|
||||
out = decman.sh(
|
||||
"echo test",
|
||||
user="bob",
|
||||
env_overrides={"X": "1"},
|
||||
mimic_login=True,
|
||||
pty=False,
|
||||
check=False,
|
||||
)
|
||||
|
||||
assert out == "output-from-prg"
|
||||
|
||||
cmd, user, env_overrides, mimic_login, pty, check = calls["prg"]
|
||||
assert cmd == ["/bin/sh", "-c", "echo test"]
|
||||
assert user == "bob"
|
||||
assert env_overrides == {"X": "1"}
|
||||
assert mimic_login is True
|
||||
assert pty is False
|
||||
assert check is False
|
||||
@@ -1,96 +0,0 @@
|
||||
# pylint: disable=missing-module-docstring,missing-class-docstring,missing-function-docstring
|
||||
|
||||
import unittest
|
||||
|
||||
from decman.error import UserFacingError
|
||||
from decman.lib import Pacman, Store
|
||||
from decman.lib.fpm import DepGraph, ExtendedPackageSearch, ForeignPackage, ForeignPackageManager
|
||||
|
||||
|
||||
class TestVersionComparisons(unittest.TestCase):
|
||||
def setUp(self):
|
||||
pacman = Pacman()
|
||||
self.pm = ForeignPackageManager(Store(), pacman, ExtendedPackageSearch(pacman))
|
||||
|
||||
def test_should_upgrade_package_returns_true_on_newer_version(self):
|
||||
self.assertTrue(self.pm.should_upgrade_package("test", "0.1.9", "0.2.0"))
|
||||
|
||||
def test_should_upgrade_package_returns_false_on_older_version(self):
|
||||
self.assertFalse(self.pm.should_upgrade_package("test", "0.1.9", "0.1.8"))
|
||||
|
||||
def test_should_upgrade_package_returns_false_on_same_version(self):
|
||||
self.assertFalse(self.pm.should_upgrade_package("test", "0.1.9", "0.1.9"))
|
||||
|
||||
def test_should_upgrade_package_returns_true_on_devel(self):
|
||||
self.assertTrue(self.pm.should_upgrade_package("test-git", "0", "0", upgrade_devel=True))
|
||||
|
||||
|
||||
class TestDepGraph(unittest.TestCase):
|
||||
def test_add_dependency(self):
|
||||
graph = DepGraph()
|
||||
|
||||
graph.add_requirement("A", None)
|
||||
graph.add_requirement("B1", "A")
|
||||
graph.add_requirement("B2", "A")
|
||||
graph.add_requirement("C", "B1")
|
||||
|
||||
self.assertIn("B1", graph.package_nodes["A"].children)
|
||||
self.assertIn("B2", graph.package_nodes["A"].children)
|
||||
self.assertIn("C", graph.package_nodes["B1"].children)
|
||||
|
||||
def test_cyclic_dep_fails(self):
|
||||
graph = DepGraph()
|
||||
|
||||
graph.add_requirement("A", None)
|
||||
graph.add_requirement("B", "A")
|
||||
graph.add_requirement("C", "B")
|
||||
|
||||
with self.assertRaises(UserFacingError):
|
||||
graph.add_requirement("A", "C")
|
||||
|
||||
def test_get_and_remove_outer_deps(self):
|
||||
graph = DepGraph()
|
||||
|
||||
graph.add_requirement("A", None)
|
||||
graph.add_requirement("V", None)
|
||||
|
||||
graph.add_requirement("B1", "A")
|
||||
graph.add_requirement("B2", "A")
|
||||
graph.add_requirement("B3", "A")
|
||||
|
||||
graph.add_requirement("B1", "B2")
|
||||
graph.add_requirement("C1", "B1")
|
||||
graph.add_requirement("C2", "B1")
|
||||
|
||||
graph.add_requirement("D", "C1")
|
||||
|
||||
graph.add_requirement("C2", "D")
|
||||
|
||||
v = ForeignPackage("V")
|
||||
|
||||
a = ForeignPackage("A")
|
||||
a.add_foreign_dependency_packages(["B1", "B2", "B3", "C1", "C2", "D"])
|
||||
|
||||
b1 = ForeignPackage("B1")
|
||||
b1.add_foreign_dependency_packages(["C1", "C2", "D"])
|
||||
|
||||
b2 = ForeignPackage("B2")
|
||||
b2.add_foreign_dependency_packages(["B1", "C1", "C2", "D"])
|
||||
|
||||
b3 = ForeignPackage("B3")
|
||||
|
||||
c1 = ForeignPackage("C1")
|
||||
c1.add_foreign_dependency_packages(["D", "C2"])
|
||||
|
||||
c2 = ForeignPackage("C2")
|
||||
|
||||
d = ForeignPackage("D")
|
||||
d.add_foreign_dependency_packages(["C2"])
|
||||
|
||||
self.assertCountEqual(graph.get_and_remove_outer_dep_pkgs(), [c2, b3, v])
|
||||
self.assertCountEqual(graph.get_and_remove_outer_dep_pkgs(), [d])
|
||||
self.assertCountEqual(graph.get_and_remove_outer_dep_pkgs(), [c1])
|
||||
self.assertCountEqual(graph.get_and_remove_outer_dep_pkgs(), [b1])
|
||||
self.assertCountEqual(graph.get_and_remove_outer_dep_pkgs(), [b2])
|
||||
self.assertCountEqual(graph.get_and_remove_outer_dep_pkgs(), [a])
|
||||
self.assertCountEqual(graph.get_and_remove_outer_dep_pkgs(), [])
|
||||
@@ -1,302 +0,0 @@
|
||||
# pylint: disable=missing-module-docstring,missing-class-docstring,missing-function-docstring
|
||||
|
||||
import unittest
|
||||
|
||||
from decman import Module, UserPackage
|
||||
from decman.lib import Source, Store
|
||||
|
||||
|
||||
class ExistingTestModule(Module):
|
||||
def __init__(self):
|
||||
self.on_enable_executed = False
|
||||
self.on_disable_executed = False
|
||||
self.after_update_executed = False
|
||||
self.after_version_change_executed = False
|
||||
super().__init__("Existing", True, "1")
|
||||
|
||||
def on_enable(self):
|
||||
self.on_enable_executed = True
|
||||
|
||||
def on_disable(self):
|
||||
self.on_disable_executed = True
|
||||
|
||||
def after_update(self):
|
||||
self.after_update_executed = True
|
||||
|
||||
def after_version_change(self):
|
||||
self.after_version_change_executed = True
|
||||
|
||||
|
||||
class ExistingChangedVersionTestModule(Module):
|
||||
def __init__(self):
|
||||
self.on_enable_executed = False
|
||||
self.on_disable_executed = False
|
||||
self.after_update_executed = False
|
||||
self.after_version_change_executed = False
|
||||
super().__init__("ExistingChanged", True, "2")
|
||||
|
||||
def on_enable(self):
|
||||
self.on_enable_executed = True
|
||||
|
||||
def on_disable(self):
|
||||
self.on_disable_executed = True
|
||||
|
||||
def after_update(self):
|
||||
self.after_update_executed = True
|
||||
|
||||
def after_version_change(self):
|
||||
self.after_version_change_executed = True
|
||||
|
||||
|
||||
class EnabledTestModule(Module):
|
||||
def __init__(self):
|
||||
self.on_enable_executed = False
|
||||
self.on_disable_executed = False
|
||||
self.after_update_executed = False
|
||||
self.after_version_change_executed = False
|
||||
super().__init__("Enabled", True, "1")
|
||||
|
||||
def on_enable(self):
|
||||
self.on_enable_executed = True
|
||||
|
||||
def on_disable(self):
|
||||
self.on_disable_executed = True
|
||||
|
||||
def after_update(self):
|
||||
self.after_update_executed = True
|
||||
|
||||
def after_version_change(self):
|
||||
self.after_version_change_executed = True
|
||||
|
||||
def pacman_packages(self) -> list[str]:
|
||||
return ["M_p1", "M_p2", "M_p3"]
|
||||
|
||||
def systemd_user_units(self) -> dict[str, list[str]]:
|
||||
return {"muser": ["M_u1.service"]}
|
||||
|
||||
def flatpak_packages(self) -> list[str]:
|
||||
return ["M_f1", "M_f2"]
|
||||
|
||||
|
||||
class DisabledTestModule(Module):
|
||||
def __init__(self):
|
||||
self.on_enable_executed = False
|
||||
self.on_disable_executed = False
|
||||
self.after_update_executed = False
|
||||
self.after_version_change_executed = False
|
||||
super().__init__("Disabled", False, "1")
|
||||
|
||||
def on_enable(self):
|
||||
self.on_enable_executed = True
|
||||
|
||||
def on_disable(self):
|
||||
self.on_disable_executed = True
|
||||
|
||||
def after_update(self):
|
||||
self.after_update_executed = True
|
||||
|
||||
def after_version_change(self):
|
||||
self.after_version_change_executed = True
|
||||
|
||||
def aur_packages(self) -> list[str]:
|
||||
return ["M_A1", "M_A2", "M_A3"]
|
||||
|
||||
def systemd_units(self) -> list[str]:
|
||||
return ["M_1.service"]
|
||||
|
||||
|
||||
class TestSource(unittest.TestCase):
|
||||
def setUp(self):
|
||||
self.disabled_module = DisabledTestModule()
|
||||
self.enabled_module = EnabledTestModule()
|
||||
self.existing_module = ExistingTestModule()
|
||||
self.existing_module_changed = ExistingChangedVersionTestModule()
|
||||
modules = {
|
||||
self.enabled_module,
|
||||
self.disabled_module,
|
||||
self.existing_module,
|
||||
self.existing_module_changed,
|
||||
}
|
||||
source = Source(
|
||||
pacman_packages={"p1", "p2", "p3"},
|
||||
aur_packages={"A1", "A2", "A3"},
|
||||
user_packages={
|
||||
UserPackage(
|
||||
pkgname="U1",
|
||||
version="1",
|
||||
dependencies=["d1"],
|
||||
git_url="/am/url/yes",
|
||||
),
|
||||
UserPackage(
|
||||
pkgname="U2",
|
||||
version="1",
|
||||
dependencies=["d2"],
|
||||
git_url="/am/url/yes",
|
||||
),
|
||||
},
|
||||
ignored_packages={"i1", "i2"},
|
||||
systemd_units={"1.service", "2.timer"},
|
||||
systemd_user_units={"user": {"u1.service", "u2.timer"}},
|
||||
modules=modules,
|
||||
files={},
|
||||
directories={},
|
||||
flatpak_packages={"f1", "f2", "f3"},
|
||||
flatpak_user_packages={"fu1", "fu2", "fu3"},
|
||||
ignored_flatpak_packages={"i1", "i2"},
|
||||
)
|
||||
|
||||
store = Store()
|
||||
store.enabled_systemd_units.extend(["1.service", "3.service", "M_1.service"])
|
||||
store.add_enabled_user_systemd_unit("user", "u1.service")
|
||||
store.add_enabled_user_systemd_unit("user", "u3.service")
|
||||
store.enabled_modules = {
|
||||
"Existing": "1",
|
||||
"ExistingChanged": "1",
|
||||
"Disabled": "1",
|
||||
}
|
||||
store.created_files = ["/test/file1", "/test/file2", "/test/file3"]
|
||||
|
||||
currently_installed_packages = [
|
||||
"p1",
|
||||
"p2",
|
||||
"p4",
|
||||
"A2",
|
||||
"A3",
|
||||
"A4",
|
||||
"U1",
|
||||
"i1",
|
||||
"M_p3",
|
||||
"M_A1",
|
||||
"M_A2",
|
||||
]
|
||||
|
||||
self.source = source
|
||||
self.store = store
|
||||
self.currently_installed_packages = currently_installed_packages
|
||||
|
||||
def test_all_enabled_modules(self):
|
||||
enabled_modules = [
|
||||
("Enabled", "1"),
|
||||
("Existing", "1"),
|
||||
("ExistingChanged", "2"),
|
||||
]
|
||||
self.assertCountEqual(self.source.all_enabled_modules(), enabled_modules)
|
||||
|
||||
def test_files_to_remove(self):
|
||||
created_files = ["/test/file1", "/test/file4"]
|
||||
self.assertCountEqual(
|
||||
self.source.files_to_remove(self.store, created_files),
|
||||
["/test/file2", "/test/file3"],
|
||||
)
|
||||
|
||||
def test_after_update_executed(self):
|
||||
self.source.run_after_update()
|
||||
|
||||
self.assertTrue(self.enabled_module.after_update_executed)
|
||||
self.assertTrue(self.existing_module.after_update_executed)
|
||||
self.assertTrue(self.existing_module_changed.after_update_executed)
|
||||
self.assertFalse(self.disabled_module.after_update_executed)
|
||||
|
||||
def test_after_version_change_executed(self):
|
||||
self.source.run_after_version_change(self.store)
|
||||
|
||||
self.assertTrue(self.enabled_module.after_version_change_executed)
|
||||
self.assertTrue(self.existing_module_changed.after_version_change_executed)
|
||||
self.assertFalse(self.existing_module.after_version_change_executed)
|
||||
self.assertFalse(self.disabled_module.after_version_change_executed)
|
||||
|
||||
def test_on_enable_executed(self):
|
||||
self.source.run_on_enable(self.store)
|
||||
|
||||
self.assertTrue(self.enabled_module.on_enable_executed)
|
||||
self.assertFalse(self.disabled_module.on_enable_executed)
|
||||
self.assertFalse(self.existing_module.on_enable_executed)
|
||||
self.assertFalse(self.existing_module_changed.on_enable_executed)
|
||||
|
||||
def test_on_disable_executed(self):
|
||||
self.source.run_on_disable(self.store)
|
||||
|
||||
self.assertTrue(self.disabled_module.on_disable_executed)
|
||||
self.assertFalse(self.enabled_module.on_disable_executed)
|
||||
self.assertFalse(self.existing_module.on_disable_executed)
|
||||
self.assertFalse(self.existing_module_changed.on_disable_executed)
|
||||
|
||||
def test_units_to_enable(self):
|
||||
self.assertCountEqual(
|
||||
self.source.units_to_enable(self.store),
|
||||
["2.timer"],
|
||||
)
|
||||
|
||||
def test_units_to_disable(self):
|
||||
self.assertCountEqual(
|
||||
self.source.units_to_disable(self.store),
|
||||
["3.service", "M_1.service"],
|
||||
)
|
||||
|
||||
def test_user_units_to_enable(self):
|
||||
self.assertDictEqual(
|
||||
self.source.user_units_to_enable(self.store),
|
||||
{"user": ["u2.timer"], "muser": ["M_u1.service"]},
|
||||
)
|
||||
|
||||
def test_user_units_to_disable(self):
|
||||
self.assertDictEqual(
|
||||
self.source.user_units_to_disable(self.store),
|
||||
{"user": ["u3.service"]},
|
||||
)
|
||||
|
||||
def test_pacman_packages_to_install(self):
|
||||
self.assertCountEqual(
|
||||
self.source.pacman_packages_to_install(self.currently_installed_packages),
|
||||
["p3", "M_p1", "M_p2"],
|
||||
)
|
||||
|
||||
def test_foreign_packages_to_install(self):
|
||||
self.assertCountEqual(
|
||||
self.source.foreign_packages_to_install(self.currently_installed_packages),
|
||||
["A1", "U2"],
|
||||
)
|
||||
|
||||
def test_packages_to_remove(self):
|
||||
self.assertCountEqual(
|
||||
self.source.packages_to_remove(self.currently_installed_packages),
|
||||
["p4", "A4", "M_A1", "M_A2"],
|
||||
)
|
||||
|
||||
|
||||
class TestModuleUserServices(unittest.TestCase):
|
||||
class ModuleWithUserServiceOne(Module):
|
||||
def __init__(self):
|
||||
super().__init__("one", True, "0")
|
||||
|
||||
def systemd_user_units(self) -> dict[str, list[str]]:
|
||||
return {"user": ["foo.service"]}
|
||||
|
||||
class ModuleWithUserServiceTwo(Module):
|
||||
def __init__(self):
|
||||
super().__init__("two", True, "0")
|
||||
|
||||
def systemd_user_units(self) -> dict[str, list[str]]:
|
||||
return {"user": ["bar.service"]}
|
||||
|
||||
def setUp(self) -> None:
|
||||
self.source = Source(
|
||||
pacman_packages=set(),
|
||||
aur_packages=set(),
|
||||
user_packages=set(),
|
||||
ignored_packages=set(),
|
||||
systemd_units=set(),
|
||||
systemd_user_units={},
|
||||
files={},
|
||||
directories={},
|
||||
modules={self.ModuleWithUserServiceOne(), self.ModuleWithUserServiceTwo()},
|
||||
flatpak_packages=set(),
|
||||
flatpak_user_packages=set(),
|
||||
ignored_flatpak_packages=set(),
|
||||
)
|
||||
self.store = Store()
|
||||
|
||||
def test_user_units_to_enable(self):
|
||||
result = self.source.user_units_to_enable(self.store)
|
||||
self.assertEqual(len(result), 1)
|
||||
self.assertCountEqual(result["user"], ["foo.service", "bar.service"])
|
||||
Reference in New Issue
Block a user