From 19fcec84d8d7d21e796c7624e521b60d28ee21ed Mon Sep 17 00:00:00 2001 From: Daniel Baumann Date: Sun, 7 Apr 2024 20:45:59 +0200 Subject: Adding upstream version 16.2.11+ds. Signed-off-by: Daniel Baumann --- src/ceph-volume/ceph_volume/tests/test_process.py | 92 +++++++++++++++++++++++ 1 file changed, 92 insertions(+) create mode 100644 src/ceph-volume/ceph_volume/tests/test_process.py (limited to 'src/ceph-volume/ceph_volume/tests/test_process.py') diff --git a/src/ceph-volume/ceph_volume/tests/test_process.py b/src/ceph-volume/ceph_volume/tests/test_process.py new file mode 100644 index 000000000..46e5c40e6 --- /dev/null +++ b/src/ceph-volume/ceph_volume/tests/test_process.py @@ -0,0 +1,92 @@ +import pytest +import logging +from ceph_volume.tests.conftest import Factory +from ceph_volume import process + + +@pytest.fixture +def mock_call(monkeypatch): + """ + Monkeypatches process.call, so that a caller can add behavior to the response + """ + def apply(stdout=None, stderr=None, returncode=0): + stdout_stream = Factory(read=lambda: stdout) + stderr_stream = Factory(read=lambda: stderr) + return_value = Factory( + stdout=stdout_stream, + stderr=stderr_stream, + wait=lambda: returncode, + communicate=lambda x: (stdout, stderr, returncode) + ) + + monkeypatch.setattr( + 'ceph_volume.process.subprocess.Popen', + lambda *a, **kw: return_value) + + return apply + + +class TestCall(object): + + def test_stderr_terminal_and_logfile(self, mock_call, caplog, capsys): + caplog.set_level(logging.INFO) + mock_call(stdout='stdout\n', stderr='some stderr message\n') + process.call(['ls'], terminal_verbose=True) + out, err = capsys.readouterr() + log_lines = [line[-1] for line in caplog.record_tuples] + assert 'Running command: ' in log_lines[0] + assert 'ls' in log_lines[0] + assert 'stderr some stderr message' in log_lines[-1] + assert 'some stderr message' in err + + def test_stderr_terminal_and_logfile_off(self, mock_call, caplog, capsys): + caplog.set_level(logging.INFO) + mock_call(stdout='stdout\n', stderr='some stderr message\n') + process.call(['ls'], terminal_verbose=False) + out, err = capsys.readouterr() + log_lines = [line[-1] for line in caplog.record_tuples] + assert 'Running command: ' in log_lines[0] + assert 'ls' in log_lines[0] + assert 'stderr some stderr message' in log_lines[-1] + assert out == '' + + def test_verbose_on_failure(self, mock_call, caplog, capsys): + caplog.set_level(logging.INFO) + mock_call(stdout='stdout\n', stderr='stderr\n', returncode=1) + process.call(['ls'], terminal_verbose=False, logfile_verbose=False) + out, err = capsys.readouterr() + log_lines = '\n'.join([line[-1] for line in caplog.record_tuples]) + assert 'Running command: ' in log_lines + assert 'ls' in log_lines + assert 'stderr' in log_lines + assert 'stdout: stdout' in err + assert out == '' + + def test_silent_verbose_on_failure(self, mock_call, caplog, capsys): + caplog.set_level(logging.INFO) + mock_call(stdout='stdout\n', stderr='stderr\n', returncode=1) + process.call(['ls'], verbose_on_failure=False) + out, err = capsys.readouterr() + log_lines = '\n'.join([line[-1] for line in caplog.record_tuples]) + assert 'Running command: ' in log_lines + assert 'ls' in log_lines + assert 'stderr' in log_lines + assert out == '' + + +class TestFunctionalCall(object): + + def test_stdin(self): + process.call(['xargs', 'ls'], stdin="echo '/'") + + def test_unicode_encoding(self): + process.call(['echo', u'\xd0']) + + def test_unicode_encoding_stdin(self): + process.call(['echo'], stdin=u'\xd0'.encode('utf-8')) + + +class TestFunctionalRun(object): + + def test_log_descriptors(self): + process.run(['ls', '-l']) -- cgit v1.2.3