ref(tests): Slack Linking Tests

tests/sentry/api/validators/sentry_apps/test_schema.py

@@ -103,7 +103,7 @@ def test_invalid_schema_type_invalid(self):
     @invalid_schema_with_error_message(
         "'uri' is a required property for element of type 'stacktrace-link'"
     )
-    def test_invalid_chema_element_missing_uri(self):
+    def test_invalid_schema_element_missing_uri(self):
         schema = {
             "elements": [{"url": "/stacktrace/github/getsentry/sentry", "type": "stacktrace-link"}]
         }

tests/sentry/integrations/slack/__init__.py

@@ -1,12 +1,28 @@
+from typing import Optional
+
+from sentry.integrations.slack.message_builder import SlackBody
 from sentry.models import (
+    ExternalActor,
     Identity,
     IdentityProvider,
     IdentityStatus,
     Integration,
     Organization,
     OrganizationIntegration,
+    Team,
     User,
 )
+from sentry.types.integrations import EXTERNAL_PROVIDERS, ExternalProviders
+
+
+def get_response_text(data: SlackBody) -> str:
+    return (
+        # If it's an attachment.
+        data.get("text")
+        or
+        # If it's blocks.
+        "\n".join(block["text"]["text"] for block in data["blocks"] if block["type"] == "section")
+    )
 
 
 def install_slack(organization: Organization, workspace_id: str = "TXXXXXXX1") -> Integration:
@@ -28,9 +44,43 @@ def add_identity(
     integration: Integration, user: User, external_id: str = "UXXXXXXX1"
 ) -> IdentityProvider:
     idp = IdentityProvider.objects.create(
-        type="slack", external_id=integration.external_id, config={}
+        type=EXTERNAL_PROVIDERS[ExternalProviders.SLACK],
+        external_id=integration.external_id,
+        config={},
     )
     Identity.objects.create(
         user=user, idp=idp, external_id=external_id, status=IdentityStatus.VALID
     )
     return idp
+
+
+def find_identity(idp: IdentityProvider, user: User) -> Optional[Identity]:
+    identities = Identity.objects.filter(
+        idp=idp,
+        user=user,
+        status=IdentityStatus.VALID,
+    )
+    if not identities:
+        return None
+    return identities[0]
+
+
+def link_user(user: User, idp: IdentityProvider, slack_id: str) -> None:
+    Identity.objects.create(
+        external_id=slack_id,
+        idp=idp,
+        user=user,
+        status=IdentityStatus.VALID,
+        scopes=[],
+    )
+
+
+def link_team(team: Team, integration: Integration, channel_name: str, channel_id: str) -> None:
+    ExternalActor.objects.create(
+        actor_id=team.actor_id,
+        organization=team.organization,
+        integration=integration,
+        provider=ExternalProviders.SLACK.value,
+        external_name=channel_name,
+        external_id=channel_id,
+    )

tests/sentry/integrations/slack/endpoints/commands/__init__.py

