1  from datetime import timedelta
 2  from unittest.mock import AsyncMock, MagicMock
 3  
 4  import pytest
 5  from _pytest.monkeypatch import MonkeyPatch
 6  from pytest_mock import MockerFixture
 7  
 8  from api.services.internal import InternalService, InternalServiceError
 9  from api.settings import settings
10  
11  
12  async def test__internal_service__get_token(mocker: MockerFixture, monkeypatch: MonkeyPatch) -> None:
13      encode_jwt = mocker.patch("api.services.internal.encode_jwt")
14      monkeypatch.setattr(settings, "internal_jwt_ttl", 123)
15      service = MagicMock()
16      service.name = "MY_SERVICE"
17  
18      result = InternalService._get_token(service)
19  
20      encode_jwt.assert_called_once_with({"aud": "my_service"}, timedelta(seconds=123))
21      assert result == encode_jwt()
22  
23  
24  @pytest.mark.parametrize(
25      "code,ok",
26      [(200, True), (201, True), (401, False), (403, False), (404, True), (500, False), (501, False), (502, False)],
27  )
28  async def test__internal_service__handle_error(code: int, ok: bool) -> None:
29      response = AsyncMock(status_code=code, text="response text asdf")
30  
31      if ok:
32          await InternalService._handle_error(response)
33          response.aread.assert_not_called()
34      else:
35          with pytest.raises(InternalServiceError) as e:
36              await InternalService._handle_error(response)
37          response.aread.assert_called_once_with()
38          assert e.value.args == (response, "response text asdf")
39  
40  
41  async def test__internal_service__client(mocker: MockerFixture, monkeypatch: MonkeyPatch) -> None:
42      async_client = mocker.patch("api.services.internal.AsyncClient")
43      service = MagicMock(value="http://example.service:1234/test/")
44  
45      result = InternalService.client.fget(service)  # type: ignore
46  
47      async_client.assert_called_once()
48      args = async_client.call_args[1]
49      assert result == async_client()
50  
51      assert args["base_url"] == "http://example.service:1234/test/_internal"
52      assert args["headers"] == {"Authorization": service._get_token()}
53  
54      event_hooks = args["event_hooks"]
55      assert [*event_hooks] == ["response"]
56      assert event_hooks["response"] == [service._handle_error]