feat(vsts): Add API-driven integration setup

Add API-driven setup steps for the Azure DevOps (VSTS) integration so it
can use the pipeline API instead of the legacy server-rendered flow.

Use the shared OAuth API step with the VSTS new identity provider and
add a dedicated account selection step for choosing the Azure DevOps
organization.

Refs [VDY-43: Azure DevOps (VSTS): API-driven integration setup](https://linear.app/getsentry/issue/VDY-43/azure-devops-vsts-api-driven-integration-setup)

Author: evanpurkhiser

src/sentry/integrations/vsts/integration.py

@@ -11,13 +11,16 @@
 from django.http.request import HttpRequest
 from django.http.response import HttpResponseBase
 from django.utils.translation import gettext as _
+from rest_framework.fields import CharField
+from rest_framework.serializers import Serializer
 
-from sentry import features, http
+from sentry import features, http, options
 from sentry.auth.exceptions import IdentityNotValid
 from sentry.constants import ObjectStatus
+from sentry.identity.oauth2 import OAuth2ApiStep
 from sentry.identity.pipeline import IdentityPipeline
 from sentry.identity.services.identity.model import RpcIdentity
-from sentry.identity.vsts.provider import get_user_info
+from sentry.identity.vsts.provider import VSTSNewIdentityProvider, get_user_info
 from sentry.integrations.base import (
     FeatureDescription,
     IntegrationData,
@@ -47,7 +50,8 @@
 from sentry.models.apitoken import generate_token
 from sentry.models.repository import Repository
 from sentry.organizations.services.organization.model import RpcOrganization
-from sentry.pipeline.views.base import PipelineView
+from sentry.pipeline.types import PipelineStepResult
+from sentry.pipeline.views.base import ApiPipelineSteps, PipelineView
 from sentry.pipeline.views.nested import NestedPipelineView
 from sentry.shared_integrations.exceptions import (
     ApiError,
@@ -135,6 +139,30 @@
 logger = logging.getLogger("sentry.integrations")
 
 
+def get_account_from_id(
+    account_id: str, accounts: Sequence[Mapping[str, Any]]
+) -> Mapping[str, Any] | None:
+    for account in accounts:
+        if account["accountId"] == account_id:
+            return account
+    return None
+
+
+def get_accounts(access_token: str, user_id: str) -> Any | None:
+    url = f"https://app.vssps.visualstudio.com/_apis/accounts?memberId={user_id}&api-version=4.1"
+    with http.build_session() as session:
+        response = session.get(
+            url,
+            headers={
+                "Content-Type": "application/json",
+                "Authorization": f"Bearer {access_token}",
+            },
+        )
+    if response.status_code == 200:
+        return response.json()
+    return None
+
+
 class VstsIntegration(RepositoryIntegration[VstsApiClient], VstsIssuesSpec):
     logger = logger
     comment_key = "sync_comments"
@@ -414,6 +442,63 @@ def default_project(self) -> str | None:
             return None
 
 
+class VstsAccountSelectionSerializer(Serializer):
+    account = CharField(required=True)
+
+
+class VstsAccountSelectionApiStep:
+    step_name = "account_selection"
+
+    def get_serializer_cls(self) -> type:
+        return VstsAccountSelectionSerializer
+
+    def get_step_data(self, pipeline: IntegrationPipeline, request: HttpRequest) -> dict[str, Any]:
+        with IntegrationPipelineViewEvent(
+            IntegrationPipelineViewType.ACCOUNT_CONFIG,
+            IntegrationDomain.SOURCE_CODE_MANAGEMENT,
+            VstsIntegrationProvider.key,
+        ).capture() as lifecycle:
+            oauth_data = pipeline.fetch_state("oauth_data") or {}
+            access_token = oauth_data.get("access_token")
+            user = get_user_info(access_token)
+
+            accounts = get_accounts(access_token, user["uuid"])
+            extra = {
+                "organization_id": pipeline.organization.id if pipeline.organization else None,
+                "user_id": request.user.id,
+                "accounts": accounts,
+            }
+
+            if not accounts or not accounts.get("value"):
+                lifecycle.record_halt(IntegrationPipelineHaltReason.NO_ACCOUNTS, extra=extra)
+                pipeline.bind_state("accounts", [])
+                return {"accounts": []}
+
+            accounts_list = accounts["value"]
+            pipeline.bind_state("accounts", accounts_list)
+            return {
+                "accounts": [
+                    {"accountId": account["accountId"], "accountName": account["accountName"]}
+                    for account in accounts_list
+                ]
+            }
+
+    def handle_post(
+        self,
+        validated_data: dict[str, str],
+        pipeline: IntegrationPipeline,
+        request: HttpRequest,
+    ) -> PipelineStepResult:
+        account_id = validated_data["account"]
+        state_accounts: Sequence[Mapping[str, Any]] | None = pipeline.fetch_state(key="accounts")
+        account = get_account_from_id(account_id, state_accounts or [])
+        if account is None:
+            return PipelineStepResult.error("Invalid Azure DevOps account")
+
+        pipeline.bind_state("account", account)
+        return PipelineStepResult.advance()
+
+
 class VstsIntegrationProvider(IntegrationProvider):
     key = IntegrationProviderSlug.AZURE_DEVOPS.value
     name = "Azure DevOps"
@@ -486,8 +571,34 @@ def get_pipeline_views(self) -> Sequence[PipelineView[IntegrationPipeline]]:
             AccountConfigView(),
         ]
 
+    def get_pipeline_api_steps(self) -> ApiPipelineSteps[IntegrationPipeline]:
+        return [
+            self._make_oauth_api_step(),
+            VstsAccountSelectionApiStep(),
+        ]
+
+    def _make_oauth_api_step(self) -> OAuth2ApiStep:
+        provider = VSTSNewIdentityProvider(
+            oauth_scopes=sorted(self.get_scopes()),
+            redirect_url=absolute_uri(self.oauth_redirect_url),
+        )
+        extra_authorize_params = {"response_mode": "query"}
+        if options.get("vsts.consent-prompt"):
+            extra_authorize_params["prompt"] = "consent"
+
+        return provider.make_oauth_api_step(
+            bind_key="oauth_data",
+            extra_authorize_params=extra_authorize_params,
+        )
+
     def build_integration(self, state: Mapping[str, Any]) -> IntegrationData:
-        data = state["identity"]["data"]
+        # TODO: legacy views write token data to state["identity"]["data"] via
+        # NestedPipelineView. API steps write directly to state["oauth_data"].
+        # Remove the legacy path once the old views are retired.
+        if "oauth_data" in state:
+            data = state["oauth_data"]
+        else:
+            data = state["identity"]["data"]
         oauth_data = self.get_oauth_data(data)
         account = state["account"]
         user = get_user_info(data["access_token"])
@@ -679,7 +790,7 @@ def dispatch(self, request: HttpRequest, pipeline: IntegrationPipeline) -> HttpR
                 state_accounts: Sequence[Mapping[str, Any]] | None = pipeline.fetch_state(
                     key="accounts"
                 )
-                account = self.get_account_from_id(account_id, state_accounts or [])
+                account = get_account_from_id(account_id, state_accounts or [])
                 if account is not None:
                     pipeline.bind_state("account", account)
                     return pipeline.next_step()
@@ -688,7 +799,7 @@ def dispatch(self, request: HttpRequest, pipeline: IntegrationPipeline) -> HttpR
             access_token = (state or {}).get("data", {}).get("access_token")
             user = get_user_info(access_token)
 
-            accounts = self.get_accounts(access_token, user["uuid"])
+            accounts = get_accounts(access_token, user["uuid"])
             extra = {
                 "organization_id": pipeline.organization.id if pipeline.organization else None,
                 "user_id": request.user.id,
@@ -710,30 +821,6 @@ def dispatch(self, request: HttpRequest, pipeline: IntegrationPipeline) -> HttpR
                 request=request,
             )
 
-    def get_account_from_id(
-        self, account_id: str, accounts: Sequence[Mapping[str, Any]]
-    ) -> Mapping[str, Any] | None:
-        for account in accounts:
-            if account["accountId"] == account_id:
-                return account
-        return None
-
-    def get_accounts(self, access_token: str, user_id: int) -> Any | None:
-        url = (
-            f"https://app.vssps.visualstudio.com/_apis/accounts?memberId={user_id}&api-version=4.1"
-        )
-        with http.build_session() as session:
-            response = session.get(
-                url,
-                headers={
-                    "Content-Type": "application/json",
-                    "Authorization": f"Bearer {access_token}",
-                },
-            )
-        if response.status_code == 200:
-            return response.json()
-        return None
-
 
 class AccountForm(forms.Form):
     def __init__(self, accounts: Sequence[Mapping[str, str]], *args: Any, **kwargs: Any) -> None:

tests/sentry/integrations/vsts/test_integration.py

@@ -6,13 +6,15 @@
 
 import pytest
 import responses
+from django.http import HttpRequest
+from django.urls import reverse
 
 from fixtures.vsts import CREATE_SUBSCRIPTION, VstsIntegrationTestCase
 from sentry.identity.vsts.provider import VSTSNewIdentityProvider
 from sentry.integrations.models.integration import Integration
 from sentry.integrations.models.integration_external_project import IntegrationExternalProject
 from sentry.integrations.models.organization_integration import OrganizationIntegration
-from sentry.integrations.pipeline import ensure_integration
+from sentry.integrations.pipeline import IntegrationPipeline, ensure_integration
 from sentry.integrations.vsts import VstsIntegration, VstsIntegrationProvider
 from sentry.models.repository import Repository
 from sentry.shared_integrations.exceptions import IntegrationError, IntegrationProviderError
@@ -318,6 +320,173 @@ def test_fix_subscription(self, mock_get_scopes: MagicMock) -> None:
         subscription = data["metadata"]["subscription"]
         assert subscription["id"] is not None and subscription["secret"] is not None
 
+
+@control_silo_test
+class VstsIntegrationApiPipelineTest(VstsIntegrationTestCase):
+    def _get_pipeline_url(self) -> str:
+        return reverse(
+            "sentry-api-0-organization-pipeline",
+            kwargs={
+                "organization_id_or_slug": self.organization.slug,
+                "pipeline_name": "integration_pipeline",
+            },
+        )
+
+    def _initialize_pipeline(self) -> Any:
+        return self.client.post(
+            self._get_pipeline_url(),
+            data={"action": "initialize", "provider": "vsts"},
+            format="json",
+        )
+
+    def _advance_step(self, data: dict[str, Any]) -> Any:
+        return self.client.post(self._get_pipeline_url(), data=data, format="json")
+
+    def _get_step_info(self) -> Any:
+        return self.client.get(self._get_pipeline_url())
+
+    def _get_pipeline_signature(self, init_resp: Any) -> str:
+        return parse_qs(urlparse(init_resp.data["data"]["oauthUrl"]).query)["state"][0]
+
+    def test_initialize_pipeline(self) -> None:
+        resp = self._initialize_pipeline()
+        assert resp.status_code == 200
+        assert resp.data["step"] == "oauth_login"
+        assert resp.data["stepIndex"] == 0
+        assert resp.data["totalSteps"] == 2
+        assert resp.data["provider"] == "vsts"
+        assert "oauthUrl" in resp.data["data"]
+
+    def test_get_oauth_step_info(self) -> None:
+        self._initialize_pipeline()
+        resp = self._get_step_info()
+        assert resp.status_code == 200
+        assert resp.data["step"] == "oauth_login"
+        oauth_url = resp.data["data"]["oauthUrl"]
+        assert "login.microsoftonline.com/common/oauth2/v2.0/authorize" in oauth_url
+        assert "client_id=" in oauth_url
+
+    def test_oauth_step_advance(self) -> None:
+        resp = self._initialize_pipeline()
+        pipeline_signature = self._get_pipeline_signature(resp)
+
+        resp = self._advance_step({"code": "oauth-code", "state": pipeline_signature})
+        assert resp.status_code == 200
+        assert resp.data["status"] == "advance"
+        assert resp.data["step"] == "account_selection"
+        assert resp.data["stepIndex"] == 1
+
+    def test_oauth_step_invalid_state(self) -> None:
+        self._initialize_pipeline()
+        resp = self._advance_step({"code": "oauth-code", "state": "invalid-state"})
+        assert resp.status_code == 400
+        assert resp.data["status"] == "error"
+        assert resp.data["data"]["detail"] == "Invalid state"
+
+    def test_account_selection_get(self) -> None:
+        resp = self._initialize_pipeline()
+        pipeline_signature = self._get_pipeline_signature(resp)
+        resp = self._advance_step({"code": "oauth-code", "state": pipeline_signature})
+        assert resp.data["step"] == "account_selection"
+
+        resp = self._get_step_info()
+        assert resp.status_code == 200
+        assert resp.data["step"] == "account_selection"
+        assert resp.data["data"]["accounts"] == [
+            {"accountId": self.vsts_account_id, "accountName": self.vsts_account_name}
+        ]
+
+    def test_account_selection_invalid_account(self) -> None:
+        resp = self._initialize_pipeline()
+        pipeline_signature = self._get_pipeline_signature(resp)
+        self._advance_step({"code": "oauth-code", "state": pipeline_signature})
+
+        resp = self._advance_step({"account": "invalid-account"})
+        assert resp.status_code == 400
+        assert resp.data["status"] == "error"
+        assert resp.data["data"]["detail"] == "Invalid Azure DevOps account"
+
+    def test_full_api_pipeline_flow(self) -> None:
+        resp = self._initialize_pipeline()
+        pipeline_signature = self._get_pipeline_signature(resp)
+
+        resp = self._advance_step({"code": "oauth-code", "state": pipeline_signature})
+        assert resp.status_code == 200
+        assert resp.data["status"] == "advance"
+        assert resp.data["step"] == "account_selection"
+
+        resp = self._advance_step({"account": self.vsts_account_id})
+        assert resp.status_code == 200
+        assert resp.data["status"] == "complete"
+        assert "data" in resp.data
+
+        integration = Integration.objects.get(provider="vsts")
+        assert integration.external_id == self.vsts_account_id
+        assert integration.name == self.vsts_account_name
+
+        assert OrganizationIntegration.objects.filter(
+            organization_id=self.organization.id,
+            integration=integration,
+        ).exists()
+
+    @patch("sentry.integrations.utils.metrics.EventLifecycle.record_event")
+    def test_account_selection_get_with_no_accounts(self, mock_record: MagicMock) -> None:
+        responses.replace(
+            responses.GET,
+            "https://app.vssps.visualstudio.com/_apis/accounts?memberId=%s&api-version=4.1"
+            % self.vsts_user_id,
+            status=403,
+            json={"$id": 1, "message": "Your account is not good"},
+        )
+
+        resp = self._initialize_pipeline()
+        pipeline_signature = self._get_pipeline_signature(resp)
+        resp = self._advance_step({"code": "oauth-code", "state": pipeline_signature})
+        assert resp.status_code == 200
+        assert resp.data["step"] == "account_selection"
+
+        resp = self._get_step_info()
+        assert resp.status_code == 200
+        assert resp.data["data"]["accounts"] == []
+
+        assert_halt_metric(mock_record, "no_accounts")
+
+    def test_build_integration_uses_oauth_data_state(self) -> None:
+        provider = VstsIntegrationProvider()
+        pipeline = Mock()
+        pipeline.organization = self.organization
+        provider.set_pipeline(pipeline)
+
+        data = provider.build_integration(
+            {
+                "account": {
+                    "accountName": self.vsts_account_name,
+                    "accountId": self.vsts_account_id,
+                },
+                "oauth_data": {
+                    "access_token": self.access_token,
+                    "expires_in": "3600",
+                    "refresh_token": self.refresh_token,
+                    "token_type": "bearer",
+                },
+            }
+        )
+
+        assert data["external_id"] == self.vsts_account_id
+        assert data["name"] == self.vsts_account_name
+
+    def test_initialize_binds_pipeline_state(self) -> None:
+        resp = self._initialize_pipeline()
+        assert resp.status_code == 200
+
+        request = HttpRequest()
+        request.session = self.client.session
+        request.user = self.user
+
+        pipeline = IntegrationPipeline.get_for_request(request)
+        assert pipeline is not None
+        assert pipeline.provider.key == "vsts"
+
     @responses.activate
     def test_source_url_matches(self) -> None:
         self.assert_installation()