Coverage for plugin/tests/test_plugin.py : 97%

Hot-keys on this page
r m x p toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
1# ============LICENSE_START==========================================
2# ===================================================================
3# Copyright (c) 2018-2020 AT&T
4#
5# Licensed under the Apache License, Version 2.0 (the "License");
6# you may not use this file except in compliance with the License.
7# You may obtain a copy of the License at
8#
9# http://www.apache.org/licenses/LICENSE-2.0
10#
11# Unless required by applicable law or agreed to in writing, software
12# distributed under the License is distributed on an "AS IS" BASIS,
13# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14# See the License for the specific language governing permissions and
15# limitations under the License.
16# ============LICENSE_END============================================
19from os import path
20import unittest
21import mock
22import plugin.tasks
24from cloudify.test_utils import workflow_test
25from cloudify.mocks import MockNodeInstanceContext
26from cloudify.mocks import MockCloudifyContext
27from cloudify.state import current_ctx
28from cloudify import ctx
31class TestPlugin(unittest.TestCase):
33 @workflow_test(path.join('blueprint', 'blueprint.yaml'),
34 resources_to_copy=[(path.join('blueprint', 'plugin',
35 'test_plugin.yaml'),
36 'plugin')])
37 @mock.patch('plugin.tasks.os.remove')
38 @mock.patch('plugin.tasks.execute_command')
39 def test_stop(self, cfy_local, mock_execute_command, mock_os_remove):
40 # execute install workflow
41 """
43 :param cfy_local:
44 """
45 with mock.patch('plugin.tasks.shutil.rmtree'):
46 cfy_local.execute('uninstall', task_retries=0)
48 # extract single node instance
49 instance = cfy_local.storage.get_node_instances()[0]
51 mock_execute_command.assert_called_with('helm delete --purge onap-test_node --host 1.1.1.1:8888 ')
53 @workflow_test(path.join('blueprint', 'blueprint.yaml'),
54 resources_to_copy=[(path.join('blueprint', 'plugin',
55 'test_plugin.yaml'),
56 'plugin')])
57 @mock.patch('plugin.tasks.execute_command')
58 def test_start(self, cfy_local, mock_execute_command):
59 # execute install workflow
60 """
62 :param cfy_local:
63 """
64 with mock.patch('plugin.tasks.config'):
65 with mock.patch('plugin.tasks.get_current_helm_value'):
66 with mock.patch('plugin.tasks.get_helm_history'):
67 cfy_local.execute('install', task_retries=0)
69 # extract single node instance
70 instance = cfy_local.storage.get_node_instances()[0]
72 mock_execute_command.assert_called_with('helm install local/test_node-2.0.0.tgz --name onap-test_node --namespace onap --host 1.1.1.1:8888 ')
74 @workflow_test(path.join('blueprint', 'blueprint.yaml'),
75 resources_to_copy=[(path.join('blueprint', 'plugin',
76 'test_plugin.yaml'),
77 'plugin')])
78 @mock.patch('plugin.tasks.execute_command')
79 def test_config(self, cfy_local, mock_execute_command):
80 # execute install workflow
81 """
83 :param cfy_local:
84 """
85 with mock.patch('plugin.tasks.start'):
86 cfy_local.execute('install', task_retries=0)
88 # extract single node instance
89 instance = cfy_local.storage.get_node_instances()[0]
91 mock_execute_command.assert_called_with('helm init --client-only --stable-repo-url http://0.0.0.0/stable')
93 @workflow_test(path.join('blueprint', 'blueprint.yaml'),
94 resources_to_copy=[(path.join('blueprint', 'plugin',
95 'test_plugin.yaml'),
96 'plugin')])
97 def test_rollback(self, cfy_local):
98 # execute install workflow
99 """
101 :param cfy_local:
102 """
103 node_instance_id = 'node_instance_id'
104 revision = 1
105 try:
106 cfy_local.execute('rollback', task_retries=0,
107 parameters={'node_instance_id': node_instance_id, 'revision': revision})
108 self.fail('Expected exception due to operation not exist')
109 except Exception as e:
110 self.assertTrue('operation not available')
112 @workflow_test(path.join('blueprint', 'blueprint.yaml'),
113 resources_to_copy=[(path.join('blueprint', 'plugin',
114 'test_plugin.yaml'),
115 'plugin')])
116 def test_upgrade(self, cfy_local):
117 # execute install workflow
118 """
120 :param cfy_local:
121 """
122 node_instance_id = 'node_instance_id'
123 config_json = ''
124 config_url = 'http://test:test@11.22.33.44:80/stable'
125 config_format = 'json'
126 chartVersion = '2.0.0'
127 chartRepo = 'repo'
128 repo_user = ''
129 repo_user_passwd = ''
130 try:
131 cfy_local.execute('upgrade', task_retries=0,
132 parameters={'node_instance_id': node_instance_id, 'config': config_json,
133 'config_url': config_url, 'config_format': config_format,
134 'chart_version': chartVersion, 'chart_repo_url': chartRepo,
135 'repo_user': repo_user, 'repo_user_password': repo_user_passwd})
136 self.fail('Expected exception due to operation not exist')
137 except Exception as e:
138 self.assertTrue('operation not available')
140 @mock.patch('plugin.tasks.execute_command')
141 def test_op_rollback(self, mock_execute_command):
142 # test operation rollback
143 """
145 :rollback operation test:
146 """
147 props = {
148 'component_name': 'test_node',
149 'namespace': 'onap',
150 'tiller_port': '8888',
151 'tiller_ip': '1.1.1.1',
152 'tls_enable': 'false'
153 }
154 args = {'revision': '1'}
155 mock_ctx = MockCloudifyContext(node_id='test_node_id', node_name='test_node_name',
156 properties=props)
157 try:
158 current_ctx.set(mock_ctx)
159 with mock.patch('plugin.tasks.get_current_helm_value'):
160 with mock.patch('plugin.tasks.get_helm_history'):
161 plugin.tasks.rollback(**args)
162 finally:
163 current_ctx.clear()
165 @mock.patch('plugin.tasks.execute_command')
166 def test_op_upgrade(self, mock_execute_command):
167 # test operation upgrade
168 """
170 :upgrade operation test:
171 """
172 props = {
173 'component_name': 'test_node',
174 'namespace': 'onap',
175 'tiller_port': '8888',
176 'tiller_ip': '1.1.1.1',
177 'tls_enable': 'false',
178 'config_dir': '/tmp'
179 }
180 args = {'revision': '1', 'config': '', 'chart_repo': 'repo', 'chart_version': '2',
181 'config_set': 'config_set', 'config_json': '', 'config_url': '',
182 'config_format': 'format', 'repo_user': '', 'repo_user_passwd': ''}
183 mock_ctx = MockCloudifyContext(node_id='test_node_id', node_name='test_node_name',
184 properties=props)
185 try:
186 current_ctx.set(mock_ctx)
187 with mock.patch('plugin.tasks.get_current_helm_value'):
188 with mock.patch('plugin.tasks.get_helm_history'):
189 with mock.patch('plugin.tasks.gen_config_str'):
190 plugin.tasks.upgrade(**args)
191 finally:
192 current_ctx.clear()