feat(bitbucket): Add API-driven pipeline backend for Bitbucket integration setup

Implement `get_pipeline_api_steps()` on `BitbucketIntegrationProvider` with
a single authorize step that verifies the Atlassian Connect JWT from
Bitbucket's addon authorization flow. The frontend posts the JWT received
from the popup callback, and the backend validates it against the original
callback path using `get_integration_from_jwt`. Updates `build_integration()`
to handle both legacy and API state paths.

Ref [VDY-41](https://linear.app/getsentry/issue/VDY-41/bitbucket-api-driven-integration-setup)

Author: evanpurkhiser

src/sentry/integrations/bitbucket/integration.py

@@ -5,8 +5,11 @@
 
 from django.http.request import HttpRequest
 from django.http.response import HttpResponseBase
+from django.urls import reverse
 from django.utils.datastructures import OrderedSet
 from django.utils.translation import gettext_lazy as _
+from rest_framework.fields import CharField
+from rest_framework.serializers import Serializer
 
 from sentry.identity.pipeline import IdentityPipeline
 from sentry.integrations.base import (
@@ -25,6 +28,7 @@
 from sentry.integrations.types import IntegrationProviderSlug
 from sentry.integrations.utils.atlassian_connect import (
     AtlassianConnectValidationError,
+    get_integration_from_jwt,
     get_integration_from_request,
 )
 from sentry.integrations.utils.metrics import (
@@ -34,7 +38,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
 from sentry.utils.http import absolute_uri
@@ -194,6 +199,46 @@ def username(self):
         return self.model.name
 
 
+class BitbucketVerifySerializer(Serializer):
+    jwt = CharField(required=True)
+
+
+class BitbucketAuthorizeApiStep:
+    step_name = "authorize"
+
+    def get_step_data(self, pipeline: IntegrationPipeline, request: HttpRequest) -> dict[str, Any]:
+        descriptor_uri = absolute_uri("/extensions/bitbucket/descriptor/")
+        authorize_url = (
+            f"https://bitbucket.org/site/addons/authorize?descriptor_uri={descriptor_uri}"
+        )
+        return {"authorizeUrl": authorize_url}
+
+    def get_serializer_cls(self) -> type:
+        return BitbucketVerifySerializer
+
+    def handle_post(
+        self,
+        validated_data: dict[str, Any],
+        pipeline: IntegrationPipeline,
+        request: HttpRequest,
+    ) -> PipelineStepResult:
+        callback_path = reverse(
+            "sentry-extension-setup",
+            kwargs={"provider_id": IntegrationProviderSlug.BITBUCKET.value},
+        )
+        try:
+            integration = get_integration_from_jwt(
+                token=validated_data["jwt"],
+                path=callback_path,
+                provider=IntegrationProviderSlug.BITBUCKET.value,
+                query_params=None,
+            )
+        except AtlassianConnectValidationError:
+            return PipelineStepResult.error("Unable to verify installation.")
+        pipeline.bind_state("external_id", integration.external_id)
+        return PipelineStepResult.advance()
+
+
 class BitbucketIntegrationProvider(IntegrationProvider):
     key = IntegrationProviderSlug.BITBUCKET.value
     name = "Bitbucket"
@@ -220,6 +265,9 @@ def get_pipeline_views(self) -> Sequence[PipelineView[IntegrationPipeline]]:
             VerifyInstallation(),
         ]
 
+    def get_pipeline_api_steps(self) -> ApiPipelineSteps[IntegrationPipeline]:
+        return [BitbucketAuthorizeApiStep()]
+
     def post_install(
         self,
         integration: Integration,

tests/sentry/integrations/bitbucket/test_integration.py

@@ -1,3 +1,6 @@
+from __future__ import annotations
+
+from typing import Any
 from unittest.mock import MagicMock, patch
 from urllib.parse import quote, urlencode
 
@@ -7,6 +10,10 @@
 
 from sentry.integrations.bitbucket.integration import BitbucketIntegrationProvider
 from sentry.integrations.models.integration import Integration
+from sentry.integrations.models.organization_integration import OrganizationIntegration
+from sentry.integrations.pipeline import IntegrationPipeline
+from sentry.integrations.services.integration.model import RpcIntegration
+from sentry.integrations.utils.atlassian_connect import AtlassianConnectValidationError
 from sentry.models.repository import Repository
 from sentry.shared_integrations.exceptions import IntegrationError
 from sentry.silo.base import SiloMode
@@ -268,3 +275,93 @@ def test_get_repository_choices_failure_lifecycle(
             installation.get_repository_choices(None, {})
         assert mock_record_halt.call_count == 0
         assert mock_record_failure.call_count == 1
+
+
+@control_silo_test
+class BitbucketApiPipelineTest(APITestCase):
+    endpoint = "sentry-api-0-organization-pipeline"
+    method = "post"
+
+    def setUp(self) -> None:
+        super().setUp()
+        self.login_as(self.user)
+        self.subject = "connect:1234567"
+        self.shared_secret = "234567890"
+        self.integration = self.create_provider_integration(
+            provider="bitbucket",
+            external_id=self.subject,
+            name="sentryuser",
+            metadata={
+                "base_url": "https://api.bitbucket.org",
+                "shared_secret": self.shared_secret,
+            },
+        )
+
+    def _get_pipeline_url(self) -> str:
+        return reverse(
+            self.endpoint,
+            args=[self.organization.slug, IntegrationPipeline.pipeline_name],
+        )
+
+    def _initialize_pipeline(self) -> Any:
+        return self.client.post(
+            self._get_pipeline_url(),
+            data={"action": "initialize", "provider": "bitbucket"},
+            format="json",
+        )
+
+    def _advance_step(self, data: dict[str, Any]) -> Any:
+        return self.client.post(self._get_pipeline_url(), data=data, format="json")
+
+    @responses.activate
+    def test_initialize_pipeline(self) -> None:
+        resp = self._initialize_pipeline()
+        assert resp.status_code == 200
+        assert resp.data["step"] == "authorize"
+        assert resp.data["stepIndex"] == 0
+        assert resp.data["totalSteps"] == 1
+        assert resp.data["provider"] == "bitbucket"
+        assert "authorizeUrl" in resp.data["data"]
+        assert "bitbucket.org/site/addons/authorize" in resp.data["data"]["authorizeUrl"]
+
+    @responses.activate
+    def test_missing_jwt(self) -> None:
+        self._initialize_pipeline()
+        resp = self._advance_step({})
+        assert resp.status_code == 400
+
+    @responses.activate
+    @patch(
+        "sentry.integrations.bitbucket.integration.get_integration_from_jwt",
+        side_effect=AtlassianConnectValidationError("Invalid JWT"),
+    )
+    def test_invalid_jwt(self, mock_verify: MagicMock) -> None:
+        self._initialize_pipeline()
+        resp = self._advance_step({"jwt": "invalid-token"})
+        assert resp.status_code == 400
+        assert "Unable to verify installation" in resp.data["jwt"][0]
+
+    @responses.activate
+    @patch("sentry.integrations.bitbucket.integration.get_integration_from_jwt")
+    def test_full_pipeline_flow(self, mock_verify: MagicMock) -> None:
+        mock_verify.return_value = RpcIntegration(
+            id=self.integration.id,
+            provider=self.integration.provider,
+            external_id=self.subject,
+            name=self.integration.name,
+            metadata=self.integration.metadata,
+            status=self.integration.status,
+        )
+
+        resp = self._initialize_pipeline()
+        assert resp.data["step"] == "authorize"
+
+        resp = self._advance_step({"jwt": "valid-jwt-token"})
+        assert resp.status_code == 200
+        assert resp.data["status"] == "complete"
+        assert "data" in resp.data
+
+        assert OrganizationIntegration.objects.filter(
+            organization_id=self.organization.id,
+            integration=self.integration,
+        ).exists()