Coverage for manila/tests/api/test_schemas.py: 68%

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

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

3# a copy of the License at 

4# 

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

6# 

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

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

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

10# License for the specific language governing permissions and limitations 

11# under the License. 

12 

13import jsonschema.exceptions 

14from oslo_log import log 

15 

16from manila.api.v2 import router 

17from manila.api.validation import validators 

18from manila import test 

19 

20LOG = log.getLogger(__name__) 

21 

22 

23class SchemaTest(test.TestCase): 

24 

25 def setUp(self): 

26 super().setUp() 

27 self.router = router.APIRouter() 

28 self.meta_schema = validators._SchemaValidator.validator_org 

29 

30 def test_schemas(self): 

31 missing_request_schemas = set() 

32 missing_query_schemas = set() 

33 missing_response_schemas = set() 

34 invalid_schemas = set() 

35 

36 def _validate_schema(func, schema): 

37 try: 

38 self.meta_schema.check_schema(schema) 

39 except jsonschema.exceptions.SchemaError: 

40 LOG.exception('schema validation failed') 

41 invalid_schemas.add(func.__qualname__) 

42 

43 def _validate_func(func, method): 

44 if getattr(func, 'removed', False): 44 ↛ 45line 44 didn't jump to line 45 because the condition on line 44 was never true

45 return 

46 

47 if method in ("POST", "PUT", "PATCH"): 

48 # request body validation 

49 if not hasattr(func, '_request_body_schema'): 49 ↛ 50line 49 didn't jump to line 50 because the condition on line 49 was never true

50 missing_request_schemas.add(func.__qualname__) 

51 else: 

52 _validate_schema(func, func._request_body_schema) 

53 elif method in ("GET",): 

54 # request query string validation 

55 if not hasattr(func, '_request_query_schema'): 55 ↛ 56line 55 didn't jump to line 56 because the condition on line 55 was never true

56 missing_query_schemas.add(func.__qualname__) 

57 else: 

58 _validate_schema(func, func._request_query_schema) 

59 

60 # response body validation 

61 if not hasattr(func, '_response_body_schema'): 61 ↛ 62line 61 didn't jump to line 62 because the condition on line 61 was never true

62 missing_response_schemas.add(func.__qualname__) 

63 else: 

64 _validate_schema(func, func._response_body_schema) 

65 

66 for route in self.router.map.matchlist: 

67 if 'controller' not in route.defaults: 

68 continue 

69 

70 controller = route.defaults['controller'] 

71 

72 if not getattr(controller.controller, '_validated', False): 

73 continue 

74 

75 # NOTE: This is effectively a reimplementation of 

76 # 'routes.route.Route.make_full_route' that uses OpenAPI-compatible 

77 # template strings instead of regexes for paramters 

78 path = "" 

79 for part in route.routelist: 

80 if isinstance(part, dict): 

81 path += "{" + part["name"] + "}" 

82 else: 

83 path += part 

84 

85 method = ( 

86 route.conditions.get("method", "GET")[0] 

87 if route.conditions 

88 else "GET" 

89 ) 

90 action = route.defaults["action"] 

91 

92 if path.endswith('/action'): 92 ↛ 94line 92 didn't jump to line 94 because the condition on line 92 was never true

93 # all actions should use POST 

94 assert method == 'POST' 

95 

96 wsgi_actions = [ 

97 (k, v, controller.controller) for k, v in 

98 controller.controller.wsgi_actions.items() 

99 ] 

100 

101 for ( 

102 wsgi_action, wsgi_method, action_controller 

103 ) in wsgi_actions: 

104 versioned_methods = getattr( 

105 action_controller, 'versioned_methods', {} 

106 ) 

107 if wsgi_method in versioned_methods: 

108 # versioned method 

109 for versioned_method in sorted( 

110 versioned_methods[action], 

111 key=lambda v: v.start_version 

112 ): 

113 func = versioned_method.func 

114 _validate_func(func, method) 

115 else: 

116 # unversioned method 

117 func = controller.wsgi_actions[wsgi_action] 

118 _validate_func(func, method) 

119 else: 

120 # body validation 

121 versioned_methods = getattr( 

122 controller.controller, 'versioned_methods', {} 

123 ) 

124 if action in versioned_methods: 

125 # versioned method 

126 for versioned_method in sorted( 

127 versioned_methods[action], 

128 key=lambda v: v.start_version 

129 ): 

130 func = versioned_method.func 

131 _validate_func(func, method) 

132 else: 

133 if not hasattr(controller.controller, action): 133 ↛ 139line 133 didn't jump to line 139 because the condition on line 133 was always true

134 # these are almost certainly because of use of 

135 # routes.mapper.Mapper.resource, which we should remove 

136 continue 

137 

138 # unversioned method 

139 func = getattr(controller.controller, action) 

140 _validate_func(func, method) 

141 

142 if missing_request_schemas: 142 ↛ 143line 142 didn't jump to line 143 because the condition on line 142 was never true

143 raise self.failureException( 

144 f"Found API resources without request body schemas: " 

145 f"{sorted(missing_request_schemas)}" 

146 ) 

147 

148 if missing_query_schemas: 148 ↛ 149line 148 didn't jump to line 149 because the condition on line 148 was never true

149 raise self.failureException( 

150 f"Found API resources without request query schemas: " 

151 f"{sorted(missing_query_schemas)}" 

152 ) 

153 

154 if missing_response_schemas: 154 ↛ 155line 154 didn't jump to line 155 because the condition on line 154 was never true

155 raise self.failureException( 

156 f"Found API resources without response body schemas: " 

157 f"{sorted(missing_response_schemas)}" 

158 ) 

159 

160 if invalid_schemas: 160 ↛ 161line 160 didn't jump to line 161 because the condition on line 160 was never true

161 raise self.failureException( 

162 f"Found API resources with invalid schemas: " 

163 f"{sorted(invalid_schemas)}" 

164 )