ceph/src/ceph-volume/ceph_volume/tests/test_process.py

   1 import pytest
   2 from ceph_volume.tests.conftest import Factory
   3 from ceph_volume import process
   4
   5
   6 @pytest.fixture
   7 def mock_call(monkeypatch):
   8     """
   9     Monkeypatches process.call, so that a caller can add behavior to the response
  10     """
  11     def apply(stdout=None, stderr=None, returncode=0):
  12         stdout_stream = Factory(read=lambda: stdout)
  13         stderr_stream = Factory(read=lambda: stderr)
  14         return_value = Factory(
  15             stdout=stdout_stream,
  16             stderr=stderr_stream,
  17             wait=lambda: returncode,
  18             communicate=lambda x: (stdout, stderr, returncode)
  19         )
  20
  21         monkeypatch.setattr(
  22             'ceph_volume.process.subprocess.Popen',
  23             lambda *a, **kw: return_value)
  24
  25     return apply
  26
  27
  28 class TestCall(object):
  29
  30     def test_stderr_terminal_and_logfile(self, mock_call, caplog, capsys):
  31         mock_call(stdout='stdout\n', stderr='some stderr message\n')
  32         process.call(['ls'], terminal_verbose=True)
  33         out, err = capsys.readouterr()
  34         log_lines = [line[-1] for line in caplog.record_tuples]
  35         assert 'Running command: ' in log_lines[0]
  36         assert 'ls' in log_lines[0]
  37         assert 'stderr some stderr message' in log_lines[-1]
  38         assert 'some stderr message' in out
  39
  40     def test_stderr_terminal_and_logfile_off(self, mock_call, caplog, capsys):
  41         mock_call(stdout='stdout\n', stderr='some stderr message\n')
  42         process.call(['ls'], terminal_verbose=False)
  43         out, err = capsys.readouterr()
  44         log_lines = [line[-1] for line in caplog.record_tuples]
  45         assert 'Running command: ' in log_lines[0]
  46         assert 'ls' in log_lines[0]
  47         assert 'stderr some stderr message' in log_lines[-1]
  48         assert out == ''
  49
  50     def test_verbose_on_failure(self, mock_call, caplog, capsys):
  51         mock_call(stdout='stdout\n', stderr='stderr\n', returncode=1)
  52         process.call(['ls'], terminal_verbose=False, logfile_verbose=False)
  53         out, err = capsys.readouterr()
  54         log_lines = '\n'.join([line[-1] for line in caplog.record_tuples])
  55         assert 'Running command: ' in log_lines
  56         assert 'ls' in log_lines
  57         assert 'stderr' in log_lines
  58         assert 'stdout: stdout' in out
  59
  60     def test_silent_verbose_on_failure(self, mock_call, caplog, capsys):
  61         mock_call(stdout='stdout\n', stderr='stderr\n', returncode=1)
  62         process.call(['ls'], verbose_on_failure=False)
  63         out, err = capsys.readouterr()
  64         log_lines = '\n'.join([line[-1] for line in caplog.record_tuples])
  65         assert 'Running command: ' in log_lines
  66         assert 'ls' in log_lines
  67         assert 'stderr' in log_lines
  68         assert out == ''
  69
  70
  71 class TestFunctionalCall(object):
  72
  73     def test_stdin(self):
  74         process.call(['xargs', 'ls'], stdin="echo '/'")
  75
  76     def test_unicode_encoding(self):
  77         process.call(['echo', u'\xd0'])
  78
  79     def test_unicode_encoding_stdin(self):
  80         process.call(['echo'], stdin=u'\xd0'.encode('utf-8'))
  81
  82
  83 class TestFunctionalRun(object):
  84
  85     def test_log_descriptors(self):
  86         process.run(['ls', '-l'])