test_sync_api_provider.py 4.3 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140
  1. from http import HTTPStatus
  2. from unittest import mock
  3. import pytest
  4. from clean_python import ctx
  5. from clean_python import Tenant
  6. from clean_python.api_client import ApiException
  7. from clean_python.api_client import SyncApiProvider
  8. MODULE = "clean_python.api_client.api_provider"
  9. @pytest.fixture
  10. def tenant() -> Tenant:
  11. ctx.tenant = Tenant(id=2, name="")
  12. return ctx.tenant
  13. @pytest.fixture
  14. def response():
  15. response = mock.Mock()
  16. response.status = int(HTTPStatus.OK)
  17. response.headers = {"Content-Type": "application/json"}
  18. return response
  19. @pytest.fixture
  20. def api_provider(tenant, response) -> SyncApiProvider:
  21. with mock.patch(MODULE + ".PoolManager"):
  22. api_provider = SyncApiProvider(
  23. url="http://testserver/foo/",
  24. fetch_token=lambda a, b: f"tenant-{b}",
  25. )
  26. api_provider._pool.request.return_value = response
  27. yield api_provider
  28. def test_get(api_provider: SyncApiProvider, response):
  29. actual = api_provider.request("GET", "")
  30. assert api_provider._pool.request.call_count == 1
  31. assert api_provider._pool.request.call_args[1] == dict(
  32. method="GET",
  33. url="http://testserver/foo",
  34. json=None,
  35. fields=None,
  36. headers={"Authorization": "Bearer tenant-2"},
  37. timeout=5.0,
  38. )
  39. assert actual == response.json.return_value
  40. def test_post_json(api_provider: SyncApiProvider, response):
  41. response.status == int(HTTPStatus.CREATED)
  42. api_provider._pool.request.return_value = response
  43. actual = api_provider.request("POST", "bar", json={"foo": 2})
  44. assert api_provider._pool.request.call_count == 1
  45. assert api_provider._pool.request.call_args[1] == dict(
  46. method="POST",
  47. url="http://testserver/foo/bar",
  48. json={"foo": 2},
  49. fields=None,
  50. headers={"Authorization": "Bearer tenant-2"},
  51. timeout=5.0,
  52. )
  53. assert actual == response.json.return_value
  54. @pytest.mark.parametrize(
  55. "path,params,expected_url",
  56. [
  57. ("", None, "http://testserver/foo"),
  58. ("bar", None, "http://testserver/foo/bar"),
  59. ("bar/", None, "http://testserver/foo/bar"),
  60. ("", {"a": 2}, "http://testserver/foo?a=2"),
  61. ("bar", {"a": 2}, "http://testserver/foo/bar?a=2"),
  62. ("bar/", {"a": 2}, "http://testserver/foo/bar?a=2"),
  63. ("", {"a": [1, 2]}, "http://testserver/foo?a=1&a=2"),
  64. ("", {"a": 1, "b": "foo"}, "http://testserver/foo?a=1&b=foo"),
  65. ],
  66. )
  67. def test_url(api_provider: SyncApiProvider, path, params, expected_url):
  68. api_provider.request("GET", path, params=params)
  69. assert api_provider._pool.request.call_args[1]["url"] == expected_url
  70. def test_timeout(api_provider: SyncApiProvider):
  71. api_provider.request("POST", "bar", timeout=2.1)
  72. assert api_provider._pool.request.call_args[1]["timeout"] == 2.1
  73. @pytest.mark.parametrize(
  74. "status", [HTTPStatus.OK, HTTPStatus.NOT_FOUND, HTTPStatus.INTERNAL_SERVER_ERROR]
  75. )
  76. def test_unexpected_content_type(api_provider: SyncApiProvider, response, status):
  77. response.status = int(status)
  78. response.headers["Content-Type"] = "text/plain"
  79. with pytest.raises(ApiException) as e:
  80. api_provider.request("GET", "bar")
  81. assert e.value.status is status
  82. assert str(e.value) == "Unexpected content type 'text/plain'"
  83. def test_no_content(api_provider: SyncApiProvider, response):
  84. response.status = int(HTTPStatus.NO_CONTENT)
  85. response.headers = {}
  86. actual = api_provider.request("DELETE", "bar/2")
  87. assert actual is not None
  88. def test_404(api_provider: SyncApiProvider, response):
  89. response.status = int(HTTPStatus.NOT_FOUND)
  90. actual = api_provider.request("GET", "bar")
  91. assert actual is None
  92. @pytest.mark.parametrize("status", [HTTPStatus.BAD_REQUEST, HTTPStatus.IM_A_TEAPOT])
  93. def test_error_response(api_provider: SyncApiProvider, response, status):
  94. response.status = int(status)
  95. with pytest.raises(ApiException) as e:
  96. api_provider.request("GET", "bar")
  97. assert e.value.status is status
  98. assert str(e.value) == str(response.json())
  99. @mock.patch(MODULE + ".PoolManager", new=mock.Mock())
  100. def test_no_token(response):
  101. api_provider = SyncApiProvider(
  102. url="http://testserver/foo/", fetch_token=lambda a, b: None
  103. )
  104. api_provider._pool.request.return_value = response
  105. api_provider.request("GET", "")
  106. assert api_provider._pool.request.call_args[1]["headers"] == {}