WindowsRunnerTestCase.test_shell_command_parameter_escaping() - Code Metrics - Inspection of "Merge pull request #2824 from StackStorm/pluggable..." - StackStorm/st2 - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — develop ( b8d4ca...421369 )

by Plexxi

created 2016-10-13 10:00 UTC

test_shell_command_parameter_escaping() A

↳ Parent: WindowsRunnerTestCase

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Importance

Changes	1
Bugs	1	Features	0

Metric	Value
cc	1
c	1
b	1
f	0
dl	0
loc	2
rs	10

# 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.

import os
from unittest2 import TestCase

import mock

from windows_runner import BaseWindowsRunner
from 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_script_runner()
        for arguments, expected_value in zip(arguments, expected_values):
            actual_value = runner._get_script_arguments(**arguments)
            self.assertEqual(actual_value, expected_value)

    def test_parse_share_information(self):
        runner = self._get_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_script_runner.run_command')
    def test_get_share_absolute_path(self, mock_run_command):
        runner = self._get_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_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_script_runner(self):
        runner = WindowsScriptRunner('id')
        runner._host = None
        runner._username = None
        runner._password = None
        runner._timeout = None

        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			import os
17			from unittest2 import TestCase
18
19			import mock
20
21			from windows_runner import BaseWindowsRunner
22			from windows_script_runner import WindowsScriptRunner
23
24			BASE_DIR = os.path.dirname(os.path.abspath(__file__))
25			FIXTURES_DIR = os.path.abspath(os.path.join(BASE_DIR, '../fixtures/windows'))
26
27
28			class WindowsRunnerTestCase(TestCase):
29			def test_get_winexe_command_args(self):
30			arguments = [
31			{
32			'host': 'localhost',
33			'username': 'Administrator1',
34			'password': 'bar1',
35			'command': 'powershell.exe "C:\\\\myscript.ps1"'
36			},
37			{
38			'host': '127.0.0.1',
39			'username': 'Administrator2',
40			'password': 'bar2',
41			'command': 'dir'
42			},
43			{
44			'host': 'localhost',
45			'username': 'Administrator3',
46			'password': 'bar3',
47			'command': 'dir',
48			'domain': 'MyDomain'
49			}
50			]
51			expected_values = [
52			[
53			'winexe',
54			'--interactive', '0',
55			'-U', 'Administrator1%bar1',
56			'//localhost',
57			'powershell.exe "C:\\\\myscript.ps1"'
58			],
59			[
60			'winexe',
61			'--interactive', '0',
62			'-U', 'Administrator2%bar2',
63			'//127.0.0.1',
64			'dir'
65			],
66			[
67			'winexe',
68			'--interactive', '0',
69			'-U', 'MyDomain\Administrator3%bar3',
70			'//localhost',
71			'dir'
72			]
73			]
74
75			runner = self._get_base_runner()
76			for arguments, expected_value in zip(arguments, expected_values):
77			actual_value = runner._get_winexe_command_args(**arguments)
78			self.assertEqual(actual_value, expected_value)
79
80			def test_get_smbclient_command_args(self):
81			arguments = [
82			{
83			'host': 'localhost',
84			'username': 'Administrator1',
85			'password': 'bar1',
86			'command': 'put /home/1.txt 1.txt',
87			'share': 'C$'
88			},
89			{
90			'host': 'localhost',
91			'username': 'Administrator2',
92			'password': 'bar2',
93			'command': 'put /home/2.txt 2.txt',
94			'share': 'D$'
95			},
96			{
97			'host': 'localhost',
98			'username': 'Administrator3',
99			'password': 'bar3',
100			'command': 'dir',
101			'share': 'E$',
102			'domain': 'MyDomain'
103			}
104			]
105			expected_values = [
106			[
107			'smbclient',
108			'-U', 'Administrator1%bar1',
109			'//localhost/C$',
110			'-c', 'put /home/1.txt 1.txt'
111			],
112			[
113			'smbclient',
114			'-U', 'Administrator2%bar2',
115			'//localhost/D$',
116			'-c', 'put /home/2.txt 2.txt'
117			],
118			[
119			'smbclient',
120			'-U', 'MyDomain\Administrator3%bar3',
121			'//localhost/E$',
122			'-c', 'dir'
123			],
124			]
125
126			runner = self._get_base_runner()
127			for arguments, expected_value in zip(arguments, expected_values):
128			actual_value = runner._get_smbclient_command_args(**arguments)
129			self.assertEqual(actual_value, expected_value)
130
131			def test_get_script_args(self):
132			arguments = [
133			{
134			'positional_args': 'a b c',
135			'named_args': {
136			'arg1': 'value1',
137			'arg2': 'value2'
138			}
139			},
140			{
141			'positional_args': 'a b c',
142			'named_args': {
143			'arg1': 'value1',
144			'arg2': True,
145			'arg3': False,
146			'arg4': ['foo', 'bar', 'baz']
147			}
148			}
149			]
150			expected_values = [
151			'a b c -arg1 value1 -arg2 value2',
152			'a b c -arg1 value1 -arg2 -arg3:$false -arg4 foo,bar,baz'
153			]
154
155			runner = self._get_script_runner()
156			for arguments, expected_value in zip(arguments, expected_values):
157			actual_value = runner._get_script_arguments(**arguments)
158			self.assertEqual(actual_value, expected_value)
159
160			def test_parse_share_information(self):
161			runner = self._get_script_runner()
162
163			fixture_path = os.path.join(FIXTURES_DIR, 'net_share_C_stdout.txt')
164			with open(fixture_path, 'r') as fp:
165			stdout = fp.read()
166
167			result = runner._parse_share_information(stdout=stdout)
168
169			expected_keys = ['share_name', 'path', 'remark', 'maximum_users', 'users', 'caching',
170			'permission']
171			for key in expected_keys:
172			self.assertTrue(key in result)
173
174			self.assertEqual(result['share_name'], 'C$')
175			self.assertEqual(result['path'], 'C:\\')
176			self.assertEqual(result['users'], None)
177
178			@mock.patch('windows_script_runner.run_command')
179			def test_get_share_absolute_path(self, mock_run_command):
180			runner = self._get_script_runner()
181
182			fixture_path = os.path.join(FIXTURES_DIR, 'net_share_C_stdout.txt')
183			with open(fixture_path, 'r') as fp:
184			stdout = fp.read()
185
186			# Failure, non-zero status code
187			mock_run_command.return_value = (2, '', '', False)
188			self.assertRaises(Exception, runner._get_share_absolute_path, share='C$')
189
190			# Failure, missing / corrupted data
191			mock_run_command.return_value = (0, '', '', False)
192			self.assertRaises(Exception, runner._get_share_absolute_path, share='C$')
193
194			# Success, everything OK
195			mock_run_command.return_value = (0, stdout, '', False)
196			share_path = runner._get_share_absolute_path(share='C$')
197			self.assertEqual(share_path, 'C:\\')
198
199			def test_shell_command_parameter_escaping(self):
200			pass
201
202			def _get_base_runner(self):
203			class Runner(BaseWindowsRunner):
204			def pre_run(self):
205			pass
206
207			def run(self):
208			pass
209
210			runner = Runner('id')
211			return runner
212
213			def _get_script_runner(self):
214			runner = WindowsScriptRunner('id')
215			runner._host = None
216			runner._username = None
217			runner._password = None
218			runner._timeout = None
219
220			return runner
221

StackStorm / st2

Push — develop ( b8d4ca...421369 )

test_shell_command_parameter_escaping() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like