Coverage for manila/tests/api/v2/test_share_instance_export_locations.py: 100%

1# Copyright (c) 2015 Mirantis Inc. 

2# All Rights Reserved. 

3# 

4# Licensed under the Apache License, Version 2.0 (the "License"); you may 

5# not use this file except in compliance with the License. You may obtain 

6# 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, WITHOUT 

12# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the 

13# License for the specific language governing permissions and limitations 

14# under the License. 

15 

16from unittest import mock 

17 

18import ddt 

19from webob import exc 

20 

21from manila.api.v2 import share_instance_export_locations as export_locations 

22from manila import context 

23from manila import db 

24from manila import exception 

25from manila import policy 

26from manila import test 

27from manila.tests.api import fakes 

28from manila.tests import db_utils 

29 

30 

31@ddt.ddt 

32class ShareInstanceExportLocationsAPITest(test.TestCase): 

33 

34 def _get_request(self, version="2.9", use_admin_context=True): 

35 req = fakes.HTTPRequest.blank( 

36 '/v2/share_instances/%s/export_locations' % self.share_instance_id, 

37 version=version, use_admin_context=use_admin_context) 

38 return req 

39 

40 def setUp(self): 

41 super(ShareInstanceExportLocationsAPITest, self).setUp() 

42 self.controller = ( 

43 export_locations.ShareInstanceExportLocationController()) 

44 self.resource_name = self.controller.resource_name 

45 self.ctxt = { 

46 'admin': context.RequestContext('admin', 'fake', True), 

47 'user': context.RequestContext('fake', 'fake'), 

48 } 

49 self.mock_policy_check = self.mock_object( 

50 policy, 'check_policy', mock.Mock(return_value=True)) 

51 self.share = db_utils.create_share() 

52 self.share_instance_id = self.share.instance.id 

53 self.req = self._get_request() 

54 data = [ 

55 { 

56 "path": ("10.254.0.3:/shares/" 

57 "share-e1c2d35e-fe67-4028-ad7a-45f668732b1d"), 

58 "share_instance_id": 

59 "e1c2d35e-fe67-4028-ad7a-45f668732b1d", 

60 "is_admin_only": False, 

61 "id": "b6bd76ce-12a2-42a9-a30a-8a43b503867d", 

62 "preferred": False 

63 }, 

64 { 

65 "path": ("10.0.0.3:/shares/" 

66 "share-e1c2d35e-fe67-4028-ad7a-45f668732b1d"), 

67 "share_instance_id": 

68 "e1c2d35e-fe67-4028-ad7a-45f668732b1d", 

69 "is_admin_only": True, 

70 "id": "6921e862-88bc-49a5-a2df-efeed9acd583", 

71 "preferred": False 

72 } 

73 ] 

74 db.export_locations_update( 

75 self.ctxt['admin'], self.share_instance_id, 

76 data, False) 

77 

78 @ddt.data({'role': 'admin', 'version': '2.9'}, 

79 {'role': 'user', 'version': '2.9'}, 

80 {'role': 'admin', 'version': '2.13'}, 

81 {'role': 'user', 'version': '2.13'}) 

82 @ddt.unpack 

83 def test_list_and_show(self, role, version): 

84 

85 summary_keys = ['id', 'path'] 

86 admin_summary_keys = summary_keys + [ 

87 'share_instance_id', 'is_admin_only'] 

88 detail_keys = summary_keys + ['created_at', 'updated_at'] 

89 admin_detail_keys = admin_summary_keys + ['created_at', 'updated_at'] 

90 

91 self._test_list_and_show(role, version, summary_keys, detail_keys, 

92 admin_summary_keys, admin_detail_keys) 

93 

94 @ddt.data('admin', 'user') 

95 def test_list_and_show_with_preferred_flag(self, role): 

96 

97 summary_keys = ['id', 'path', 'preferred'] 

98 admin_summary_keys = summary_keys + [ 

99 'share_instance_id', 'is_admin_only'] 

100 detail_keys = summary_keys + ['created_at', 'updated_at'] 

101 admin_detail_keys = admin_summary_keys + ['created_at', 'updated_at'] 

102 

103 self._test_list_and_show(role, '2.14', summary_keys, detail_keys, 

104 admin_summary_keys, admin_detail_keys) 

105 

106 def _test_list_and_show(self, role, version, summary_keys, detail_keys, 

107 admin_summary_keys, admin_detail_keys): 

108 

109 req = self._get_request(version=version, 

110 use_admin_context=(role == 'admin')) 

111 index_result = self.controller.index(req, self.share_instance_id) 

112 

113 self.assertIn('export_locations', index_result) 

114 self.assertEqual(1, len(index_result)) 

115 self.assertEqual(2, len(index_result['export_locations'])) 

116 

117 for index_el in index_result['export_locations']: 

118 self.assertIn('id', index_el) 

119 show_result = self.controller.show( 

120 req, self.share_instance_id, index_el['id']) 

121 self.assertIn('export_location', show_result) 

122 self.assertEqual(1, len(show_result)) 

123 

124 show_el = show_result['export_location'] 

125 

126 # Check summary keys in index result & detail keys in show result 

127 if role == 'admin': 

128 self.assertEqual(len(admin_summary_keys), len(index_el)) 

129 for key in admin_summary_keys: 

130 self.assertIn(key, index_el) 

131 self.assertEqual(len(admin_detail_keys), len(show_el)) 

132 for key in admin_detail_keys: 

133 self.assertIn(key, show_el) 

134 else: 

135 self.assertEqual(len(summary_keys), len(index_el)) 

136 for key in summary_keys: 

137 self.assertIn(key, index_el) 

138 self.assertEqual(len(detail_keys), len(show_el)) 

139 for key in detail_keys: 

140 self.assertIn(key, show_el) 

141 

142 # Ensure keys common to index & show results have matching values 

143 for key in summary_keys: 

144 self.assertEqual(index_el[key], show_el[key]) 

145 

146 def test_list_export_locations_share_instance_not_found(self): 

147 self.assertRaises( 

148 exc.HTTPNotFound, 

149 self.controller.index, 

150 self.req, 'inexistent_share_instance_id', 

151 ) 

152 

153 def test_show_export_location_share_instance_not_found(self): 

154 index_result = self.controller.index(self.req, self.share_instance_id) 

155 el_id = index_result['export_locations'][0]['id'] 

156 

157 self.assertRaises( 

158 exc.HTTPNotFound, 

159 self.controller.show, 

160 self.req, 'inexistent_share_id', el_id, 

161 ) 

162 

163 @ddt.data('1.0', '2.0', '2.8') 

164 def test_list_with_unsupported_version(self, version): 

165 self.assertRaises( 

166 exception.VersionNotFoundForAPIMethod, 

167 self.controller.index, 

168 self._get_request(version), 

169 self.share_instance_id, 

170 ) 

171 

172 @ddt.data('1.0', '2.0', '2.8') 

173 def test_show_with_unsupported_version(self, version): 

174 index_result = self.controller.index(self.req, self.share_instance_id) 

175 

176 self.assertRaises( 

177 exception.VersionNotFoundForAPIMethod, 

178 self.controller.show, 

179 self._get_request(version), 

180 self.share_instance_id, 

181 index_result['export_locations'][0]['id'] 

182 )