kstyagi
d8a25209b9
- Introduced a new method `installed_kernel` in `ImageStore` to handle kernel verification and checksum validation. - Updated `Installer` to include additional packages: `nftables` and `conntrack`. - Refined `LifecycleService` to improve error handling and state management during VM creation, starting, stopping, and deletion. - Enhanced port management with new methods for setting and syncing ports, including validation for port rules. - Improved error handling in `StateStore` to raise specific `NotFoundError` and `ConflictError` exceptions. - Updated network management to ensure proper TAP device handling and firewall integration. - Enhanced API responses in `vms.py` to include policy status for better client-side handling. - Added validation for VM names and ports in `validation.py` to enforce stricter rules and prevent conflicts.
100 lines
4.3 KiB
Python
100 lines
4.3 KiB
Python
from __future__ import annotations
|
|
|
|
import io
|
|
import subprocess
|
|
import unittest
|
|
from types import SimpleNamespace
|
|
from unittest.mock import patch
|
|
|
|
from uvm.errors import CommandError
|
|
from uvm.system import CommandRunner
|
|
|
|
|
|
class CommandRunnerTests(unittest.TestCase):
|
|
def test_uncaptured_child_stdout_is_redirected_to_current_stderr(self) -> None:
|
|
stderr = io.StringIO()
|
|
completed = SimpleNamespace(returncode=0, stdout=None, stderr=None)
|
|
|
|
with (
|
|
patch("uvm.system.sys.stderr", stderr),
|
|
patch("uvm.system.subprocess.run", return_value=completed) as run,
|
|
):
|
|
result = CommandRunner(emit=None).run(("host-tool", "--check"), timeout=2.5)
|
|
|
|
run.assert_called_once_with(
|
|
("host-tool", "--check"), check=False, text=True, input=None,
|
|
stdout=stderr, stderr=None, timeout=2.5,
|
|
)
|
|
self.assertEqual(result.args, ("host-tool", "--check"))
|
|
self.assertEqual(result.returncode, 0)
|
|
self.assertEqual(result.stdout, "")
|
|
self.assertEqual(result.stderr, "")
|
|
|
|
def test_captured_stdout_and_stderr_still_use_separate_pipes(self) -> None:
|
|
completed = SimpleNamespace(returncode=0, stdout="result\n", stderr="diagnostic\n")
|
|
|
|
with patch("uvm.system.subprocess.run", return_value=completed) as run:
|
|
result = CommandRunner(emit=None).run(("host-tool",), capture=True)
|
|
|
|
self.assertEqual(run.call_args.kwargs["stdout"], subprocess.PIPE)
|
|
self.assertEqual(run.call_args.kwargs["stderr"], subprocess.PIPE)
|
|
self.assertEqual(result.stdout, "result\n")
|
|
self.assertEqual(result.stderr, "diagnostic\n")
|
|
|
|
def test_passes_sensitive_input_over_stdin_without_logging_it(self) -> None:
|
|
emitted: list[str] = []
|
|
completed = SimpleNamespace(
|
|
returncode=0, stdout="result secret-value\n", stderr="warning secret-value\n",
|
|
)
|
|
|
|
with patch("uvm.system.subprocess.run", return_value=completed) as run:
|
|
result = CommandRunner(emit=emitted.append).run(
|
|
("credential-tool", "--stdin"),
|
|
capture=True,
|
|
input_text="secret-value\n",
|
|
sensitive=True,
|
|
)
|
|
|
|
self.assertEqual(result.stdout, "result secret-value\n")
|
|
self.assertEqual(result.stderr, "warning secret-value\n")
|
|
self.assertEqual(run.call_args.kwargs["stdout"], subprocess.PIPE)
|
|
self.assertEqual(run.call_args.kwargs["stderr"], subprocess.PIPE)
|
|
self.assertEqual(run.call_args.kwargs["input"], "secret-value\n")
|
|
self.assertEqual(emitted, ["+ credential-tool --stdin"])
|
|
|
|
def test_redacts_sensitive_command_output_from_errors(self) -> None:
|
|
for stdout, stderr in (
|
|
("failure mentioning secret-value", ""),
|
|
("", "failure mentioning secret-value"),
|
|
("secret-value", "failure mentioning secret-value"),
|
|
):
|
|
with self.subTest(stdout=stdout, stderr=stderr):
|
|
completed = SimpleNamespace(returncode=1, stdout=stdout, stderr=stderr)
|
|
emitted: list[str] = []
|
|
with (
|
|
patch("uvm.system.subprocess.run", return_value=completed),
|
|
self.assertRaises(CommandError) as raised,
|
|
):
|
|
CommandRunner(emit=emitted.append).run(
|
|
("credential-tool", "--stdin"), capture=True,
|
|
input_text="secret-value\n", sensitive=True,
|
|
)
|
|
|
|
self.assertNotIn("secret-value", str(raised.exception))
|
|
self.assertEqual(emitted, ["+ credential-tool --stdin"])
|
|
|
|
def test_unchecked_sensitive_failure_preserves_captured_output_for_the_caller(self) -> None:
|
|
completed = SimpleNamespace(
|
|
returncode=1, stdout="secret-value\n", stderr="failure mentioning secret-value",
|
|
)
|
|
|
|
with patch("uvm.system.subprocess.run", return_value=completed):
|
|
result = CommandRunner(emit=None).run(
|
|
("credential-tool", "--stdin"), check=False, capture=True,
|
|
input_text="secret-value\n", sensitive=True,
|
|
)
|
|
|
|
self.assertEqual(result.returncode, 1)
|
|
self.assertEqual(result.stdout, completed.stdout)
|
|
self.assertEqual(result.stderr, completed.stderr)
|