@@ -0,0 +1,75 @@
+from typing import Any, Mapping, Optional
+from urllib.parse import urlencode
+
+from django.urls import reverse
+from requests import Response
+from rest_framework import status
+
+from sentry import options
+from sentry.integrations.slack.utils import set_signing_secret
+from sentry.models import Identity, IdentityProvider, Team
+from sentry.testutils import APITestCase, TestCase
+from sentry.types.integrations import EXTERNAL_PROVIDERS, ExternalProviders
+from sentry.utils import json
+from tests.sentry.integrations.slack import find_identity, install_slack, link_team, link_user
+
+
+class SlackCommandsTest(APITestCase, TestCase):
+    endpoint = "sentry-integration-slack-commands"
+    method = "post"
+
+    def setUp(self):
+        super().setUp()
+
+        self.slack_id = "UXXXXXXX1"
+        self.external_id = "new-slack-id"
+        self.channel_name = "my-channel"
+        self.channel_id = "my-channel_id"
+        self.response_url = "http://example.slack.com/response_url"
+
+        self.integration = install_slack(self.organization, self.external_id)
+        self.idp = IdentityProvider.objects.create(
+            type=EXTERNAL_PROVIDERS[ExternalProviders.SLACK],
+            external_id=self.external_id,
+            config={},
+        )
+        self.login_as(self.user)
+
+    def send_slack_message(self, command: str, **kwargs: Any) -> Mapping[str, str]:
+        response = self.get_slack_response(
+            {
+                "text": command,
+                "team_id": self.external_id,
+                "user_id": self.slack_id,
+                **kwargs,
+            }
+        )
+        return json.loads(str(response.content.decode("utf-8")))
+
+    def find_identity(self) -> Optional[Identity]:
+        return find_identity(idp=self.idp, user=self.user)
+
+    def link_user(self) -> None:
+        return link_user(user=self.user, idp=self.idp, slack_id=self.slack_id)
+
+    def link_team(self, team: Optional[Team] = None) -> None:
+        return link_team(
+            team=team or self.team,
+            integration=self.integration,
+            channel_name=self.channel_name,
+            channel_id=self.channel_id,
+        )
+
+    def get_slack_response(
+        self, payload: Mapping[str, str], status_code: Optional[str] = None
+    ) -> Response:
+        """Shadow get_success_response but with a non-JSON payload."""
+        data = urlencode(payload).encode("utf-8")
+        response = self.client.post(
+            reverse(self.endpoint),
+            content_type="application/x-www-form-urlencoded",
+            data=data,
+            **set_signing_secret(options.get("slack.signing-secret"), data),
+        )
+        assert response.status_code == (status_code or status.HTTP_200_OK)
+        return response

tests/sentry/integrations/slack/endpoints/commands/test_get.py

@@ -0,0 +1,10 @@
+from rest_framework import status
+
+from tests.sentry.integrations.slack.endpoints.commands import SlackCommandsTest
+
+
+class SlackCommandsGetTest(SlackCommandsTest):
+    method = "get"
+
+    def test_method_get_not_allowed(self):
+        self.get_error_response(status_code=status.HTTP_405_METHOD_NOT_ALLOWED)

tests/sentry/integrations/slack/endpoints/commands/test_help.py

@@ -0,0 +1,32 @@
+from typing import Optional
+
+from sentry.integrations.slack.message_builder import SlackBody
+from tests.sentry.integrations.slack import get_response_text
+from tests.sentry.integrations.slack.endpoints.commands import SlackCommandsTest
+
+
+def assert_is_help_text(data: SlackBody, expected_command: Optional[str] = None) -> None:
+    text = get_response_text(data)
+    assert "Here are the commands you can use" in text
+    if expected_command:
+        assert expected_command in text
+
+
+def assert_unknown_command_text(data: SlackBody, unknown_command: Optional[str] = None) -> None:
+    text = get_response_text(data)
+    assert f"Unknown command: `{unknown_command}`" in text
+    assert "Here are the commands you can use" in text
+
+
+class SlackCommandsHelpTest(SlackCommandsTest):
+    def test_missing_command(self):
+        data = self.send_slack_message("")
+        assert_is_help_text(data)
+
+    def test_invalid_command(self):
+        data = self.send_slack_message("invalid command")
+        assert_unknown_command_text(data, "invalid command")
+
+    def test_help_command(self):
+        data = self.send_slack_message("help")
+        assert_is_help_text(data)

tests/sentry/integrations/slack/endpoints/commands/test_link_team.py

