from __future__ import annotations from typing import Any from urllib.parse import parse_qs, urlencode, urlparse import pytest import responses from sentry.integrations.vsts import VstsIntegrationProvider from sentry.silo.base import SiloMode from sentry.testutils.cases import IntegrationTestCase from sentry.testutils.silo import assume_test_silo_mode class VstsIntegrationTestCase(IntegrationTestCase): provider = VstsIntegrationProvider() @pytest.fixture(autouse=True) def setup_data(self): self.access_token = "9d646e20-7a62-4bcc-abc0-cb2d4d075e36" self.refresh_token = "32004633-a3c0-4616-9aa0-a40632adac77" self.vsts_account_id = "c8a585ae-b61f-4ba6-833c-9e8d5d1674d8" self.vsts_account_name = "MyVSTSAccount" self.vsts_account_uri = "https://MyVSTSAccount.vssps.visualstudio.com:443/" self.vsts_base_url = "https://MyVSTSAccount.visualstudio.com/" self.vsts_user_id = "d6245f20-2af8-44f4-9451-8107cb2767db" self.vsts_user_name = "Foo Bar" self.vsts_user_email = "foobar@example.com" self.repo_id = "47166099-3e16-4868-9137-22ac6b05b06e" self.repo_name = "cool-service" self.project_a = {"id": "eb6e4656-77fc-42a1-9181-4c6d8e9da5d1", "name": "ProjectA"} self.project_b = {"id": "6ce954b1-ce1f-45d1-b94d-e6bf2464ba2c", "name": "ProjectB"} with responses.mock: self._stub_vsts() yield def _stub_vsts(self): responses.reset() responses.add( responses.POST, "https://app.vssps.visualstudio.com/oauth2/token", json={ "access_token": self.access_token, "token_type": "grant", "expires_in": 300, # seconds (5 min) "refresh_token": self.refresh_token, }, ) responses.add( responses.GET, "https://app.vssps.visualstudio.com/_apis/accounts?memberId=%s&api-version=4.1" % self.vsts_user_id, json={ "count": 1, "value": [ { "accountId": self.vsts_account_id, "accountUri": self.vsts_account_uri, "accountName": self.vsts_account_name, "properties": {}, } ], }, ) responses.add( responses.GET, "https://app.vssps.visualstudio.com/_apis/resourceareas/79134C72-4A58-4B42-976C-04E7115F32BF?hostId=%s&api-preview=5.0-preview.1" % self.vsts_account_id, json={"locationUrl": self.vsts_base_url}, ) responses.add( responses.GET, "https://app.vssps.visualstudio.com/_apis/profile/profiles/me?api-version=1.0", json={ "id": self.vsts_user_id, "displayName": self.vsts_user_name, "emailAddress": self.vsts_user_email, }, ) responses.add( responses.GET, "https://app.vssps.visualstudio.com/_apis/connectionData/", json={"authenticatedUser": {"subjectDescriptor": self.vsts_account_id}}, ) responses.add( responses.GET, f"https://{self.vsts_account_name.lower()}.visualstudio.com/_apis/projects", json={"value": [self.project_a, self.project_b], "count": 2}, ) responses.add( responses.POST, f"https://{self.vsts_account_name.lower()}.visualstudio.com/_apis/hooks/subscriptions", json=CREATE_SUBSCRIPTION, ) responses.add( responses.GET, f"https://{self.vsts_account_name.lower()}.visualstudio.com/_apis/git/repositories", json={ "value": [ { "id": self.repo_id, "name": self.repo_name, "project": {"name": self.project_a["name"]}, } ] }, ) responses.add( responses.GET, f"https://{self.vsts_account_name.lower()}.visualstudio.com/ProjectA/_apis/git/repositories/ProjectA", json={ "repository": { "id": self.repo_id, "name": self.repo_name, "project": {"name": self.project_a["name"]}, } }, ) for project in [self.project_a, self.project_b]: responses.add( responses.GET, "https://{}.visualstudio.com/{}/_apis/wit/workitemtypes/{}/states".format( self.vsts_account_name.lower(), project["id"], "Bug" ), json={ "count": 6, "value": [ {"name": "resolve_status"}, {"name": "resolve_when"}, {"name": "regression_status"}, {"name": "sync_comments"}, {"name": "sync_forward_assignment"}, {"name": "sync_reverse_assignment"}, ], }, ) responses.add( responses.GET, "https://{}.visualstudio.com/{}/_apis/wit/workitemtypes/{}/states".format( self.vsts_account_name.lower(), project["id"], "Issue" ), json={ "count": 0, "value": [], }, ) responses.add( responses.GET, "https://{}.visualstudio.com/{}/_apis/wit/workitemtypes/{}/states".format( self.vsts_account_name.lower(), project["id"], "Task" ), json={ "count": 0, "value": [], }, ) def make_init_request(self, path=None, body=None): return self.client.get(path or self.init_path, body or {}) def make_oauth_redirect_request(self, state): return self.client.get( "{}?{}".format(self.setup_path, urlencode({"code": "oauth-code", "state": state})) ) def assert_vsts_oauth_redirect(self, redirect): assert redirect.scheme == "https" assert redirect.netloc == "app.vssps.visualstudio.com" assert redirect.path == "/oauth2/authorize" def assert_account_selection(self, response, account_id=None): account_id = account_id or self.vsts_account_id assert response.status_code == 200 assert f'