Coverage for manila/tests/api/middleware/test_faults.py: 97%

1# Copyright 2010 OpenStack LLC. 

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 oslo_serialization import jsonutils 

17import webob 

18import webob.dec 

19import webob.exc 

20 

21from manila.api.middleware import fault 

22from manila.api.openstack import wsgi 

23from manila import exception 

24from manila import test 

25 

26 

27class TestFaults(test.TestCase): 

28 """Tests covering `manila.api.openstack.faults:Fault` class.""" 

29 

30 def _prepare_xml(self, xml_string): 

31 """Remove characters from string which hinder XML equality testing.""" 

32 xml_string = xml_string.replace(" ", "") 

33 xml_string = xml_string.replace("\n", "") 

34 xml_string = xml_string.replace("\t", "") 

35 return xml_string 

36 

37 def test_400_fault_json(self): 

38 """Test fault serialized to JSON via file-extension and/or header.""" 

39 requests = [ 

40 webob.Request.blank('/.json'), 

41 webob.Request.blank('/', headers={"Accept": "application/json"}), 

42 ] 

43 

44 for request in requests: 

45 fault = wsgi.Fault(webob.exc.HTTPBadRequest(explanation='scram')) 

46 response = request.get_response(fault) 

47 

48 expected = { 

49 "badRequest": { 

50 "message": "scram", 

51 "code": 400, 

52 }, 

53 } 

54 actual = jsonutils.loads(response.body) 

55 

56 self.assertEqual("application/json", response.content_type) 

57 self.assertEqual(expected, actual) 

58 

59 def test_413_fault_json(self): 

60 """Test fault serialized to JSON via file-extension and/or header.""" 

61 requests = [ 

62 webob.Request.blank('/.json'), 

63 webob.Request.blank('/', headers={"Accept": "application/json"}), 

64 ] 

65 

66 for request in requests: 

67 exc = webob.exc.HTTPRequestEntityTooLarge 

68 fault = wsgi.Fault(exc(explanation='sorry', 

69 headers={'Retry-After': 4})) 

70 response = request.get_response(fault) 

71 

72 expected = { 

73 "overLimit": { 

74 "message": "sorry", 

75 "code": 413, 

76 "retryAfter": '4', 

77 }, 

78 } 

79 actual = jsonutils.loads(response.body) 

80 

81 self.assertEqual("application/json", response.content_type) 

82 self.assertEqual(expected, actual) 

83 

84 def test_raise(self): 

85 """Ensure the ability to raise :class:`Fault` in WSGI-ified methods.""" 

86 @webob.dec.wsgify 

87 def raiser(req): 

88 raise wsgi.Fault(webob.exc.HTTPNotFound(explanation='whut?')) 

89 

90 req = webob.Request.blank('/.json') 

91 resp = req.get_response(raiser) 

92 self.assertEqual("application/json", resp.content_type) 

93 self.assertEqual(404, resp.status_int) 

94 self.assertIn('whut?'.encode("utf-8"), resp.body) 

95 

96 def test_raise_403(self): 

97 """Ensure the ability to raise :class:`Fault` in WSGI-ified methods.""" 

98 @webob.dec.wsgify 

99 def raiser(req): 

100 raise wsgi.Fault(webob.exc.HTTPForbidden(explanation='whut?')) 

101 

102 req = webob.Request.blank('/.json') 

103 resp = req.get_response(raiser) 

104 self.assertEqual("application/json", resp.content_type) 

105 self.assertEqual(403, resp.status_int) 

106 self.assertNotIn('resizeNotAllowed'.encode("utf-8"), resp.body) 

107 self.assertIn('forbidden'.encode("utf-8"), resp.body) 

108 

109 def test_fault_has_status_int(self): 

110 """Ensure the status_int is set correctly on faults.""" 

111 fault = wsgi.Fault(webob.exc.HTTPBadRequest(explanation='what?')) 

112 self.assertEqual(400, fault.status_int) 

113 

114 

115class ExceptionTest(test.TestCase): 

116 

117 def _wsgi_app(self, inner_app): 

118 return fault.FaultWrapper(inner_app) 

119 

120 def _do_test_exception_safety_reflected_in_faults(self, expose): 

121 class ExceptionWithSafety(exception.ManilaException): 

122 safe = expose 

123 

124 @webob.dec.wsgify 

125 def fail(req): 

126 raise ExceptionWithSafety('some explanation') 

127 

128 api = self._wsgi_app(fail) 

129 resp = webob.Request.blank('/').get_response(api) 

130 self.assertIn('{"computeFault', str(resp.body), resp.body) 

131 expected = ('ExceptionWithSafety: some explanation' if expose else 

132 'The server has either erred or is incapable ' 

133 'of performing the requested operation.') 

134 self.assertIn(expected, str(resp.body), resp.body) 

135 self.assertEqual(500, resp.status_int, resp.body) 

136 

137 def test_safe_exceptions_are_described_in_faults(self): 

138 self._do_test_exception_safety_reflected_in_faults(True) 

139 

140 def test_unsafe_exceptions_are_not_described_in_faults(self): 

141 self._do_test_exception_safety_reflected_in_faults(False) 

142 

143 def _do_test_exception_mapping(self, exception_type, msg): 

144 @webob.dec.wsgify 

145 def fail(req): 

146 raise exception_type(msg) 

147 

148 api = self._wsgi_app(fail) 

149 resp = webob.Request.blank('/').get_response(api) 

150 self.assertIn(msg, str(resp.body), resp.body) 

151 self.assertEqual(exception_type.code, resp.status_int, resp.body) 

152 

153 if hasattr(exception_type, 'headers'): 

154 for (key, value) in exception_type.headers.items(): 

155 self.assertIn(key, resp.headers) 

156 self.assertEqual(value, resp.headers[key]) 

157 

158 def test_quota_error_mapping(self): 

159 self._do_test_exception_mapping(exception.QuotaError, 'too many used') 

160 

161 def test_non_manila_notfound_exception_mapping(self): 

162 class ExceptionWithCode(Exception): 

163 code = 404 

164 

165 self._do_test_exception_mapping(ExceptionWithCode, 

166 'NotFound') 

167 

168 def test_non_manila_exception_mapping(self): 

169 class ExceptionWithCode(Exception): 

170 code = 417 

171 

172 self._do_test_exception_mapping(ExceptionWithCode, 

173 'Expectation failed') 

174 

175 def test_exception_with_none_code_throws_500(self): 

176 class ExceptionWithNoneCode(Exception): 

177 code = None 

178 

179 @webob.dec.wsgify 

180 def fail(req): 

181 raise ExceptionWithNoneCode() 

182 

183 api = self._wsgi_app(fail) 

184 resp = webob.Request.blank('/').get_response(api) 

185 self.assertEqual(500, resp.status_int) 

186 

187 def test_validate_request_unicode_decode_fault(self): 

188 @webob.dec.wsgify 

189 def unicode_error(req): 

190 raise UnicodeDecodeError("ascii", "test".encode(), 0, 1, "bad") 

191 

192 api = self._wsgi_app(unicode_error) 

193 resp = webob.Request.blank('/test?foo=%88').get_response(api) 

194 self.assertEqual(400, resp.status_int)