test_decorator.py 10KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334
  1. # -*- coding: utf-8 -*-
  2. import json
  3. import pytest
  4. import typing
  5. from hapic.exception import OutputValidationException
  6. try: # Python 3.5+
  7. from http import HTTPStatus
  8. except ImportError:
  9. from http import client as HTTPStatus
  10. import marshmallow
  11. from multidict import MultiDict
  12. from hapic.data import HapicData
  13. from hapic.decorator import ExceptionHandlerControllerWrapper
  14. from hapic.decorator import InputQueryControllerWrapper
  15. from hapic.decorator import InputControllerWrapper
  16. from hapic.decorator import InputOutputControllerWrapper
  17. from hapic.decorator import OutputControllerWrapper
  18. from hapic.error import DefaultErrorBuilder
  19. from hapic.processor import MarshmallowOutputProcessor
  20. from hapic.processor import ProcessValidationError
  21. from hapic.processor import ProcessorInterface
  22. from hapic.processor import RequestParameters
  23. from tests.base import Base
  24. from tests.base import MyContext
  25. class MyProcessor(ProcessorInterface):
  26. def process(self, value):
  27. return value + 1
  28. def get_validation_error(
  29. self,
  30. request_context: RequestParameters,
  31. ) -> ProcessValidationError:
  32. return ProcessValidationError(
  33. details={
  34. 'original_request_context': request_context,
  35. },
  36. message='ERROR',
  37. )
  38. class MySimpleProcessor(ProcessorInterface):
  39. def process(self, value):
  40. return value
  41. def get_validation_error(
  42. self,
  43. request_context: RequestParameters,
  44. ) -> ProcessValidationError:
  45. return ProcessValidationError(
  46. details={
  47. 'original_request_context': request_context,
  48. },
  49. message='ERROR',
  50. )
  51. class MyControllerWrapper(InputOutputControllerWrapper):
  52. def before_wrapped_func(
  53. self,
  54. func_args: typing.Tuple[typing.Any, ...],
  55. func_kwargs: typing.Dict[str, typing.Any],
  56. ) -> typing.Union[None, typing.Any]:
  57. if func_args and func_args[0] == 666:
  58. return {
  59. 'error_response': 'we are testing'
  60. }
  61. func_kwargs['added_parameter'] = 'a value'
  62. def after_wrapped_function(self, response: typing.Any) -> typing.Any:
  63. return response * 2
  64. class MyInputQueryControllerWrapper(InputControllerWrapper):
  65. def get_processed_data(
  66. self,
  67. request_parameters: RequestParameters,
  68. ) -> typing.Any:
  69. return request_parameters.query_parameters
  70. def update_hapic_data(
  71. self,
  72. hapic_data: HapicData,
  73. processed_data: typing.Dict[str, typing.Any],
  74. ) -> typing.Any:
  75. hapic_data.query = processed_data
  76. class MySchema(marshmallow.Schema):
  77. name = marshmallow.fields.String(required=True)
  78. class TestControllerWrapper(Base):
  79. def test_unit__base_controller_wrapper__ok__no_behaviour(self):
  80. context = MyContext(app=None)
  81. processor = MyProcessor()
  82. wrapper = InputOutputControllerWrapper(context, processor)
  83. @wrapper.get_wrapper
  84. def func(foo):
  85. return foo
  86. result = func(42)
  87. assert result == 42
  88. def test_unit__base_controller__ok__replaced_response(self):
  89. context = MyContext(app=None)
  90. processor = MyProcessor()
  91. wrapper = MyControllerWrapper(context, processor)
  92. @wrapper.get_wrapper
  93. def func(foo):
  94. return foo
  95. # see MyControllerWrapper#before_wrapped_func
  96. result = func(666)
  97. # result have been replaced by MyControllerWrapper#before_wrapped_func
  98. assert {'error_response': 'we are testing'} == result
  99. def test_unit__controller_wrapper__ok__overload_input(self):
  100. context = MyContext(app=None)
  101. processor = MyProcessor()
  102. wrapper = MyControllerWrapper(context, processor)
  103. @wrapper.get_wrapper
  104. def func(foo, added_parameter=None):
  105. # see MyControllerWrapper#before_wrapped_func
  106. assert added_parameter == 'a value'
  107. return foo
  108. result = func(42)
  109. # See MyControllerWrapper#after_wrapped_function
  110. assert result == 84
  111. class TestInputControllerWrapper(Base):
  112. def test_unit__input_data_wrapping__ok__nominal_case(self):
  113. context = MyContext(
  114. app=None,
  115. fake_query_parameters=MultiDict(
  116. (
  117. ('foo', 'bar',),
  118. )
  119. )
  120. )
  121. processor = MyProcessor()
  122. wrapper = MyInputQueryControllerWrapper(context, processor)
  123. @wrapper.get_wrapper
  124. def func(foo, hapic_data=None):
  125. assert hapic_data
  126. assert isinstance(hapic_data, HapicData)
  127. # see MyControllerWrapper#before_wrapped_func
  128. assert hapic_data.query == {'foo': 'bar'}
  129. return foo
  130. result = func(42)
  131. assert result == 42
  132. def test_unit__multi_query_param_values__ok__use_as_list(self):
  133. context = MyContext(
  134. app=None,
  135. fake_query_parameters=MultiDict(
  136. (
  137. ('user_id', 'abc'),
  138. ('user_id', 'def'),
  139. ),
  140. )
  141. )
  142. processor = MySimpleProcessor()
  143. wrapper = InputQueryControllerWrapper(
  144. context,
  145. processor,
  146. as_list=['user_id'],
  147. )
  148. @wrapper.get_wrapper
  149. def func(hapic_data=None):
  150. assert hapic_data
  151. assert isinstance(hapic_data, HapicData)
  152. # see MyControllerWrapper#before_wrapped_func
  153. assert ['abc', 'def'] == hapic_data.query.get('user_id')
  154. return hapic_data.query.get('user_id')
  155. result = func()
  156. assert result == ['abc', 'def']
  157. def test_unit__multi_query_param_values__ok__without_as_list(self):
  158. context = MyContext(
  159. app=None,
  160. fake_query_parameters=MultiDict(
  161. (
  162. ('user_id', 'abc'),
  163. ('user_id', 'def'),
  164. ),
  165. )
  166. )
  167. processor = MySimpleProcessor()
  168. wrapper = InputQueryControllerWrapper(
  169. context,
  170. processor,
  171. )
  172. @wrapper.get_wrapper
  173. def func(hapic_data=None):
  174. assert hapic_data
  175. assert isinstance(hapic_data, HapicData)
  176. # see MyControllerWrapper#before_wrapped_func
  177. assert 'abc' == hapic_data.query.get('user_id')
  178. return hapic_data.query.get('user_id')
  179. result = func()
  180. assert result == 'abc'
  181. class TestOutputControllerWrapper(Base):
  182. def test_unit__output_data_wrapping__ok__nominal_case(self):
  183. context = MyContext(app=None)
  184. processor = MyProcessor()
  185. wrapper = OutputControllerWrapper(context, processor)
  186. @wrapper.get_wrapper
  187. def func(foo, hapic_data=None):
  188. # If no use of input wrapper, no hapic_data is given
  189. assert not hapic_data
  190. return foo
  191. result = func(42)
  192. assert HTTPStatus.OK == result.status_code
  193. assert '43' == result.body
  194. def test_unit__output_data_wrapping__fail__error_response(self):
  195. context = MyContext(app=None)
  196. processor = MarshmallowOutputProcessor()
  197. processor.schema = MySchema()
  198. wrapper = OutputControllerWrapper(context, processor)
  199. @wrapper.get_wrapper
  200. def func(foo):
  201. return 'wrong result format'
  202. result = func(42)
  203. assert HTTPStatus.INTERNAL_SERVER_ERROR == result.status_code
  204. assert {
  205. 'original_error': {
  206. 'details': {
  207. 'name': ['Missing data for required field.']
  208. },
  209. 'message': 'Validation error of output data'
  210. },
  211. 'http_code': 500,
  212. } == json.loads(result.body)
  213. class TestExceptionHandlerControllerWrapper(Base):
  214. def test_unit__exception_handled__ok__nominal_case(self):
  215. context = MyContext(app=None)
  216. wrapper = ExceptionHandlerControllerWrapper(
  217. ZeroDivisionError,
  218. context,
  219. error_builder=DefaultErrorBuilder(),
  220. http_code=HTTPStatus.INTERNAL_SERVER_ERROR,
  221. )
  222. @wrapper.get_wrapper
  223. def func(foo):
  224. raise ZeroDivisionError('We are testing')
  225. response = func(42)
  226. assert HTTPStatus.INTERNAL_SERVER_ERROR == response.status_code
  227. assert {
  228. 'details': {},
  229. 'message': 'We are testing',
  230. 'code': None,
  231. } == json.loads(response.body)
  232. def test_unit__exception_handled__ok__exception_error_dict(self):
  233. class MyException(Exception):
  234. def __init__(self, *args, **kwargs):
  235. super().__init__(*args, **kwargs)
  236. self.error_dict = {}
  237. context = MyContext(app=None)
  238. wrapper = ExceptionHandlerControllerWrapper(
  239. MyException,
  240. context,
  241. error_builder=DefaultErrorBuilder(),
  242. http_code=HTTPStatus.INTERNAL_SERVER_ERROR,
  243. )
  244. @wrapper.get_wrapper
  245. def func(foo):
  246. exc = MyException('We are testing')
  247. exc.error_detail = {'foo': 'bar'}
  248. raise exc
  249. response = func(42)
  250. assert response.status_code == HTTPStatus.INTERNAL_SERVER_ERROR
  251. assert {
  252. 'message': 'We are testing',
  253. 'details': {'foo': 'bar'},
  254. 'code': None,
  255. } == json.loads(response.body)
  256. def test_unit__exception_handler__error__error_content_malformed(self):
  257. class MyException(Exception):
  258. pass
  259. class MyErrorBuilder(DefaultErrorBuilder):
  260. def build_from_exception(self, exception: Exception) -> dict:
  261. # this is not matching with DefaultErrorBuilder schema
  262. return {}
  263. context = MyContext(app=None)
  264. wrapper = ExceptionHandlerControllerWrapper(
  265. MyException,
  266. context,
  267. error_builder=MyErrorBuilder(),
  268. )
  269. def raise_it():
  270. raise MyException()
  271. wrapper = wrapper.get_wrapper(raise_it)
  272. with pytest.raises(OutputValidationException):
  273. wrapper()