test_client.py 5.0 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143
  1. from unittest.mock import patch
  2. import pytest
  3. import responses
  4. from sentry.services.hybrid_cloud.usersocialauth.serial import serialize_usersocialauth
  5. from sentry.shared_integrations.exceptions import (
  6. ApiError,
  7. ApiHostError,
  8. ApiUnauthorized,
  9. UnsupportedResponseType,
  10. )
  11. from sentry.shared_integrations.response.base import BaseApiResponse
  12. from sentry.testutils.cases import TestCase
  13. from sentry.testutils.silo import region_silo_test
  14. from sentry_plugins.client import ApiClient, AuthApiClient
  15. class ApiClientTest(TestCase):
  16. @responses.activate
  17. def test_get(self):
  18. responses.add(responses.GET, "http://example.com", json={})
  19. resp = ApiClient().get("http://example.com")
  20. assert isinstance(resp, BaseApiResponse)
  21. assert resp.status_code == 200
  22. @responses.activate
  23. def test_post(self):
  24. responses.add(responses.POST, "http://example.com", json={})
  25. resp = ApiClient().post("http://example.com")
  26. assert isinstance(resp, BaseApiResponse)
  27. assert resp.status_code == 200
  28. @responses.activate
  29. def test_delete(self):
  30. responses.add(responses.DELETE, "http://example.com", json={})
  31. resp = ApiClient().delete("http://example.com")
  32. assert isinstance(resp, BaseApiResponse)
  33. assert resp.status_code == 200
  34. @responses.activate
  35. def test_put(self):
  36. responses.add(responses.PUT, "http://example.com", json={})
  37. resp = ApiClient().put("http://example.com")
  38. assert isinstance(resp, BaseApiResponse)
  39. assert resp.status_code == 200
  40. @responses.activate
  41. def test_patch(self):
  42. responses.add(responses.PATCH, "http://example.com", json={})
  43. resp = ApiClient().patch("http://example.com")
  44. assert isinstance(resp, BaseApiResponse)
  45. assert resp.status_code == 200
  46. @region_silo_test(stable=True)
  47. class AuthApiClientTest(TestCase):
  48. @responses.activate
  49. def test_without_authorization(self):
  50. responses.add(responses.GET, "http://example.com", json={})
  51. resp = AuthApiClient().get("http://example.com")
  52. assert isinstance(resp, BaseApiResponse)
  53. assert resp.status_code == 200
  54. request = responses.calls[-1].request
  55. assert not request.headers.get("Authorization")
  56. @responses.activate
  57. def test_with_authorization(self):
  58. responses.add(responses.GET, "http://example.com", json={})
  59. auth = self.create_usersocialauth(extra_data={"access_token": "access-token"})
  60. rpc_auth = serialize_usersocialauth(auth=auth)
  61. resp = AuthApiClient(auth=rpc_auth).get("http://example.com")
  62. assert isinstance(resp, BaseApiResponse)
  63. assert resp.status_code == 200
  64. request = responses.calls[-1].request
  65. assert request.headers.get("Authorization") == "Bearer access-token"
  66. @responses.activate
  67. def test_with_authorization_and_no_auth(self):
  68. responses.add(responses.GET, "http://example.com", json={})
  69. auth = self.create_usersocialauth(extra_data={"access_token": "access-token"})
  70. rpc_auth = serialize_usersocialauth(auth=auth)
  71. resp = AuthApiClient(auth=rpc_auth).get("http://example.com", auth=None)
  72. assert isinstance(resp, BaseApiResponse)
  73. assert resp.status_code == 200
  74. request = responses.calls[-1].request
  75. assert not request.headers.get("Authorization")
  76. @responses.activate
  77. def test_with_authorized_token_refresh(self):
  78. # First attempt
  79. responses.add(responses.GET, "http://example.com", json={}, status=401)
  80. # After refresh
  81. responses.add(responses.GET, "http://example.com", json={}, status=200)
  82. auth = self.create_usersocialauth(extra_data={"access_token": "access-token"})
  83. rpc_auth = serialize_usersocialauth(auth=auth)
  84. with patch("social_auth.models.UserSocialAuth.refresh_token") as mock_refresh_token:
  85. resp = AuthApiClient(auth=rpc_auth).get("http://example.com")
  86. assert mock_refresh_token.called
  87. assert isinstance(resp, BaseApiResponse)
  88. assert resp.status_code == 200
  89. request = responses.calls[-1].request
  90. assert request.headers.get("Authorization") == "Bearer access-token"
  91. @responses.activate
  92. def test_invalid_host(self):
  93. with pytest.raises(ApiHostError):
  94. AuthApiClient().get("http://example.com")
  95. @responses.activate
  96. def test_unauthorized(self):
  97. responses.add(responses.GET, "http://example.com", status=404)
  98. with pytest.raises(ApiError):
  99. AuthApiClient().get("http://example.com")
  100. @responses.activate
  101. def test_forbidden(self):
  102. responses.add(responses.GET, "http://example.com", status=401)
  103. with pytest.raises(ApiUnauthorized):
  104. AuthApiClient().get("http://example.com")
  105. @responses.activate
  106. def test_invalid_plaintext(self):
  107. responses.add(responses.GET, "http://example.com", body="")
  108. with pytest.raises(UnsupportedResponseType):
  109. AuthApiClient().get("http://example.com")