1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
|
# Copyright 2016: Mirantis Inc.
# All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
import subprocess
from unittest import mock
import ddt
from rally import consts
from rally.plugins.task.hooks import sys_call
from rally.task import hook
from tests.unit import fakes
from tests.unit import test
@ddt.ddt
class SysCallHookTestCase(test.TestCase):
@ddt.data(("ls", True), (50, False))
@ddt.unpack
def test_validate(self, config, valid):
results = hook.HookAction.validate(
"sys_call", None, None, config)
if valid:
self.assertEqual([], results)
else:
self.assertEqual(1, len(results))
@ddt.data(
{"stdout": "foo output",
"expected": {
"additive": [],
"complete": [{"chart_plugin": "TextArea",
"data": ["RetCode: 0", "StdOut: foo output",
"StdErr: (empty)"],
"description": "Args: foo cmd",
"title": "System call"}]}},
{"stdout": """{"additive": [],
"complete": [
{"chart_plugin": "Pie", "title": "Bar Pie",
"data": [["A", 4], ["B", 2]]}]}""",
"expected": {
"additive": [],
"complete": [{"chart_plugin": "Pie", "data": [["A", 4], ["B", 2]],
"title": "Bar Pie"}]}})
@ddt.unpack
@mock.patch("rally.common.utils.Timer", side_effect=fakes.FakeTimer)
@mock.patch("rally.plugins.task.hooks.sys_call.subprocess.Popen")
def test_run(self, mock_popen, mock_timer, stdout, expected):
popen_instance = mock_popen.return_value
popen_instance.returncode = 0
popen_instance.communicate.return_value = (stdout, "")
hook = sys_call.SysCallHook(mock.Mock(), "foo cmd", {"iteration": 1})
hook.run_sync()
self.assertEqual(
{"finished_at": fakes.FakeTimer().finish_timestamp(),
"output": expected,
"started_at": fakes.FakeTimer().timestamp(),
"status": consts.HookStatus.SUCCESS,
"triggered_by": {"iteration": 1}},
hook.result())
mock_popen.assert_called_once_with(["foo", "cmd"],
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
universal_newlines=True)
@ddt.data(
{"communicate_streams": ("foo out", "foo err"),
"expected_error_details": "foo err",
"expected_data_stderr": "StdErr: foo err"},
{"communicate_streams": ("foo out", ""),
"expected_error_details": "stdout: foo out",
"expected_data_stderr": "StdErr: (empty)"})
@ddt.unpack
@mock.patch("rally.common.utils.Timer", side_effect=fakes.FakeTimer)
@mock.patch("rally.plugins.task.hooks.sys_call.subprocess.Popen")
def test_run_error(self, mock_popen, mock_timer, communicate_streams,
expected_error_details, expected_data_stderr):
popen_instance = mock_popen.return_value
popen_instance.communicate.return_value = communicate_streams
popen_instance.returncode = 1
task = mock.MagicMock()
sys_call_hook = sys_call.SysCallHook(task, "/bin/bash -c 'ls'",
{"iteration": 1})
sys_call_hook.run_sync()
self.assertEqual(
{"error": {"details": expected_error_details,
"etype": "n/a",
"msg": "Subprocess returned 1"},
"finished_at": fakes.FakeTimer().finish_timestamp(),
"output": {
"additive": [],
"complete": [{"chart_plugin": "TextArea",
"data": ["RetCode: 1",
"StdOut: foo out",
expected_data_stderr],
"description": "Args: /bin/bash -c 'ls'",
"title": "System call"}]},
"started_at": fakes.FakeTimer().timestamp(),
"status": "failed",
"triggered_by": {"iteration": 1}}, sys_call_hook.result())
mock_popen.assert_called_once_with(
["/bin/bash", "-c", "ls"],
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
universal_newlines=True)
|
