1  import json
 2  from typing import Any, AsyncIterator
 3  from unittest.mock import AsyncMock, MagicMock
 4  
 5  import pytest
 6  from pydantic import create_model
 7  from pytest_mock import MockerFixture
 8  
 9  from api.utils.debug import _check_response_schema, check_responses
10  
11  
12  @pytest.mark.parametrize(
13      "responses,in_schema,status_code,body,ok",
14      [
15          ({}, True, 422, {}, True),
16          ({}, True, 405, {}, True),
17          ({}, False, 200, {"foo": "bar"}, True),
18          ({}, True, 200, {"foo": "bar"}, False),
19          ({200: {"model": create_model("test", foo=(str, ...))}}, True, 200, {}, False),
20          ({200: {"model": create_model("test", foo=(int, ...))}}, True, 200, {"foo": "bar"}, False),
21          ({200: {"model": create_model("test", foo=(str, ...))}}, True, 200, {"foo": "bar"}, True),
22          ({404: {"content": {"application/json": {"examples": {}}}}}, True, 404, {"detail": "Not Found"}, False),
23          (
24              {404: {"content": {"application/json": {"examples": {"1": {"value": {"detail": "Not Found"}}}}}}},
25              True,
26              404,
27              {"detail": "blubb"},
28              False,
29          ),
30          (
31              {404: {"content": {"application/json": {"examples": {"1": {"value": {"detail": "Not Found"}}}}}}},
32              True,
33              404,
34              {"detail": "Not Found"},
35              True,
36          ),
37      ],
38  )
39  async def test___check_response_schema(
40      responses: Any, in_schema: bool, status_code: int, body: Any, ok: bool, mocker: MockerFixture
41  ) -> None:
42      logger = mocker.patch("api.utils.debug.logger")
43  
44      _check_response_schema(
45          "GET", MagicMock(responses=responses, include_in_schema=in_schema), status_code, json.dumps(body).encode()
46      )
47  
48      if ok:
49          logger.error.assert_not_called()
50      else:
51          logger.error.assert_called_once()
52  
53  
54  @pytest.mark.parametrize("json,has_route", [(False, True), (True, False), (True, True)])
55  async def test__check_responses(json: bool, has_route: bool, mocker: MockerFixture) -> None:
56      route = MagicMock() if has_route else None
57      request = MagicMock(scope={"route": route})
58      response = MagicMock(headers={"Content-type": "application/json" if json else MagicMock()})
59      call_next = AsyncMock(return_value=response)
60      check_response_schema = mocker.patch("api.utils.debug._check_response_schema")
61  
62      async def body_iterator() -> AsyncIterator[bytes]:
63          yield b"foo"
64          yield b"bar"
65          yield b"12345"
66          yield b""
67  
68      response.body_iterator = body_iterator()
69  
70      result = await check_responses(request, call_next)
71  
72      assert result is response
73      assert [chunk async for chunk in result.body_iterator] == [b"foo", b"bar", b"12345", b""]
74  
75      call_next.assert_called_once_with(request)
76      if json and has_route:
77          check_response_schema.assert_called_once_with(request.method, route, response.status_code, b"foobar12345")
78      else:
79          check_response_schema.assert_not_called()