test_proxy.py 2.6 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172
  1. from __future__ import annotations
  2. from dataclasses import asdict
  3. import pytest
  4. from django.urls import reverse
  5. from pytest_django.live_server_helper import LiveServer
  6. from rest_framework.test import APIClient
  7. from sentry.models.apikey import ApiKey
  8. from sentry.models.organization import Organization
  9. from sentry.models.team import Team
  10. from sentry.silo import SiloMode
  11. from sentry.testutils.asserts import assert_status_code
  12. from sentry.testutils.cases import TransactionTestCase
  13. from sentry.testutils.factories import Factories
  14. from sentry.testutils.region import override_regions
  15. from sentry.testutils.silo import region_silo_test
  16. from sentry.types.region import Region
  17. from sentry.utils import json
  18. from tests.sentry.middleware.test_proxy import test_region
  19. @pytest.fixture(scope="function")
  20. def local_live_server(request, live_server):
  21. if hasattr(request, "cls"):
  22. request.cls.live_server = live_server
  23. request.node.live_server = live_server
  24. @region_silo_test(regions=[test_region])
  25. @pytest.mark.usefixtures("local_live_server")
  26. class EndToEndAPIProxyTest(TransactionTestCase):
  27. live_server: LiveServer
  28. endpoint = "sentry-api-0-organization-teams"
  29. method = "post"
  30. organization: Organization
  31. api_key: ApiKey
  32. def get_response(self, *args, **params):
  33. url = reverse(self.endpoint, args=args)
  34. headers = params.pop("extra_headers", {})
  35. return getattr(self.client, self.method)(url, format="json", data=params, **headers)
  36. def test_through_api_gateway(self):
  37. if SiloMode.get_current_mode() == SiloMode.MONOLITH:
  38. return
  39. self.client = APIClient()
  40. config = asdict(test_region)
  41. config["address"] = self.live_server.url
  42. with override_regions([Region(**config)]):
  43. self.organization = Factories.create_organization(owner=self.user, region="us")
  44. self.api_key = Factories.create_api_key(
  45. organization=self.organization, scope_list=["org:write", "org:admin", "team:write"]
  46. )
  47. with SiloMode.enter_single_process_silo_context(SiloMode.CONTROL):
  48. resp = self.get_response(
  49. self.organization.slug,
  50. name="hello world",
  51. idp_provisioned=True,
  52. extra_headers=dict(
  53. HTTP_AUTHORIZATION=self.create_basic_auth_header(self.api_key.key)
  54. ),
  55. )
  56. assert_status_code(resp, 201)
  57. result = json.loads(resp.getvalue())
  58. team = Team.objects.get(id=result["id"])
  59. assert team.idp_provisioned