mirror of
https://github.com/kiviktnm/decman.git
synced 2026-09-19 12:08:28 +00:00
Improve command output printing
This commit is contained in:
@@ -1,9 +1,111 @@
|
||||
import json
|
||||
import sys
|
||||
import typing
|
||||
|
||||
import pytest
|
||||
|
||||
import decman.core.command as command
|
||||
import decman.core.output
|
||||
|
||||
|
||||
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, pass_environment=None, mimic_login=False):
|
||||
calls["pty_run"] = (cmd, user, env_overrides, mimic_login)
|
||||
return 0, "ok"
|
||||
|
||||
def fake_check_run_result(cmd, result, include_output=None):
|
||||
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(command, "pty_run", fake_pty_run)
|
||||
monkeypatch.setattr(command, "check_run_result", fake_check_run_result)
|
||||
monkeypatch.setattr(decman.core.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, pass_environment=None, mimic_login=False):
|
||||
calls["run"] = (cmd, user, env_overrides, mimic_login)
|
||||
return 0, "no-pty"
|
||||
|
||||
def fake_check_run_result(cmd, result, include_output=None):
|
||||
return result
|
||||
|
||||
def fake_print_warning(msg: str):
|
||||
raise AssertionError("print_warning must not be called when code == 0")
|
||||
|
||||
monkeypatch.setattr(command, "run", fake_run)
|
||||
monkeypatch.setattr(command, "check_run_result", fake_check_run_result)
|
||||
monkeypatch.setattr(decman.core.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, pass_environment=None, mimic_login=False):
|
||||
# non-zero exit code
|
||||
return 3, "bad"
|
||||
|
||||
def fake_check_run_result(cmd, result, include_output=None):
|
||||
raise AssertionError("check_run_result must not be called when check=False")
|
||||
|
||||
def fake_print_warning(msg: str):
|
||||
calls["warning"] = msg
|
||||
|
||||
monkeypatch.setattr(command, "run", fake_run)
|
||||
monkeypatch.setattr(command, "check_run_result", fake_check_run_result)
|
||||
monkeypatch.setattr(decman.core.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, pass_environment=None, mimic_login=False):
|
||||
return 42, "boom"
|
||||
|
||||
def fake_check_run_result(cmd, result, include_output=None):
|
||||
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(command, "run", fake_run)
|
||||
monkeypatch.setattr(command, "check_run_result", fake_check_run_result)
|
||||
monkeypatch.setattr(decman.core.output, "print_warning", fake_print_warning)
|
||||
|
||||
with pytest.raises(CommandFailedError):
|
||||
decman.prg(["boom"], pty=False, check=True)
|
||||
|
||||
|
||||
def test_run_simple():
|
||||
|
||||
@@ -5,108 +5,6 @@ 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] = {}
|
||||
|
||||
|
||||
@@ -201,7 +201,7 @@ def test_apply_returns_false_on_command_failure(monkeypatch: pytest.MonkeyPatch)
|
||||
ok = pacman.apply(store, dry_run=False)
|
||||
|
||||
assert ok is False
|
||||
assert any("pacman command failed" in msg for msg in errors_logged)
|
||||
assert any("Pacman command exited with an unexpected" in msg for msg in errors_logged)
|
||||
assert any("boom" in msg for msg in continuations)
|
||||
assert traceback_called # at least once
|
||||
|
||||
|
||||
@@ -208,7 +208,7 @@ def test_apply_dry_run_does_not_mutate_store_or_call_commands(store):
|
||||
def test_enable_units_success(monkeypatch, store, systemd):
|
||||
store["systemd_units"] = {"old.service"}
|
||||
|
||||
def fake_run(cmd):
|
||||
def fake_run(cmd, **kwargs):
|
||||
assert cmd[0] == "systemctl"
|
||||
assert cmd[1] == "enable"
|
||||
assert "new.service" in cmd[2:]
|
||||
@@ -223,7 +223,7 @@ def test_enable_units_success(monkeypatch, store, systemd):
|
||||
def test_enable_units_failure_does_not_update_store(monkeypatch, store, systemd):
|
||||
store["systemd_units"] = {"old.service"}
|
||||
|
||||
def fake_run(cmd):
|
||||
def fake_run(cmd, **kwargs):
|
||||
return 1, "error"
|
||||
|
||||
monkeypatch.setattr(systemd_mod.command, "run", fake_run)
|
||||
@@ -237,7 +237,7 @@ def test_enable_units_failure_does_not_update_store(monkeypatch, store, systemd)
|
||||
def test_disable_units_success(monkeypatch, store, systemd):
|
||||
store["systemd_units"] = {"old.service", "new.service"}
|
||||
|
||||
def fake_run(cmd):
|
||||
def fake_run(cmd, **kwargs):
|
||||
assert cmd[0] == "systemctl"
|
||||
assert cmd[1] == "disable"
|
||||
assert "new.service" in cmd[2:]
|
||||
@@ -252,7 +252,7 @@ def test_disable_units_success(monkeypatch, store, systemd):
|
||||
def test_disable_units_failure_does_not_update_store(monkeypatch, store, systemd):
|
||||
store["systemd_units"] = {"old.service", "new.service"}
|
||||
|
||||
def fake_run(cmd):
|
||||
def fake_run(cmd, **kwargs):
|
||||
return 1, "error"
|
||||
|
||||
monkeypatch.setattr(systemd_mod.command, "run", fake_run)
|
||||
@@ -265,7 +265,7 @@ def test_disable_units_failure_does_not_update_store(monkeypatch, store, systemd
|
||||
def test_enable_user_units_success(monkeypatch, store, systemd):
|
||||
store["systemd_user_units"] = {"alice": {"olduser.service"}}
|
||||
|
||||
def fake_run(cmd):
|
||||
def fake_run(cmd, **kwargs):
|
||||
assert cmd[0] == "systemctl"
|
||||
assert "--user" in cmd
|
||||
assert "enable" in cmd
|
||||
@@ -284,7 +284,7 @@ def test_enable_user_units_success(monkeypatch, store, systemd):
|
||||
def test_enable_user_units_failure_does_not_update_store(monkeypatch, store, systemd):
|
||||
store["systemd_user_units"] = {"alice": {"olduser.service"}}
|
||||
|
||||
def fake_run(cmd):
|
||||
def fake_run(cmd, **kwargs):
|
||||
return 1, "error"
|
||||
|
||||
monkeypatch.setattr(systemd_mod.command, "run", fake_run)
|
||||
@@ -297,7 +297,7 @@ def test_enable_user_units_failure_does_not_update_store(monkeypatch, store, sys
|
||||
def test_disable_user_units_success(monkeypatch, store, systemd):
|
||||
store["systemd_user_units"] = {"alice": {"olduser.service", "newuser.service"}}
|
||||
|
||||
def fake_run(cmd):
|
||||
def fake_run(cmd, **kwargs):
|
||||
assert cmd[0] == "systemctl"
|
||||
assert "--user" in cmd
|
||||
assert "disable" in cmd
|
||||
@@ -313,7 +313,7 @@ def test_disable_user_units_success(monkeypatch, store, systemd):
|
||||
def test_disable_user_units_failure_does_not_update_store(monkeypatch, store, systemd):
|
||||
store["systemd_user_units"] = {"alice": {"olduser.service", "newuser.service"}}
|
||||
|
||||
def fake_run(cmd):
|
||||
def fake_run(cmd, **kwargs):
|
||||
return 1, "error"
|
||||
|
||||
monkeypatch.setattr(systemd_mod.command, "run", fake_run)
|
||||
@@ -330,7 +330,7 @@ def test_disable_user_units_failure_does_not_update_store(monkeypatch, store, sy
|
||||
def test_reload_daemon_uses_command_run(monkeypatch, systemd):
|
||||
called = {}
|
||||
|
||||
def fake_run(cmd):
|
||||
def fake_run(cmd, **kwargs):
|
||||
called["cmd"] = cmd
|
||||
return 0, "ok"
|
||||
|
||||
@@ -342,7 +342,7 @@ def test_reload_daemon_uses_command_run(monkeypatch, systemd):
|
||||
def test_reload_user_daemon_uses_command_run(monkeypatch, systemd):
|
||||
called = {}
|
||||
|
||||
def fake_run(cmd):
|
||||
def fake_run(cmd, **kwargs):
|
||||
called["cmd"] = cmd
|
||||
return 0, "ok"
|
||||
|
||||
|
||||
Reference in New Issue
Block a user