CloudSlangRunnerTestCase.test_temp_file_deletes_when_exception_occurs() - 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_temp_file_deletes_when_exception_occurs() A

↳ Parent: CloudSlangRunnerTestCase

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Importance

Changes	2
Bugs	1	Features	0

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

# 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
import mock

from unittest2 import TestCase

from st2common.constants.action import LIVEACTION_STATUS_SUCCEEDED
from st2common.constants.action import LIVEACTION_STATUS_FAILED
from cloudslang 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)
        self.assertEqual(runner._timeout, timeout)

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

StackStorm / st2

Push — develop ( b8d4ca...421369 )

test_temp_file_deletes_when_exception_occurs() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like