Issues in test_windows_runners.py - All Issues - Inspection of "Merge pull request #4321 from StackStorm/circleci-..." - StackStorm/st2 - Measure and Improve Code Quality continuously with Scrutinizer

Test Failed

Push — master ( e380d0...f5671d )

by W

created 2018-08-30 16:25 UTC

windows_runner/tests/unit/test_windows_runners.py (2 issues)

Labels

Severity

James Sigurðarson 2

# Licensed to the StackStorm, Inc ('StackStorm') under one or more
# contributor license agreements.  See the NOTICE file distributed with
# this work for additional information regarding copyright ownership.
# The ASF licenses this file to You 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.

from __future__ import absolute_import
import os
from unittest2 import TestCase

import mock
from six.moves import zip

from st2common.constants.action import LIVEACTION_STATUS_SUCCEEDED
from st2common.constants.action import LIVEACTION_STATUS_FAILED
from st2common.constants.action import LIVEACTION_STATUS_TIMED_OUT

from windows_runner.windows_command_runner import BaseWindowsRunner
from windows_runner.windows_script_runner import WindowsScriptRunner

BASE_DIR = os.path.dirname(os.path.abspath(__file__))
FIXTURES_DIR = os.path.abspath(os.path.join(BASE_DIR, '../fixtures/windows'))


class WindowsRunnerTestCase(TestCase):
    def test_get_winexe_command_args(self):
        arguments = [
            {
                'host': 'localhost',
                'username': 'Administrator1',
                'password': 'bar1',
                'command': 'powershell.exe "C:\\\\myscript.ps1"'
            },
            {
                'host': '127.0.0.1',
                'username': 'Administrator2',
                'password': 'bar2',
                'command': 'dir'
            },
            {
                'host': 'localhost',
                'username': 'Administrator3',
                'password': 'bar3',
                'command': 'dir',
                'domain': 'MyDomain'
            }
        ]
        expected_values = [
            [
                'winexe',
                '--interactive', '0',
                '-U', 'Administrator1%bar1',
                '//localhost',
                'powershell.exe "C:\\\\myscript.ps1"'
            ],
            [
                'winexe',
                '--interactive', '0',
                '-U', 'Administrator2%bar2',
                '//127.0.0.1',
                'dir'
            ],
            [
                'winexe',
                '--interactive', '0',
                '-U', 'MyDomain\Administrator3%bar3',

                '//localhost',
                'dir'
            ]
        ]

        runner = self._get_base_runner()
        for arguments, expected_value in zip(arguments, expected_values):
            actual_value = runner._get_winexe_command_args(**arguments)
            self.assertEqual(actual_value, expected_value)

    def test_get_smbclient_command_args(self):
        arguments = [
            {
                'host': 'localhost',
                'username': 'Administrator1',
                'password': 'bar1',
                'command': 'put /home/1.txt 1.txt',
                'share': 'C$'
            },
            {
                'host': 'localhost',
                'username': 'Administrator2',
                'password': 'bar2',
                'command': 'put /home/2.txt 2.txt',
                'share': 'D$'
            },
            {
                'host': 'localhost',
                'username': 'Administrator3',
                'password': 'bar3',
                'command': 'dir',
                'share': 'E$',
                'domain': 'MyDomain'
            }
        ]
        expected_values = [
            [
                'smbclient',
                '-U', 'Administrator1%bar1',
                '//localhost/C$',
                '-c', 'put /home/1.txt 1.txt'
            ],
            [
                'smbclient',
                '-U', 'Administrator2%bar2',
                '//localhost/D$',
                '-c', 'put /home/2.txt 2.txt'
            ],
            [
                'smbclient',
                '-U', 'MyDomain\Administrator3%bar3',
                '//localhost/E$',
                '-c', 'dir'
            ],
        ]

        runner = self._get_base_runner()
        for arguments, expected_value in zip(arguments, expected_values):
            actual_value = runner._get_smbclient_command_args(**arguments)
            self.assertEqual(actual_value, expected_value)

    def test_get_script_args(self):
        arguments = [
            {
                'positional_args': 'a b c',
                'named_args': {
                    'arg1': 'value1',
                    'arg2': 'value2'
                }
            },
            {
                'positional_args': 'a b c',
                'named_args': {
                    'arg1': 'value1',
                    'arg2': True,
                    'arg3': False,
                    'arg4': ['foo', 'bar', 'baz']
                }
            }
        ]
        expected_values = [
            'a b c -arg1 value1 -arg2 value2',
            'a b c -arg1 value1 -arg2 -arg3:$false -arg4 foo,bar,baz'
        ]

        runner = self._get_mock_script_runner()
        for arguments, expected_value in zip(arguments, expected_values):
            actual_value = runner._get_script_arguments(**arguments)
