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

1# All Rights Reserved. 

2# 

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

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

5# a copy of the License at 

6# 

7# http://www.apache.org/licenses/LICENSE-2.0 

8# 

9# Unless required by applicable law or agreed to in writing, software 

10# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT 

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

12# License for the specific language governing permissions and limitations 

13# under the License. 

14 

15from unittest import mock 

16 

17import ddt 

18from webob import exc 

19 

20from manila.api.v2 import share_replica_export_locations as export_locations 

21from manila.common import constants 

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 

31GRADUATION_VERSION = '2.56' 

32 

33 

34@ddt.ddt 

35class ShareReplicaExportLocationsAPITest(test.TestCase): 

36 

37 def _get_request(self, version="2.47", use_admin_context=False): 

38 req = fakes.HTTPRequest.blank( 

39 '/v2/share-replicas/%s/export-locations' % self.active_replica_id, 

40 version=version, use_admin_context=use_admin_context, 

41 experimental=True) 

42 return req 

43 

44 def setUp(self): 

45 super(ShareReplicaExportLocationsAPITest, self).setUp() 

46 self.controller = ( 

47 export_locations.ShareReplicaExportLocationController()) 

48 self.resource_name = 'share_replica_export_location' 

49 self.ctxt = context.RequestContext('fake', 'fake') 

50 self.mock_policy_check = self.mock_object( 

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

52 self.share = db_utils.create_share( 

53 replication_type=constants.REPLICATION_TYPE_READABLE, 

54 replica_state=constants.REPLICA_STATE_ACTIVE) 

55 self.active_replica_id = self.share.instance.id 

56 self.req = self._get_request() 

57 exports = [ 

58 {'path': 'myshare.mydomain/active-replica-exp1', 

59 'is_admin_only': False}, 

60 {'path': 'myshare.mydomain/active-replica-exp2', 

61 'is_admin_only': False}, 

62 ] 

63 db.export_locations_update( 

64 self.ctxt, self.active_replica_id, exports) 

65 

66 # Replicas 

67 self.share_replica2 = db_utils.create_share_replica( 

68 share_id=self.share.id, 

69 replica_state=constants.REPLICA_STATE_IN_SYNC) 

70 self.share_replica3 = db_utils.create_share_replica( 

71 share_id=self.share.id, 

72 replica_state=constants.REPLICA_STATE_OUT_OF_SYNC) 

73 replica2_exports = [ 

74 {'path': 'myshare.mydomain/insync-replica-exp', 

75 'is_admin_only': False}, 

76 {'path': 'myshare.mydomain/insync-replica-exp2', 

77 'is_admin_only': False} 

78 ] 

79 replica3_exports = [ 

80 {'path': 'myshare.mydomain/outofsync-replica-exp', 

81 'is_admin_only': False}, 

82 {'path': 'myshare.mydomain/outofsync-replica-exp2', 

83 'is_admin_only': False} 

84 ] 

85 db.export_locations_update( 

86 self.ctxt, self.share_replica2.id, replica2_exports) 

87 db.export_locations_update( 

88 self.ctxt, self.share_replica3.id, replica3_exports) 

89 

90 @ddt.data(('user', '2.47'), ('admin', GRADUATION_VERSION)) 

91 @ddt.unpack 

92 def test_list_and_show(self, role, microversion): 

93 summary_keys = [ 

94 'id', 'path', 'replica_state', 'availability_zone', 'preferred' 

95 ] 

96 admin_summary_keys = summary_keys + [ 

97 'share_instance_id', 'is_admin_only' 

98 ] 

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

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

101 

102 self._test_list_and_show(role, summary_keys, detail_keys, 

103 admin_summary_keys, admin_detail_keys, 

104 microversion=microversion) 

105 

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

107 admin_summary_keys, admin_detail_keys, 

108 microversion='2.47'): 

109 

110 req = self._get_request(version=microversion, 

111 use_admin_context=(role == 'admin')) 

112 for replica_id in (self.active_replica_id, self.share_replica2.id, 

113 self.share_replica3.id): 

114 index_result = self.controller.index(req, replica_id) 

115 

116 self.assertIn('export_locations', index_result) 

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

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

119 

120 for index_el in index_result['export_locations']: 

121 self.assertIn('id', index_el) 

122 show_result = self.controller.show( 

123 req, replica_id, index_el['id']) 

124 self.assertIn('export_location', show_result) 

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

126 

127 show_el = show_result['export_location'] 

128 

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

130 if role == 'admin': 

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

132 for key in admin_summary_keys: 

133 self.assertIn(key, index_el) 

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

135 for key in admin_detail_keys: 

136 self.assertIn(key, show_el) 

137 else: 

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

139 for key in summary_keys: 

140 self.assertIn(key, index_el) 

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

142 for key in detail_keys: 

143 self.assertIn(key, show_el) 

144 

145 # Ensure keys common to index & show have matching values 

146 for key in summary_keys: 

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

148 

149 def test_list_and_show_with_non_replicas(self): 

150 non_replicated_share = db_utils.create_share() 

151 instance_id = non_replicated_share.instance.id 

152 exports = [ 

153 {'path': 'myshare.mydomain/non-replicated-share', 

154 'is_admin_only': False}, 

155 {'path': 'myshare.mydomain/non-replicated-share-2', 

156 'is_admin_only': False}, 

157 ] 

158 db.export_locations_update(self.ctxt, instance_id, exports) 

159 updated_exports = db.export_location_get_all_by_share_id( 

160 self.ctxt, non_replicated_share.id) 

161 

162 self.assertRaises(exc.HTTPNotFound, self.controller.index, self.req, 

163 instance_id) 

164 

165 for export in updated_exports: 

166 self.assertRaises(exc.HTTPNotFound, self.controller.show, self.req, 

167 instance_id, export['id']) 

168 

169 def test_list_export_locations_share_replica_not_found(self): 

170 self.assertRaises( 

171 exc.HTTPNotFound, 

172 self.controller.index, 

173 self.req, 'non-existent-share-replica-id') 

174 

175 def test_show_export_location_share_replica_not_found(self): 

176 index_result = self.controller.index(self.req, self.active_replica_id) 

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

178 

179 self.assertRaises( 

180 exc.HTTPNotFound, 

181 self.controller.show, 

182 self.req, 'non-existent-share-replica-id', el_id) 

183 

184 self.assertRaises( 

185 exc.HTTPNotFound, 

186 self.controller.show, 

187 self.req, self.active_replica_id, 

188 'non-existent-export-location-id') 

189 

190 @ddt.data('1.0', '2.0', '2.46') 

191 def test_list_with_unsupported_version(self, version): 

192 self.assertRaises( 

193 exception.VersionNotFoundForAPIMethod, 

194 self.controller.index, 

195 self._get_request(version), 

196 self.active_replica_id) 

197 

198 @ddt.data('1.0', '2.0', '2.46') 

199 def test_show_with_unsupported_version(self, version): 

200 index_result = self.controller.index(self.req, self.active_replica_id) 

201 

202 self.assertRaises( 

203 exception.VersionNotFoundForAPIMethod, 

204 self.controller.show, 

205 self._get_request(version), 

206 self.active_replica_id, 

207 index_result['export_locations'][0]['id'])