feat(integrations): accept Azure DevOps work item URLs when linking

src/sentry/integrations/vsts/issues.py

@@ -1,8 +1,10 @@
 from __future__ import annotations
 
+import re
 from abc import ABC
 from collections.abc import Mapping, MutableMapping, Sequence
 from typing import TYPE_CHECKING, Any, NoReturn
+from urllib.parse import ParseResult
 
 from django.urls import reverse
 from django.utils.translation import gettext as _
@@ -16,6 +18,7 @@
 from sentry.integrations.services.integration import integration_service
 from sentry.integrations.source_code_management.issues import SourceCodeIssueIntegration
 from sentry.integrations.types import IntegrationIssueConfigField, IntegrationProviderSlug
+from sentry.integrations.utils.issue_url import get_url_origin, parse_issue_url
 from sentry.models.activity import Activity
 from sentry.shared_integrations.exceptions import (
     ApiError,
@@ -248,6 +251,37 @@ def create_issue(self, data: Mapping[str, str], **kwargs: Any) -> Mapping[str, A
             "metadata": {"display_name": "{}#{}".format(project_name, created_item["id"])},
         }
 
+    def get_issue_link_data(self, url: str) -> dict[str, str]:
+        parsed = parse_issue_url(url)
+        base = parse_issue_url(self.instance)
+
+        def account_and_path(parsed_url: ParseResult) -> tuple[str | None, str]:
+            if parsed_url.hostname == "dev.azure.com":
+                account, _, path = parsed_url.path.lstrip("/").partition("/")
+                return account.casefold(), f"/{path}"
+            if parsed_url.hostname and parsed_url.hostname.endswith(".visualstudio.com"):
+                return parsed_url.hostname.removesuffix(".visualstudio.com"), parsed_url.path
+            return None, parsed_url.path
+
+        account, path = account_and_path(parsed)
+        installed_account, _ = account_and_path(base)
+        match = re.fullmatch(r"/(?:[^/]+/)?_workitems/edit/(\d+)", path)
+        origin = get_url_origin(parsed)
+        base_origin = get_url_origin(base)
+        if (
+            not account
+            or account != installed_account
+            or origin is None
+            or base_origin is None
+            or origin[0] != base_origin[0]
+            or origin[2] != base_origin[2]
+            or not match
+        ):
+            raise IntegrationFormError(
+                {"externalIssue": "Issue URL does not belong to this installation"}
+            )
+        return {"externalIssue": match[1]}
+
     def get_issue(self, issue_id: int, **kwargs: Any) -> Mapping[str, Any]:
         client = self.get_client()
         work_item = client.get_work_item(issue_id)

tests/sentry/integrations/vsts/test_issues.py

@@ -260,6 +260,25 @@ def test_create_issue_failure(self) -> None:
         with pytest.raises(IntegrationFormError):
             self.integration.create_issue(form_data)
 
+    def test_issue_url(self) -> None:
+        self.integration.model.metadata["domain_name"] = (
+            "https://Fabrikam-Fiber-Inc.VisualStudio.COM/"
+        )
+        for url in (
+            "https://fabrikam-fiber-inc.visualstudio.com/project/_workitems/edit/309",
+            "https://FABRIKAM-FIBER-INC.visualstudio.com/project/_workitems/edit/309",
+            "https://dev.azure.com/FABRIKAM-FIBER-INC/project/_workitems/edit/309?view=1",
+            "https://dev.azure.com:443/fabrikam-fiber-inc/project/_workitems/edit/309",
+        ):
+            assert self.integration.get_issue_link_data(url) == {"externalIssue": "309"}
+        for url in (
+            "https://dev.azure.com/other/_workitems/edit/309",
+            "http://dev.azure.com/fabrikam-fiber-inc/_workitems/edit/309",
+            "https://dev.azure.com:8443/fabrikam-fiber-inc/_workitems/edit/309",
+        ):
+            with pytest.raises(IntegrationFormError):
+                self.integration.get_issue_link_data(url)
+
     @responses.activate
     def test_get_issue(self) -> None:
         responses.add(

tests/sentry/issues/endpoints/test_group_integration_details.py

@@ -1,9 +1,11 @@
+from time import time
 from typing import Any
 from unittest import mock
 
 import responses
 from django.db.utils import IntegrityError
 
+from fixtures.vsts import WORK_ITEM_RESPONSE
 from sentry.integrations.example.integration import ExampleIntegration
 from sentry.integrations.models import Integration
 from sentry.integrations.models.external_issue import ExternalIssue
@@ -322,6 +324,47 @@ def test_put_github_issue_url(self) -> None:
             }
         assert len(responses.calls) == 3
 
+    @responses.activate
+    def test_put_azure_issue_url(self) -> None:
+        self.login_as(self.user)
+        identity = self.create_identity(
+            user=self.user,
+            identity_provider=self.create_identity_provider(type="vsts"),
+            external_id="vsts",
+            data={"access_token": "access-token", "expires": time() + 3600},
+        )
+        integration = self.create_integration(
+            organization=self.organization,
+            provider="vsts",
+            external_id="vsts:1",
+            name="fabrikam-fiber-inc",
+            metadata={"domain_name": "https://Fabrikam-Fiber-Inc.VisualStudio.COM/"},
+            oi_params={"default_auth_id": identity.id},
+        )
+        responses.get(
+            "https://fabrikam-fiber-inc.visualstudio.com/_apis/wit/workitems/309",
+            body=WORK_ITEM_RESPONSE,
+            content_type="application/json",
+        )
+
+        with self.feature("organizations:integrations-issue-basic"):
+            for url in (
+                "https://fabrikam-fiber-inc.visualstudio.com/project/_workitems/edit/309",
+                "https://dev.azure.com/FABRIKAM-FIBER-INC/project/_workitems/edit/309?view=1",
+            ):
+                group = self.create_group(project=self.project)
+                path = f"/api/0/organizations/{self.organization.slug}/issues/{group.id}/integrations/{integration.id}/"
+                response = self.client.put(path, data={"externalIssue": url})
+                assert response.status_code == 201
+                assert response.data["key"] == "309"
+                assert GroupLink.objects.filter(
+                    group_id=group.id,
+                    linked_id=response.data["id"],
+                    linked_type=GroupLink.LinkedType.issue,
+                    relationship=GroupLink.Relationship.references,
+                ).exists()
+        assert len(responses.calls) == 2
+
     @responses.activate
     def test_put_jira_issue_url(self) -> None:
         self.login_as(self.user)