class MyParent:
    def __init__(self):
        self._x = 1;
        self.y = 2;

class MyChild(MyParent):
    def some_method(self):
        return self._x    # Ok, since accessed from a child class

class AnotherClass:
    def some_method(self, instance_of_my_child):
        return instance_of_my_child._x   # Would be flagged as AnotherClass is not
                                         # a child class of MyParent
            self.assertEqual(actual_value, expected_value)

    def test_parse_share_information(self):
        runner = self._get_mock_script_runner()

        fixture_path = os.path.join(FIXTURES_DIR, 'net_share_C_stdout.txt')
        with open(fixture_path, 'r') as fp:
            stdout = fp.read()

        result = runner._parse_share_information(stdout=stdout)

        expected_keys = ['share_name', 'path', 'remark', 'maximum_users', 'users', 'caching',
                         'permission']
        for key in expected_keys:
            self.assertTrue(key in result)

        self.assertEqual(result['share_name'], 'C$')
        self.assertEqual(result['path'], 'C:\\')
        self.assertEqual(result['users'], None)

    @mock.patch('windows_runner.windows_script_runner.run_command')
    def test_get_share_absolute_path(self, mock_run_command):
        runner = self._get_mock_script_runner()

        fixture_path = os.path.join(FIXTURES_DIR, 'net_share_C_stdout.txt')
        with open(fixture_path, 'r') as fp:
            stdout = fp.read()

        # Failure, non-zero status code
        mock_run_command.return_value = (2, '', '', False)
        self.assertRaises(Exception, runner._get_share_absolute_path, share='C$')

        # Failure, missing / corrupted data
        mock_run_command.return_value = (0, '', '', False)
        self.assertRaises(Exception, runner._get_share_absolute_path, share='C$')

        # Success, everything OK
        mock_run_command.return_value = (0, stdout, '', False)
        share_path = runner._get_share_absolute_path(share='C$')
        self.assertEqual(share_path, 'C:\\')

    def test_run_output_object_and_status(self):
        runner = self._get_mock_script_runner()

        runner._upload_file = mock.Mock(return_value=('/tmp/a', '/tmp/b'))
        runner._delete_directory = mock.Mock()
        runner._get_share_absolute_path = mock.Mock(return_value='/tmp')
        runner._parse_winexe_error = mock.Mock(return_value='')
        runner._verify_winexe_exists = mock.Mock()
        runner._verify_smbclient_exists = mock.Mock()

        # success
        exit_code, stdout, stderr, timed_out = 0, 'stdout foo', 'stderr bar', False
        runner._run_script = mock.Mock(return_value=(exit_code, stdout, stderr, timed_out))

        runner.runner_parameters = {}
        (status, output, _) = runner.run({})

        expected_output = {
            'stdout': 'stdout foo',
            'stderr': 'stderr bar',
            'return_code': 0,
            'succeeded': True,
            'failed': False
        }

        self.assertEqual(status, LIVEACTION_STATUS_SUCCEEDED)
        self.assertDictEqual(output, expected_output)

        # failure
        exit_code, stdout, stderr, timed_out = 1, 'stdout fail', 'stderr fail', False
        runner._run_script = mock.Mock(return_value=(exit_code, stdout, stderr, timed_out))

        runner.runner_parameters = {}
        (status, output, _) = runner.run({})

        expected_output = {
            'stdout': 'stdout fail',
            'stderr': 'stderr fail',
            'return_code': 1,
            'succeeded': False,
            'failed': True
        }

        self.assertEqual(status, LIVEACTION_STATUS_FAILED)
        self.assertDictEqual(output, expected_output)

        # failure with winexe error
        exit_code, stdout, stderr, timed_out = 1, 'stdout fail 2', 'stderr fail 2', False
        runner._run_script = mock.Mock(return_value=(exit_code, stdout, stderr, timed_out))
        runner._parse_winexe_error = mock.Mock(return_value='winexe error 2')

        runner.runner_parameters = {}
        (status, output, _) = runner.run({})

        expected_output = {
            'stdout': 'stdout fail 2',
            'stderr': 'stderr fail 2',
            'return_code': 1,
            'succeeded': False,
            'failed': True,
            'error': 'winexe error 2'
        }

        # timeout with non zero exit code
        exit_code, stdout, stderr, timed_out = 200, 'stdout timeout', 'stderr timeout', True
        runner._run_script = mock.Mock(return_value=(exit_code, stdout, stderr, timed_out))
        runner._parse_winexe_error = mock.Mock(return_value=None)
        runner._timeout = 5

        runner.runner_parameters = {}
        (status, output, _) = runner.run({})

        expected_output = {
            'stdout': 'stdout timeout',
            'stderr': 'stderr timeout',
            'return_code': 200,
            'succeeded': False,
            'failed': True,
            'error': 'Action failed to complete in 5 seconds'
        }

        self.assertEqual(status, LIVEACTION_STATUS_TIMED_OUT)
        self.assertDictEqual(output, expected_output)

        # timeout with zero exit code
        exit_code, stdout, stderr, timed_out = 0, 'stdout timeout', 'stderr timeout', True
        runner._run_script = mock.Mock(return_value=(exit_code, stdout, stderr, timed_out))
        runner._timeout = 5
        runner._parse_winexe_error = mock.Mock(return_value='winexe error')

        runner.runner_parameters = {}
        (status, output, _) = runner.run({})

        expected_output = {
            'stdout': 'stdout timeout',
            'stderr': 'stderr timeout',
            'return_code': 0,
            'succeeded': False,
            'failed': True,
            'error': 'Action failed to complete in 5 seconds: winexe error'
        }

        self.assertEqual(status, LIVEACTION_STATUS_TIMED_OUT)
        self.assertDictEqual(output, expected_output)

    def test_shell_command_parameter_escaping(self):
        pass

    def _get_base_runner(self):
        class Runner(BaseWindowsRunner):
            def pre_run(self):
                pass

            def run(self):
                pass

        runner = Runner('id')
        return runner

    def _get_mock_script_runner(self, action_parameters=None):
        runner = WindowsScriptRunner('id')
        runner._host = None
        runner._username = None
        runner._password = None
        runner._timeout = None
        runner._share = None

        action_db = mock.Mock()
        action_db.pack = 'dummy_pack_1'
        action_db.entry_point = 'foo.py'
        action_db.parameters = action_parameters or {}

        runner.action = action_db

        return runner


