Issues in test_cloudslang.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

cloudslang_runner/tests/unit/test_cloudslang.py (2 issues)

Labels

Severity

Major 2

Eliya Sadan 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
import mock

from unittest2 import TestCase

from st2common.constants.action import LIVEACTION_STATUS_SUCCEEDED
from st2common.constants.action import LIVEACTION_STATUS_FAILED
from cloudslang_runner import cloudslang_runner as csr

import st2tests.config as tests_config
tests_config.parse_args()


class CloudSlangRunnerTestCase(TestCase):

    def test_runner_creation(self):
        runner = csr.get_runner()
        self.assertTrue(runner)
        self.assertTrue(runner.runner_id)

    def test_pre_run_sets_attributes(self):
        entry_point = 'path'
        inputs = {'a': 1}
        timeout = 10

        runner = csr.get_runner()
        runner.entry_point = entry_point
        runner.runner_parameters = {
            csr.RUNNER_INPUTS: inputs,
            csr.RUNNER_TIMEOUT: timeout,
        }
        runner.pre_run()
        self.assertEqual(runner.entry_point, entry_point)
        self.assertEqual(runner._inputs, inputs)
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(runner._timeout, timeout)
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

    @mock.patch('cloudslang_runner.cloudslang_runner.quote_unix')
    @mock.patch('cloudslang_runner.cloudslang_runner.run_command')
    def test_run_calls_a_new_process_success(self, mock_run_command, mock_quote_unix):
        entry_point = 'path'
        timeout = 1

        runner = csr.get_runner()
        runner.entry_point = entry_point
        runner.runner_parameters = {
            csr.RUNNER_INPUTS: None,
            csr.RUNNER_TIMEOUT: timeout,
        }
        runner.pre_run()
        mock_run_command.return_value = (0, "", "", False)
        mock_quote_unix.return_value = ""
        result = runner.run({})
        mock_quote_unix.assert_called_with(tests_config.CONF.cloudslang.home_dir)
        self.assertTrue(mock_run_command.called)
        self.assertEqual(LIVEACTION_STATUS_SUCCEEDED, result[0])

    @mock.patch('cloudslang_runner.cloudslang_runner.quote_unix')
    @mock.patch('cloudslang_runner.cloudslang_runner.run_command')
    def test_run_calls_a_new_process_failure(self, mock_run_command, mock_quote_unix):
        timeout = 1
        runner = csr.get_runner()
        runner.runner_parameters = {
            csr.RUNNER_INPUTS: None,
            csr.RUNNER_TIMEOUT: timeout,
        }
        runner.pre_run()
        mock_run_command.return_value = (1, "", "", False)
        mock_quote_unix.return_value = ""
        result = runner.run({})
        mock_quote_unix.assert_called_with(tests_config.CONF.cloudslang.home_dir)
        self.assertTrue(mock_run_command.called)
        self.assertEqual(LIVEACTION_STATUS_FAILED, result[0])

    @mock.patch('cloudslang_runner.cloudslang_runner.run_command')
    def test_run_calls_a_new_process_timeout(self, mock_run_command):
        entry_point = 'path'
        timeout = 1
        runner = csr.get_runner()
        runner.entry_point = entry_point
        runner.runner_parameters = {
            csr.RUNNER_INPUTS: None,
            csr.RUNNER_TIMEOUT: timeout,
        }
        runner.pre_run()
        mock_run_command.return_value = (0, "", "", True)
        result = runner.run({})
        self.assertTrue(mock_run_command.called)
        self.assertEqual(LIVEACTION_STATUS_FAILED, result[0])

    @mock.patch('cloudslang_runner.cloudslang_runner.run_command')
    @mock.patch('cloudslang_runner.cloudslang_runner.yaml.safe_dump')
    def test_inputs_are_save_to_file_properly(self, mock_yaml_dump, mock_run_command):
        entry_point = 'path'
        inputs = {'a': 1}
        timeout = 1
        runner = csr.get_runner()
        runner.entry_point = entry_point
        runner.runner_parameters = {
            csr.RUNNER_INPUTS: inputs,
            csr.RUNNER_TIMEOUT: timeout,
        }
        runner.pre_run()
        mock_run_command.return_value = (0, "", "", True)
        mock_yaml_dump.return_value = ""
        result = runner.run({})
        self.assertTrue(mock_run_command.called)
        mock_yaml_dump.assert_called_with(inputs, default_flow_style=False)
        self.assertEqual(LIVEACTION_STATUS_FAILED, result[0])

    @mock.patch('cloudslang_runner.cloudslang_runner.run_command')
    @mock.patch('cloudslang_runner.cloudslang_runner.os.remove')
    def test_temp_file_deletes_when_exception_occurs(self, mock_os_remove, mock_run_command):
        entry_point = 'path'
        inputs = {'a': 1}
        timeout = 1
        runner = csr.get_runner()
        runner.entry_point = entry_point
        runner.runner_parameters = {
            csr.RUNNER_INPUTS: inputs,
            csr.RUNNER_TIMEOUT: timeout,
        }
        runner.pre_run()
        mock_run_command.return_value = (0, "", "", True)
        mock_run_command.side_effect = IOError('Boom!')
        with self.assertRaisesRegex(IOError, "Boom!"):
            runner.run({})
        self.assertTrue(mock_os_remove.called)

        # lets really remove it now
        os.remove(mock_os_remove.call_args[0][0])

    @mock.patch('cloudslang_runner.cloudslang_runner.run_command')
    def test_inputs_provided_via_inputs_runner_parameter(self, mock_run_command):
        entry_point = 'path'
        inputs = {'a': 1}
        timeout = 1

        runner = csr.get_runner()
        runner.entry_point = entry_point
        runner.runner_parameters = {
            csr.RUNNER_INPUTS: inputs,
            csr.RUNNER_TIMEOUT: timeout,
        }
        runner._write_inputs_to_a_temp_file = mock.Mock()
        runner._write_inputs_to_a_temp_file.return_value = None

        mock_run_command.return_value = (0, "", "", False)
        runner.pre_run()
        runner.run({})
        runner._write_inputs_to_a_temp_file.assert_called_with(inputs=inputs)

    @mock.patch('cloudslang_runner.cloudslang_runner.run_command')
    def test_inputs_provided_via_action_parameters(self, mock_run_command):
        entry_point = 'path'
        inputs = None
        timeout = 1
        action_parameters = {'foo': 'bar'}

        runner = csr.get_runner()
        runner.entry_point = entry_point
        runner.runner_parameters = {
            csr.RUNNER_INPUTS: inputs,
            csr.RUNNER_TIMEOUT: timeout,
        }
        runner._write_inputs_to_a_temp_file = mock.Mock()
        runner._write_inputs_to_a_temp_file.return_value = None

        mock_run_command.return_value = (0, "", "", False)
        runner.pre_run()
        runner.run(action_parameters)
        runner._write_inputs_to_a_temp_file.assert_called_with(inputs=action_parameters)

    def test_prepare_command(self):
        entry_point = 'flow_path'
        inputs = None
        timeout = 1

        runner = csr.get_runner()
        runner.entry_point = entry_point
        runner.runner_parameters = {
            csr.RUNNER_INPUTS: inputs,
            csr.RUNNER_TIMEOUT: timeout,
        }
        runner.pre_run()

        # No inputs
        result = runner._prepare_command(has_inputs=False, inputs_file_path=None)
        expected = '/opt/cslang/bin/cslang run --f flow_path --cp /opt/cslang'
        self.assertEqual(result, expected)

        # Inputs
        result = runner._prepare_command(has_inputs=True, inputs_file_path='inputs_file')
        expected = '/opt/cslang/bin/cslang run --f flow_path --cp /opt/cslang --if inputs_file'
        self.assertEqual(result, expected)


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			import mock
19
20			from unittest2 import TestCase
21
22			from st2common.constants.action import LIVEACTION_STATUS_SUCCEEDED
23			from st2common.constants.action import LIVEACTION_STATUS_FAILED
24			from cloudslang_runner import cloudslang_runner as csr
25
26			import st2tests.config as tests_config
27			tests_config.parse_args()
28
29
30			class CloudSlangRunnerTestCase(TestCase):
31
32			def test_runner_creation(self):
33			runner = csr.get_runner()
34			self.assertTrue(runner)
35			self.assertTrue(runner.runner_id)
36
37			def test_pre_run_sets_attributes(self):
38			entry_point = 'path'
39			inputs = {'a': 1}
40			timeout = 10
41
42			runner = csr.get_runner()
43			runner.entry_point = entry_point
44			runner.runner_parameters = {
45			csr.RUNNER_INPUTS: inputs,
46			csr.RUNNER_TIMEOUT: timeout,
47			}
48			runner.pre_run()
49			self.assertEqual(runner.entry_point, entry_point)
50			self.assertEqual(runner._inputs, inputs)
			0 ignored issues – show Coding Style Best Practice introduced 2017-10-26 18:09 UTC by Report Bug Copy Issue Report Show Similar Issues like this It seems like `_inputs` 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...
