sentry/tests/sentry_plugins/test_client.py

from unittest.mock import patch

import pytest
import responses

from sentry.services.hybrid_cloud.usersocialauth.serial import serialize_usersocialauth
from sentry.shared_integrations.exceptions import (
    ApiError,
    ApiHostError,
    ApiUnauthorized,
    UnsupportedResponseType,
)
from sentry.shared_integrations.response.base import BaseApiResponse
from sentry.testutils.cases import TestCase
from sentry_plugins.client import ApiClient, AuthApiClient


class ApiClientTest(TestCase):
    @responses.activate
    def test_get(self):
        responses.add(responses.GET, "http://example.com", json={})

        resp = ApiClient().get("http://example.com")
        assert isinstance(resp, BaseApiResponse)
        assert resp.status_code == 200

    @responses.activate
    def test_post(self):
        responses.add(responses.POST, "http://example.com", json={})

        resp = ApiClient().post("http://example.com")
        assert isinstance(resp, BaseApiResponse)
        assert resp.status_code == 200

    @responses.activate
    def test_delete(self):
        responses.add(responses.DELETE, "http://example.com", json={})

        resp = ApiClient().delete("http://example.com")
        assert isinstance(resp, BaseApiResponse)
        assert resp.status_code == 200

    @responses.activate
    def test_put(self):
        responses.add(responses.PUT, "http://example.com", json={})

        resp = ApiClient().put("http://example.com")
        assert isinstance(resp, BaseApiResponse)
        assert resp.status_code == 200

    @responses.activate
    def test_patch(self):
        responses.add(responses.PATCH, "http://example.com", json={})

        resp = ApiClient().patch("http://example.com")
        assert isinstance(resp, BaseApiResponse)
        assert resp.status_code == 200


class AuthApiClientTest(TestCase):
    @responses.activate
    def test_without_authorization(self):
        responses.add(responses.GET, "http://example.com", json={})

        resp = AuthApiClient().get("http://example.com")
        assert isinstance(resp, BaseApiResponse)
        assert resp.status_code == 200

        request = responses.calls[-1].request
        assert not request.headers.get("Authorization")

    @responses.activate
    def test_with_authorization(self):
        responses.add(responses.GET, "http://example.com", json={})

        auth = self.create_usersocialauth(extra_data={"access_token": "access-token"})
        rpc_auth = serialize_usersocialauth(auth=auth)

        resp = AuthApiClient(auth=rpc_auth).get("http://example.com")
        assert isinstance(resp, BaseApiResponse)
        assert resp.status_code == 200

        request = responses.calls[-1].request
        assert request.headers.get("Authorization") == "Bearer access-token"

    @responses.activate
    def test_with_authorization_and_no_auth(self):
        responses.add(responses.GET, "http://example.com", json={})

        auth = self.create_usersocialauth(extra_data={"access_token": "access-token"})
        rpc_auth = serialize_usersocialauth(auth=auth)
        resp = AuthApiClient(auth=rpc_auth).get("http://example.com", auth=None)

        assert isinstance(resp, BaseApiResponse)
        assert resp.status_code == 200

        request = responses.calls[-1].request
        assert not request.headers.get("Authorization")

    @responses.activate
    def test_with_authorized_token_refresh(self):
        # First attempt
        responses.add(responses.GET, "http://example.com", json={}, status=401)
        # After refresh
        responses.add(responses.GET, "http://example.com", json={}, status=200)

        auth = self.create_usersocialauth(extra_data={"access_token": "access-token"})
        rpc_auth = serialize_usersocialauth(auth=auth)

        with patch("social_auth.models.UserSocialAuth.refresh_token") as mock_refresh_token:
            resp = AuthApiClient(auth=rpc_auth).get("http://example.com")
            assert mock_refresh_token.called

        assert isinstance(resp, BaseApiResponse)
        assert resp.status_code == 200

        request = responses.calls[-1].request
        assert request.headers.get("Authorization") == "Bearer access-token"

    @responses.activate
    def test_invalid_host(self):
        with pytest.raises(ApiHostError):
            AuthApiClient().get("http://example.com")

    @responses.activate
    def test_unauthorized(self):
        responses.add(responses.GET, "http://example.com", status=404)
        with pytest.raises(ApiError):
            AuthApiClient().get("http://example.com")

    @responses.activate
    def test_forbidden(self):
        responses.add(responses.GET, "http://example.com", status=401)
        with pytest.raises(ApiUnauthorized):
            AuthApiClient().get("http://example.com")

    @responses.activate
    def test_invalid_plaintext(self):
        responses.add(responses.GET, "http://example.com", body="")
        with pytest.raises(UnsupportedResponseType):
            AuthApiClient().get("http://example.com")