1			# Licensed to the StackStorm, Inc ('StackStorm') under one or more
2			# contributor license agreements. See the NOTICE file distributed with
3			# this work for additional information regarding copyright ownership.
4			# The ASF licenses this file to You under the Apache License, Version 2.0
5			# (the "License"); you may not use this file except in compliance with
6			# the License. You may obtain a copy of the License at
7			#
8			# http://www.apache.org/licenses/LICENSE-2.0
9			#
10			# Unless required by applicable law or agreed to in writing, software
11			# distributed under the License is distributed on an "AS IS" BASIS,
12			# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13			# See the License for the specific language governing permissions and
14			# limitations under the License.
15
16			from __future__ import absolute_import
17			import os
18			from unittest2 import TestCase
19
20			import mock
21			from six.moves import zip
22
23			from st2common.constants.action import LIVEACTION_STATUS_SUCCEEDED
24			from st2common.constants.action import LIVEACTION_STATUS_FAILED
25			from st2common.constants.action import LIVEACTION_STATUS_TIMED_OUT
26
27			from windows_runner.windows_command_runner import BaseWindowsRunner
28			from windows_runner.windows_script_runner import WindowsScriptRunner
29
30			BASE_DIR = os.path.dirname(os.path.abspath(__file__))
31			FIXTURES_DIR = os.path.abspath(os.path.join(BASE_DIR, '../fixtures/windows'))
32
33
34			class WindowsRunnerTestCase(TestCase):
35			def test_get_winexe_command_args(self):
36			arguments = [
37			{
38			'host': 'localhost',
39			'username': 'Administrator1',
40			'password': 'bar1',
41			'command': 'powershell.exe "C:\\\\myscript.ps1"'
42			},
43			{
44			'host': '127.0.0.1',
45			'username': 'Administrator2',
46			'password': 'bar2',
47			'command': 'dir'
48			},
49			{
50			'host': 'localhost',
51			'username': 'Administrator3',
52			'password': 'bar3',
53			'command': 'dir',
54			'domain': 'MyDomain'
55			}
56			]
57			expected_values = [
58			[
59			'winexe',
60			'--interactive', '0',
61			'-U', 'Administrator1%bar1',
62			'//localhost',
63			'powershell.exe "C:\\\\myscript.ps1"'
64			],
65			[
66			'winexe',
67			'--interactive', '0',
68			'-U', 'Administrator2%bar2',
69			'//127.0.0.1',
70			'dir'
71			],
72			[
73			'winexe',
74			'--interactive', '0',
75			'-U', 'MyDomain\Administrator3%bar3',
			0 ignored issues – show Bug introduced 2016-11-23 23:57 UTC by Report Bug Copy Issue Report Show Similar Issues like this A suspicious escape sequence `\A` was found. Did you maybe forget to add an `r` prefix? Escape sequences in Python are generally interpreted according to rules similar to standard C. Only if strings are prefixed with `r` or `R` are they interpreted as regular expressions. The escape sequence that was used indicates that you might have intended to write a regular expression. Learn more about the available escape sequences. in the Python documentation. Loading history...
76			'//localhost',
77			'dir'
78			]
79			]
80
81			runner = self._get_base_runner()
82			for arguments, expected_value in zip(arguments, expected_values):
83			actual_value = runner._get_winexe_command_args(**arguments)
84			self.assertEqual(actual_value, expected_value)
85
86			def test_get_smbclient_command_args(self):
87			arguments = [
88			{
89			'host': 'localhost',
90			'username': 'Administrator1',
91			'password': 'bar1',
92			'command': 'put /home/1.txt 1.txt',
93			'share': 'C$'
94			},
95			{
96			'host': 'localhost',
97			'username': 'Administrator2',
98			'password': 'bar2',
99			'command': 'put /home/2.txt 2.txt',
100			'share': 'D$'
101			},
102			{
103			'host': 'localhost',
104			'username': 'Administrator3',
105			'password': 'bar3',
106			'command': 'dir',
107			'share': 'E$',
108			'domain': 'MyDomain'
109			}
110			]
111			expected_values = [
112			[
113			'smbclient',
114			'-U', 'Administrator1%bar1',
115			'//localhost/C$',
116			'-c', 'put /home/1.txt 1.txt'
117			],
118			[
119			'smbclient',
120			'-U', 'Administrator2%bar2',
121			'//localhost/D$',
122			'-c', 'put /home/2.txt 2.txt'
123			],
124			[
125			'smbclient',
126			'-U', 'MyDomain\Administrator3%bar3',
127			'//localhost/E$',
128			'-c', 'dir'
129			],
130			]
131
132			runner = self._get_base_runner()
133			for arguments, expected_value in zip(arguments, expected_values):
134			actual_value = runner._get_smbclient_command_args(**arguments)
135			self.assertEqual(actual_value, expected_value)
136
137			def test_get_script_args(self):
138			arguments = [
139			{
140			'positional_args': 'a b c',
141			'named_args': {
142			'arg1': 'value1',
143			'arg2': 'value2'
144			}
145			},
146			{
147			'positional_args': 'a b c',
148			'named_args': {
149			'arg1': 'value1',
150			'arg2': True,
151			'arg3': False,
152			'arg4': ['foo', 'bar', 'baz']
153			}
154			}
155			]
156			expected_values = [
157			'a b c -arg1 value1 -arg2 value2',
158			'a b c -arg1 value1 -arg2 -arg3:$false -arg4 foo,bar,baz'
159			]
160
161			runner = self._get_mock_script_runner()
162			for arguments, expected_value in zip(arguments, expected_values):
163			actual_value = runner._get_script_arguments(**arguments)
			0 ignored issues – show Coding Style Best Practice introduced 2017-04-18 08:10 UTC by Report Bug Copy Issue Report Show Similar Issues like this It seems like `_get_script_arguments` was declared protected and should not be accessed from this context. Prefixing a member variable `_` is usually regarded as the equivalent of declaring it with protected visibility that exists in other languages. Consequentially, such a member should only be accessed from the same class or a child class: class MyParent: def __init__(self): self._x = 1; self.y = 2; class MyChild(MyParent): def some_method(self): return self._x # Ok, since accessed from a child class class AnotherClass: def some_method(self, instance_of_my_child): return instance_of_my_child._x # Would be flagged as AnotherClass is not # a child class of MyParent Loading history...