51			self.assertEqual(runner._timeout, timeout)
			0 ignored issues – show Coding Style Best Practice introduced 2017-10-26 18:09 UTC by Report Bug Copy Issue Report Show Similar Issues like this It seems like `_timeout` 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...
52
53			@mock.patch('cloudslang_runner.cloudslang_runner.quote_unix')
54			@mock.patch('cloudslang_runner.cloudslang_runner.run_command')
55			def test_run_calls_a_new_process_success(self, mock_run_command, mock_quote_unix):
56			entry_point = 'path'
57			timeout = 1
58
59			runner = csr.get_runner()
60			runner.entry_point = entry_point
61			runner.runner_parameters = {
62			csr.RUNNER_INPUTS: None,
63			csr.RUNNER_TIMEOUT: timeout,
64			}
65			runner.pre_run()
66			mock_run_command.return_value = (0, "", "", False)
67			mock_quote_unix.return_value = ""
68			result = runner.run({})
69			mock_quote_unix.assert_called_with(tests_config.CONF.cloudslang.home_dir)
70			self.assertTrue(mock_run_command.called)
71			self.assertEqual(LIVEACTION_STATUS_SUCCEEDED, result[0])
72
73			@mock.patch('cloudslang_runner.cloudslang_runner.quote_unix')
74			@mock.patch('cloudslang_runner.cloudslang_runner.run_command')
75			def test_run_calls_a_new_process_failure(self, mock_run_command, mock_quote_unix):
76			timeout = 1
77			runner = csr.get_runner()
78			runner.runner_parameters = {
79			csr.RUNNER_INPUTS: None,
80			csr.RUNNER_TIMEOUT: timeout,
81			}
82			runner.pre_run()
83			mock_run_command.return_value = (1, "", "", False)
84			mock_quote_unix.return_value = ""
85			result = runner.run({})
86			mock_quote_unix.assert_called_with(tests_config.CONF.cloudslang.home_dir)
87			self.assertTrue(mock_run_command.called)
88			self.assertEqual(LIVEACTION_STATUS_FAILED, result[0])
89
90			@mock.patch('cloudslang_runner.cloudslang_runner.run_command')
91			def test_run_calls_a_new_process_timeout(self, mock_run_command):
92			entry_point = 'path'
93			timeout = 1
94			runner = csr.get_runner()
95			runner.entry_point = entry_point
96			runner.runner_parameters = {
97			csr.RUNNER_INPUTS: None,
98			csr.RUNNER_TIMEOUT: timeout,
99			}
100			runner.pre_run()
101			mock_run_command.return_value = (0, "", "", True)
102			result = runner.run({})
103			self.assertTrue(mock_run_command.called)
104			self.assertEqual(LIVEACTION_STATUS_FAILED, result[0])
105
106			@mock.patch('cloudslang_runner.cloudslang_runner.run_command')
107			@mock.patch('cloudslang_runner.cloudslang_runner.yaml.safe_dump')
108			def test_inputs_are_save_to_file_properly(self, mock_yaml_dump, mock_run_command):
109			entry_point = 'path'
110			inputs = {'a': 1}
111			timeout = 1
112			runner = csr.get_runner()
113			runner.entry_point = entry_point
114			runner.runner_parameters = {
115			csr.RUNNER_INPUTS: inputs,
116			csr.RUNNER_TIMEOUT: timeout,
117			}
118			runner.pre_run()
119			mock_run_command.return_value = (0, "", "", True)
120			mock_yaml_dump.return_value = ""
121			result = runner.run({})
122			self.assertTrue(mock_run_command.called)
123			mock_yaml_dump.assert_called_with(inputs, default_flow_style=False)
124			self.assertEqual(LIVEACTION_STATUS_FAILED, result[0])
125
126			@mock.patch('cloudslang_runner.cloudslang_runner.run_command')
127			@mock.patch('cloudslang_runner.cloudslang_runner.os.remove')
128			def test_temp_file_deletes_when_exception_occurs(self, mock_os_remove, mock_run_command):
129			entry_point = 'path'
130			inputs = {'a': 1}
131			timeout = 1
132			runner = csr.get_runner()
133			runner.entry_point = entry_point
134			runner.runner_parameters = {
135			csr.RUNNER_INPUTS: inputs,
136			csr.RUNNER_TIMEOUT: timeout,
137			}
138			runner.pre_run()
139			mock_run_command.return_value = (0, "", "", True)
140			mock_run_command.side_effect = IOError('Boom!')
141			with self.assertRaisesRegex(IOError, "Boom!"):
142			runner.run({})
143			self.assertTrue(mock_os_remove.called)
144
145			# lets really remove it now
146			os.remove(mock_os_remove.call_args[0][0])
147
148			@mock.patch('cloudslang_runner.cloudslang_runner.run_command')
149			def test_inputs_provided_via_inputs_runner_parameter(self, mock_run_command):
150			entry_point = 'path'
151			inputs = {'a': 1}
152			timeout = 1
153
154			runner = csr.get_runner()
155			runner.entry_point = entry_point
156			runner.runner_parameters = {
157			csr.RUNNER_INPUTS: inputs,
158			csr.RUNNER_TIMEOUT: timeout,
159			}
160			runner._write_inputs_to_a_temp_file = mock.Mock()
161			runner._write_inputs_to_a_temp_file.return_value = None
162
163			mock_run_command.return_value = (0, "", "", False)
164			runner.pre_run()
165			runner.run({})
166			runner._write_inputs_to_a_temp_file.assert_called_with(inputs=inputs)
167
168			@mock.patch('cloudslang_runner.cloudslang_runner.run_command')
169			def test_inputs_provided_via_action_parameters(self, mock_run_command):
170			entry_point = 'path'
171			inputs = None
172			timeout = 1
173			action_parameters = {'foo': 'bar'}
174
175			runner = csr.get_runner()
176			runner.entry_point = entry_point
177			runner.runner_parameters = {
178			csr.RUNNER_INPUTS: inputs,
179			csr.RUNNER_TIMEOUT: timeout,
180			}
181			runner._write_inputs_to_a_temp_file = mock.Mock()
182			runner._write_inputs_to_a_temp_file.return_value = None
183
184			mock_run_command.return_value = (0, "", "", False)
185			runner.pre_run()
186			runner.run(action_parameters)
187			runner._write_inputs_to_a_temp_file.assert_called_with(inputs=action_parameters)
188
189			def test_prepare_command(self):
190			entry_point = 'flow_path'
191			inputs = None
192			timeout = 1
193
194			runner = csr.get_runner()
195			runner.entry_point = entry_point
196			runner.runner_parameters = {
197			csr.RUNNER_INPUTS: inputs,
198			csr.RUNNER_TIMEOUT: timeout,
199			}
200			runner.pre_run()
201
202			# No inputs
203			result = runner._prepare_command(has_inputs=False, inputs_file_path=None)
204			expected = '/opt/cslang/bin/cslang run --f flow_path --cp /opt/cslang'
205			self.assertEqual(result, expected)
206
207			# Inputs
208			result = runner._prepare_command(has_inputs=True, inputs_file_path='inputs_file')
209			expected = '/opt/cslang/bin/cslang run --f flow_path --cp /opt/cslang --if inputs_file'
210			self.assertEqual(result, expected)
211

StackStorm / st2

Push — master ( e380d0...f5671d )

cloudslang_runner/tests/unit/test_cloudslang.py (2 issues)

Labels

Severity

Introduced By

Duplication Side-by-Side

Filter issues like