Coverage for manila/tests/api/test_extensions.py: 95%

1# Copyright (c) 2011 X.commerce, a business unit of eBay Inc. 

2# Copyright 2011 OpenStack LLC. 

3# All Rights Reserved. 

4# 

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

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

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

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

14# License for the specific language governing permissions and limitations 

15# under the License. 

16 

17from unittest import mock 

18 

19import ddt 

20from oslo_config import cfg 

21from oslo_serialization import jsonutils 

22from oslo_utils import timeutils 

23import webob 

24 

25from manila.api import extensions 

26from manila.api.v1 import router 

27from manila import policy 

28from manila import test 

29 

30CONF = cfg.CONF 

31 

32 

33class ExtensionTestCase(test.TestCase): 

34 def setUp(self): 

35 super(ExtensionTestCase, self).setUp() 

36 ext_list = CONF.osapi_share_extension[:] 

37 fox = ('manila.tests.api.extensions.foxinsocks.Foxinsocks') 

38 if fox not in ext_list: 38 ↛ exitline 38 didn't return from function 'setUp' because the condition on line 38 was always true

39 ext_list.append(fox) 

40 self.flags(osapi_share_extension=ext_list) 

41 

42 

43class ExtensionControllerTest(ExtensionTestCase): 

44 

45 def setUp(self): 

46 super(ExtensionControllerTest, self).setUp() 

47 self.ext_list = [] 

48 self.ext_list.sort() 

49 

50 def test_list_extensions_json(self): 

51 app = router.APIRouter() 

52 request = webob.Request.blank("/fake/extensions") 

53 response = request.get_response(app) 

54 self.assertEqual(200, response.status_int) 

55 

56 # Make sure we have all the extensions, extra extensions being OK. 

57 data = jsonutils.loads(response.body) 

58 names = [str(x['name']) for x in data['extensions'] 

59 if str(x['name']) in self.ext_list] 

60 names.sort() 

61 self.assertEqual(self.ext_list, names) 

62 

63 # Ensure all the timestamps are valid according to iso8601 

64 for ext in data['extensions']: 

65 timeutils.parse_isotime(ext['updated']) 

66 

67 # Make sure that at least Fox in Sox is correct. 

68 (fox_ext, ) = [ 

69 x for x in data['extensions'] if x['alias'] == 'FOXNSOX'] 

70 self.assertEqual( 

71 {'name': 'Fox In Socks', 

72 'updated': '2011-01-22T13:25:27-06:00', 

73 'description': 'The Fox In Socks Extension.', 

74 'alias': 'FOXNSOX', 

75 'links': []}, 

76 fox_ext) 

77 

78 for ext in data['extensions']: 

79 url = '/fake/extensions/%s' % ext['alias'] 

80 request = webob.Request.blank(url) 

81 response = request.get_response(app) 

82 output = jsonutils.loads(response.body) 

83 self.assertEqual(ext['alias'], output['extension']['alias']) 

84 

85 def test_get_extension_json(self): 

86 app = router.APIRouter() 

87 request = webob.Request.blank("/fake/extensions/FOXNSOX") 

88 response = request.get_response(app) 

89 self.assertEqual(200, response.status_int) 

90 

91 data = jsonutils.loads(response.body) 

92 self.assertEqual( 

93 {"name": "Fox In Socks", 

94 "updated": "2011-01-22T13:25:27-06:00", 

95 "description": "The Fox In Socks Extension.", 

96 "alias": "FOXNSOX", 

97 "links": []}, 

98 data['extension']) 

99 

100 def test_get_non_existing_extension_json(self): 

101 app = router.APIRouter() 

102 request = webob.Request.blank("/fake/extensions/4") 

103 response = request.get_response(app) 

104 self.assertEqual(404, response.status_int) 

105 

106 

107@ddt.ddt 

108class ExtensionAuthorizeTestCase(test.TestCase): 

109 

110 @ddt.unpack 

111 @ddt.data({'action': 'fake', 'valid': 'api_extension:fake:fake'}, 

112 {'action': None, 'valid': 'api_extension:fake'}) 

113 def test_extension_authorizer(self, action, valid): 

114 self.mock_object(policy, 'enforce') 

115 target = 'fake' 

116 

117 extensions.extension_authorizer('api', 'fake')( 

118 {}, target, action) 

119 

120 policy.enforce.assert_called_once_with(mock.ANY, valid, target) 

121 

122 def test_extension_authorizer_empty_target(self): 

123 self.mock_object(policy, 'enforce') 

124 target = None 

125 context = mock.Mock() 

126 context.project_id = 'fake' 

127 context.user_id = 'fake' 

128 

129 extensions.extension_authorizer('api', 'fake')( 

130 context, target, 'fake') 

131 

132 policy.enforce.assert_called_once_with( 

133 mock.ANY, mock.ANY, {'project_id': 'fake', 'user_id': 'fake'}) 

134 

135 

136class StubExtensionManager(object): 

137 """Provides access to Tweedle Beetles.""" 

138 

139 name = "Tweedle Beetle Extension" 

140 alias = "TWDLBETL" 

141 

142 def __init__(self, resource_ext=None, action_ext=None, request_ext=None, 

143 controller_ext=None): 

144 self.resource_ext = resource_ext 

145 self.controller_ext = controller_ext 

146 self.extra_resource_ext = None 

147 

148 def get_resources(self): 

149 resource_exts = [] 

150 if self.resource_ext: 150 ↛ 152line 150 didn't jump to line 152 because the condition on line 150 was always true

151 resource_exts.append(self.resource_ext) 

152 if self.extra_resource_ext: 152 ↛ 153line 152 didn't jump to line 153 because the condition on line 152 was never true

153 resource_exts.append(self.extra_resource_ext) 

154 return resource_exts 

155 

156 def get_controller_extensions(self): 

157 controller_extensions = [] 

158 if self.controller_ext: 158 ↛ 159line 158 didn't jump to line 159 because the condition on line 158 was never true

159 controller_extensions.append(self.controller_ext) 

160 return controller_extensions 

161 

162 

163class ExtensionControllerIdFormatTest(test.TestCase): 

164 

165 def _bounce_id(self, test_id): 

166 

167 class BounceController(object): 

168 def show(self, req, id): 

169 return id 

170 res_ext = extensions.ResourceExtension('bounce', 

171 BounceController()) 

172 manager = StubExtensionManager(res_ext) 

173 app = router.APIRouter(manager) 

174 request = webob.Request.blank("/fake/bounce/%s" % test_id) 

175 response = request.get_response(app) 

176 return response.body 

177 

178 def test_id_with_xml_format(self): 

179 result = self._bounce_id('foo.xml') 

180 self.assertEqual('foo', result.decode('UTF-8')) 

181 

182 def test_id_with_json_format(self): 

183 result = self._bounce_id('foo.json') 

184 self.assertEqual('foo', result.decode('UTF-8')) 

185 

186 def test_id_with_bad_format(self): 

187 result = self._bounce_id('foo.bad') 

188 self.assertEqual('foo.bad', result.decode('UTF-8'))