Hide keyboard shortcuts

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============================================ 

17 

18 

19from os import path 

20import unittest 

21import mock 

22import plugin.tasks 

23 

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 

29 

30 

31class TestPlugin(unittest.TestCase): 

32 

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 """ 

42 

43 :param cfy_local: 

44 """ 

45 with mock.patch('plugin.tasks.shutil.rmtree'): 

46 cfy_local.execute('uninstall', task_retries=0) 

47 

48 # extract single node instance 

49 instance = cfy_local.storage.get_node_instances()[0] 

50 

51 mock_execute_command.assert_called_with('helm delete --purge onap-test_node --host 1.1.1.1:8888 ') 

52 

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 """ 

61 

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) 

68 

69 # extract single node instance 

70 instance = cfy_local.storage.get_node_instances()[0] 

71 

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 ') 

73 

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 """ 

82 

83 :param cfy_local: 

84 """ 

85 with mock.patch('plugin.tasks.start'): 

86 cfy_local.execute('install', task_retries=0) 

87 

88 # extract single node instance 

89 instance = cfy_local.storage.get_node_instances()[0] 

90 

91 mock_execute_command.assert_called_with('helm init --client-only --stable-repo-url http://0.0.0.0/stable') 

92 

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 """ 

100 

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') 

111 

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 """ 

119 

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') 

139 

140 @mock.patch('plugin.tasks.execute_command') 

141 def test_op_rollback(self, mock_execute_command): 

142 # test operation rollback 

143 """ 

144 

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() 

164 

165 @mock.patch('plugin.tasks.execute_command') 

166 def test_op_upgrade(self, mock_execute_command): 

167 # test operation upgrade 

168 """ 

169 

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()