@@ -0,0 +1,134 @@
+import responses
+from rest_framework import status
+
+from sentry.integrations.slack.endpoints.command import (
+    CHANNEL_ALREADY_LINKED_MESSAGE,
+    INSUFFICIENT_ROLE_MESSAGE,
+    LINK_FROM_CHANNEL_MESSAGE,
+    LINK_USER_FIRST_MESSAGE,
+    TEAM_NOT_LINKED_MESSAGE,
+)
+from sentry.models import OrganizationIntegration
+from sentry.utils import json
+from tests.sentry.integrations.slack import get_response_text, link_user
+from tests.sentry.integrations.slack.endpoints.commands import SlackCommandsTest
+
+OTHER_SLACK_ID = "UXXXXXXX2"
+
+
+class SlackCommandsLinkTeamTestBase(SlackCommandsTest):
+    def setUp(self):
+        super().setUp()
+        self.link_user()
+        responses.add(
+            method=responses.POST,
+            url="https://slack.com/api/chat.postMessage",
+            body='{"ok": true}',
+            status=status.HTTP_200_OK,
+            content_type="application/json",
+        )
+
+
+class SlackCommandsLinkTeamTest(SlackCommandsLinkTeamTestBase):
+    def test_link_another_team_to_channel(self):
+        """
+        Test that we block a user who tries to link a second team to a
+        channel that already has a team linked to it.
+        """
+        self.link_team()
+
+        response = self.get_slack_response(
+            {
+                "text": "link team",
+                "team_id": self.external_id,
+                "user_id": self.slack_id,
+                "channel_name": self.channel_name,
+                "channel_id": self.channel_id,
+            }
+        )
+        data = json.loads(str(response.content.decode("utf-8")))
+        assert CHANNEL_ALREADY_LINKED_MESSAGE in get_response_text(data)
+
+    def test_link_team_from_dm(self):
+        """
+        Test that if a user types `/sentry link team` from a DM instead of a
+        channel, we reply with an error message.
+        """
+        response = self.get_slack_response(
+            {
+                "text": "link team",
+                "team_id": self.external_id,
+                "user_id": OTHER_SLACK_ID,
+                "channel_name": "directmessage",
+            }
+        )
+        data = json.loads(str(response.content.decode("utf-8")))
+        assert LINK_FROM_CHANNEL_MESSAGE in get_response_text(data)
+
+    def test_link_team_identity_does_not_exist(self):
+        """Test that get_identity fails if the user has no Identity and we reply with the LINK_USER_MESSAGE"""
+        user2 = self.create_user()
+        self.create_member(
+            teams=[self.team], user=user2, role="member", organization=self.organization
+        )
+        self.login_as(user2)
+        data = self.send_slack_message("link team", user_id=OTHER_SLACK_ID)
+        assert LINK_USER_FIRST_MESSAGE in get_response_text(data)
+
+    @responses.activate
+    def test_link_team_insufficient_role(self):
+        """
+        Test that when a user whose role is insufficient attempts to link a
+        team, we reject them and reply with the INSUFFICIENT_ROLE_MESSAGE.
+        """
+        user2 = self.create_user()
+        self.create_member(
+            teams=[self.team], user=user2, role="member", organization=self.organization
+        )
+        self.login_as(user2)
+        link_user(user2, self.idp, slack_id=OTHER_SLACK_ID)
+
+        data = self.send_slack_message("link team", user_id=OTHER_SLACK_ID)
+        assert INSUFFICIENT_ROLE_MESSAGE in get_response_text(data)
+
+
+class SlackCommandsUnlinkTeamTest(SlackCommandsLinkTeamTestBase):
+    def setUp(self):
+        super().setUp()
+        self.link_team()
+
+    def test_unlink_team(self):
+        data = self.send_slack_message(
+            "unlink team",
+            channel_name=self.channel_name,
+            channel_id=self.channel_id,
+        )
+        assert "Click here to unlink your team from this channel" in get_response_text(data)
+
+    def test_unlink_no_team(self):
+        """
+        Test for when a user attempts to remove a link between a Slack channel
+        and a Sentry team that does not exist.
+        """
+        data = self.send_slack_message(
+            "unlink team",
+            channel_name="specific",
+            channel_id=OTHER_SLACK_ID,
+        )
+        assert TEAM_NOT_LINKED_MESSAGE in get_response_text(data)
+
+    def test_unlink_multiple_orgs(self):
+        # Create another organization and team for this user that is linked through `self.integration`.
+        organization2 = self.create_organization(owner=self.user)
+        team2 = self.create_team(organization=organization2, members=[self.user])
+        OrganizationIntegration.objects.create(
+            organization=organization2, integration=self.integration
+        )
+        self.link_team(team2)
+
+        data = self.send_slack_message(
+            "unlink team",
+            channel_name=self.channel_name,
+            channel_id=self.channel_id,
+        )
+        assert "Click here to unlink your team from this channel" in get_response_text(data)