test_decorator.py 10KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339
  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. # see MyProcessor#process
  193. assert {
  194. 'http_code': HTTPStatus.OK,
  195. 'original_response': '43',
  196. } == result
  197. def test_unit__output_data_wrapping__fail__error_response(self):
  198. context = MyContext(app=None)
  199. processor = MarshmallowOutputProcessor()
  200. processor.schema = MySchema()
  201. wrapper = OutputControllerWrapper(context, processor)
  202. @wrapper.get_wrapper
  203. def func(foo):
  204. return 'wrong result format'
  205. result = func(42)
  206. # see MyProcessor#process
  207. assert isinstance(result, dict)
  208. assert 'http_code' in result
  209. assert result['http_code'] == HTTPStatus.INTERNAL_SERVER_ERROR
  210. assert 'original_error' in result
  211. assert result['original_error'].details == {
  212. 'name': ['Missing data for required field.']
  213. }
  214. class TestExceptionHandlerControllerWrapper(Base):
  215. def test_unit__exception_handled__ok__nominal_case(self):
  216. context = MyContext(app=None)
  217. wrapper = ExceptionHandlerControllerWrapper(
  218. ZeroDivisionError,
  219. context,
  220. error_builder=DefaultErrorBuilder(),
  221. http_code=HTTPStatus.INTERNAL_SERVER_ERROR,
  222. )
  223. @wrapper.get_wrapper
  224. def func(foo):
  225. raise ZeroDivisionError('We are testing')
  226. response = func(42)
  227. assert 'http_code' in response
  228. assert response['http_code'] == HTTPStatus.INTERNAL_SERVER_ERROR
  229. assert 'original_response' in response
  230. assert json.loads(response['original_response']) == {
  231. 'message': 'We are testing',
  232. 'details': {},
  233. 'code': None,
  234. }
  235. def test_unit__exception_handled__ok__exception_error_dict(self):
  236. class MyException(Exception):
  237. def __init__(self, *args, **kwargs):
  238. super().__init__(*args, **kwargs)
  239. self.error_dict = {}
  240. context = MyContext(app=None)
  241. wrapper = ExceptionHandlerControllerWrapper(
  242. MyException,
  243. context,
  244. error_builder=DefaultErrorBuilder(),
  245. http_code=HTTPStatus.INTERNAL_SERVER_ERROR,
  246. )
  247. @wrapper.get_wrapper
  248. def func(foo):
  249. exc = MyException('We are testing')
  250. exc.error_detail = {'foo': 'bar'}
  251. raise exc
  252. response = func(42)
  253. assert 'http_code' in response
  254. assert response['http_code'] == HTTPStatus.INTERNAL_SERVER_ERROR
  255. assert 'original_response' in response
  256. assert json.loads(response['original_response']) == {
  257. 'message': 'We are testing',
  258. 'details': {'foo': 'bar'},
  259. 'code': None,
  260. }
  261. def test_unit__exception_handler__error__error_content_malformed(self):
  262. class MyException(Exception):
  263. pass
  264. class MyErrorBuilder(DefaultErrorBuilder):
  265. def build_from_exception(self, exception: Exception) -> dict:
  266. # this is not matching with DefaultErrorBuilder schema
  267. return {}
  268. context = MyContext(app=None)
  269. wrapper = ExceptionHandlerControllerWrapper(
  270. MyException,
  271. context,
  272. error_builder=MyErrorBuilder(),
  273. )
  274. def raise_it():
  275. raise MyException()
  276. wrapper = wrapper.get_wrapper(raise_it)
  277. with pytest.raises(OutputValidationException):
  278. wrapper()