164			self.assertEqual(actual_value, expected_value)
165
166			def test_parse_share_information(self):
167			runner = self._get_mock_script_runner()
168
169			fixture_path = os.path.join(FIXTURES_DIR, 'net_share_C_stdout.txt')
170			with open(fixture_path, 'r') as fp:
171			stdout = fp.read()
172
173			result = runner._parse_share_information(stdout=stdout)
174
175			expected_keys = ['share_name', 'path', 'remark', 'maximum_users', 'users', 'caching',
176			'permission']
177			for key in expected_keys:
178			self.assertTrue(key in result)
179
180			self.assertEqual(result['share_name'], 'C$')
181			self.assertEqual(result['path'], 'C:\\')
182			self.assertEqual(result['users'], None)
183
184			@mock.patch('windows_runner.windows_script_runner.run_command')
185			def test_get_share_absolute_path(self, mock_run_command):
186			runner = self._get_mock_script_runner()
187
188			fixture_path = os.path.join(FIXTURES_DIR, 'net_share_C_stdout.txt')
189			with open(fixture_path, 'r') as fp:
190			stdout = fp.read()
191
192			# Failure, non-zero status code
193			mock_run_command.return_value = (2, '', '', False)
194			self.assertRaises(Exception, runner._get_share_absolute_path, share='C$')
195
196			# Failure, missing / corrupted data
197			mock_run_command.return_value = (0, '', '', False)
198			self.assertRaises(Exception, runner._get_share_absolute_path, share='C$')
199
200			# Success, everything OK
201			mock_run_command.return_value = (0, stdout, '', False)
202			share_path = runner._get_share_absolute_path(share='C$')
203			self.assertEqual(share_path, 'C:\\')
204
205			def test_run_output_object_and_status(self):
206			runner = self._get_mock_script_runner()
207
208			runner._upload_file = mock.Mock(return_value=('/tmp/a', '/tmp/b'))
209			runner._delete_directory = mock.Mock()
210			runner._get_share_absolute_path = mock.Mock(return_value='/tmp')
211			runner._parse_winexe_error = mock.Mock(return_value='')
212			runner._verify_winexe_exists = mock.Mock()
213			runner._verify_smbclient_exists = mock.Mock()
214
215			# success
216			exit_code, stdout, stderr, timed_out = 0, 'stdout foo', 'stderr bar', False
217			runner._run_script = mock.Mock(return_value=(exit_code, stdout, stderr, timed_out))
218
219			runner.runner_parameters = {}
220			(status, output, _) = runner.run({})
221
222			expected_output = {
223			'stdout': 'stdout foo',
224			'stderr': 'stderr bar',
225			'return_code': 0,
226			'succeeded': True,
227			'failed': False
228			}
229
230			self.assertEqual(status, LIVEACTION_STATUS_SUCCEEDED)
231			self.assertDictEqual(output, expected_output)
232
233			# failure
234			exit_code, stdout, stderr, timed_out = 1, 'stdout fail', 'stderr fail', False
235			runner._run_script = mock.Mock(return_value=(exit_code, stdout, stderr, timed_out))
236
237			runner.runner_parameters = {}
238			(status, output, _) = runner.run({})
239
240			expected_output = {
241			'stdout': 'stdout fail',
242			'stderr': 'stderr fail',
243			'return_code': 1,
244			'succeeded': False,
245			'failed': True
246			}
247
248			self.assertEqual(status, LIVEACTION_STATUS_FAILED)
249			self.assertDictEqual(output, expected_output)
250
251			# failure with winexe error
252			exit_code, stdout, stderr, timed_out = 1, 'stdout fail 2', 'stderr fail 2', False
253			runner._run_script = mock.Mock(return_value=(exit_code, stdout, stderr, timed_out))
254			runner._parse_winexe_error = mock.Mock(return_value='winexe error 2')
255
256			runner.runner_parameters = {}
257			(status, output, _) = runner.run({})
258
259			expected_output = {
260			'stdout': 'stdout fail 2',
261			'stderr': 'stderr fail 2',
262			'return_code': 1,
263			'succeeded': False,
264			'failed': True,
265			'error': 'winexe error 2'
266			}
267
268			# timeout with non zero exit code
269			exit_code, stdout, stderr, timed_out = 200, 'stdout timeout', 'stderr timeout', True
270			runner._run_script = mock.Mock(return_value=(exit_code, stdout, stderr, timed_out))
271			runner._parse_winexe_error = mock.Mock(return_value=None)
272			runner._timeout = 5
273
274			runner.runner_parameters = {}
275			(status, output, _) = runner.run({})
276
277			expected_output = {
278			'stdout': 'stdout timeout',
279			'stderr': 'stderr timeout',
280			'return_code': 200,
281			'succeeded': False,
282			'failed': True,
283			'error': 'Action failed to complete in 5 seconds'
284			}
285
286			self.assertEqual(status, LIVEACTION_STATUS_TIMED_OUT)
287			self.assertDictEqual(output, expected_output)
288
289			# timeout with zero exit code
290			exit_code, stdout, stderr, timed_out = 0, 'stdout timeout', 'stderr timeout', True
291			runner._run_script = mock.Mock(return_value=(exit_code, stdout, stderr, timed_out))
292			runner._timeout = 5
293			runner._parse_winexe_error = mock.Mock(return_value='winexe error')
294
295			runner.runner_parameters = {}
296			(status, output, _) = runner.run({})
297
298			expected_output = {
299			'stdout': 'stdout timeout',
300			'stderr': 'stderr timeout',
301			'return_code': 0,
302			'succeeded': False,
303			'failed': True,
304			'error': 'Action failed to complete in 5 seconds: winexe error'
305			}
306
307			self.assertEqual(status, LIVEACTION_STATUS_TIMED_OUT)
308			self.assertDictEqual(output, expected_output)
309
310			def test_shell_command_parameter_escaping(self):
311			pass
312
313			def _get_base_runner(self):
314			class Runner(BaseWindowsRunner):
315			def pre_run(self):
316			pass
317
318			def run(self):
319			pass
320
321			runner = Runner('id')
322			return runner
323
324			def _get_mock_script_runner(self, action_parameters=None):
325			runner = WindowsScriptRunner('id')
326			runner._host = None
327			runner._username = None
328			runner._password = None
329			runner._timeout = None
330			runner._share = None
331
332			action_db = mock.Mock()
333			action_db.pack = 'dummy_pack_1'
334			action_db.entry_point = 'foo.py'
335			action_db.parameters = action_parameters or {}
336
337			runner.action = action_db
338
339			return runner
340

StackStorm / st2

Push — master ( e380d0...f5671d )

windows_runner/tests/unit/test_windows_runners.py (2 issues)

Labels

Severity

Introduced By

Duplication Side-by-Side

Filter issues like