diff --git a/tests/sentry/api/validators/sentry_apps/test_schema.py b/tests/sentry/api/validators/sentry_apps/test_schema.py index ea1590a3f75370..97e43304b68e93 100644 --- a/tests/sentry/api/validators/sentry_apps/test_schema.py +++ b/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"}] } diff --git a/tests/sentry/integrations/slack/__init__.py b/tests/sentry/integrations/slack/__init__.py index f093a266c59715..76b68b5c6f92a0 100644 --- a/tests/sentry/integrations/slack/__init__.py +++ b/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, + ) diff --git a/tests/sentry/integrations/slack/endpoints/commands/__init__.py b/tests/sentry/integrations/slack/endpoints/commands/__init__.py new file mode 100644 index 00000000000000..a86a7977f8cbcc --- /dev/null +++ b/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 diff --git a/tests/sentry/integrations/slack/endpoints/commands/test_get.py b/tests/sentry/integrations/slack/endpoints/commands/test_get.py new file mode 100644 index 00000000000000..f190c73c5654a6 --- /dev/null +++ b/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) diff --git a/tests/sentry/integrations/slack/endpoints/commands/test_help.py b/tests/sentry/integrations/slack/endpoints/commands/test_help.py new file mode 100644 index 00000000000000..7b36ff6a9fffa4 --- /dev/null +++ b/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) diff --git a/tests/sentry/integrations/slack/endpoints/commands/test_link_team.py b/tests/sentry/integrations/slack/endpoints/commands/test_link_team.py new file mode 100644 index 00000000000000..0d43a2a4089e50 --- /dev/null +++ b/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) diff --git a/tests/sentry/integrations/slack/endpoints/commands/test_link_user.py b/tests/sentry/integrations/slack/endpoints/commands/test_link_user.py new file mode 100644 index 00000000000000..ff6993f30d6688 --- /dev/null +++ b/tests/sentry/integrations/slack/endpoints/commands/test_link_user.py @@ -0,0 +1,64 @@ +import responses + +from sentry.integrations.slack.endpoints.base import NOT_LINKED_MESSAGE +from sentry.integrations.slack.views.link_identity import SUCCESS_LINKED_MESSAGE, build_linking_url +from sentry.integrations.slack.views.unlink_identity import ( + SUCCESS_UNLINKED_MESSAGE, + build_unlinking_url, +) +from sentry.utils import json +from tests.sentry.integrations.slack import get_response_text +from tests.sentry.integrations.slack.endpoints.commands import SlackCommandsTest + + +class SlackCommandsLinkUserTest(SlackCommandsTest): + @responses.activate + def test_link_user_identity(self): + linking_url = build_linking_url( + self.integration, self.external_id, self.channel_id, self.response_url + ) + + response = self.client.post(linking_url) + assert response.status_code == 200 + + assert len(responses.calls) >= 1 + data = json.loads(str(responses.calls[0].request.body.decode("utf-8"))) + assert SUCCESS_LINKED_MESSAGE in get_response_text(data) + + def test_link_command(self): + data = self.send_slack_message("link") + assert "Link your Slack identity" in get_response_text(data) + + def test_link_command_already_linked(self): + self.link_user() + data = self.send_slack_message("link") + assert "You are already linked as" in get_response_text(data) + + +class SlackCommandsUnlinkUserTest(SlackCommandsTest): + @responses.activate + def test_unlink_user_identity(self): + self.link_user() + + unlinking_url = build_unlinking_url( + self.integration.id, + self.slack_id, + self.external_id, + self.response_url, + ) + + response = self.client.post(unlinking_url) + assert response.status_code == 200 + + assert len(responses.calls) >= 1 + data = json.loads(str(responses.calls[0].request.body.decode("utf-8"))) + assert SUCCESS_UNLINKED_MESSAGE in get_response_text(data) + + def test_unlink_command(self): + self.link_user() + data = self.send_slack_message("unlink") + assert "to unlink your identity" in get_response_text(data) + + def test_unlink_command_already_unlinked(self): + data = self.send_slack_message("unlink") + assert NOT_LINKED_MESSAGE in get_response_text(data) diff --git a/tests/sentry/integrations/slack/endpoints/commands/test_post.py b/tests/sentry/integrations/slack/endpoints/commands/test_post.py new file mode 100644 index 00000000000000..d686f73ec247a8 --- /dev/null +++ b/tests/sentry/integrations/slack/endpoints/commands/test_post.py @@ -0,0 +1,19 @@ +from rest_framework import status + +from sentry.integrations.slack.message_builder.disconnected import DISCONNECTED_MESSAGE +from tests.sentry.integrations.slack import get_response_text +from tests.sentry.integrations.slack.endpoints.commands import SlackCommandsTest + + +class SlackCommandsPostTest(SlackCommandsTest): + def test_invalid_signature(self): + # The `get_error_response` method doesn't add a signature to the request. + self.get_error_response(status_code=status.HTTP_400_BAD_REQUEST) + + def test_missing_team(self): + self.get_slack_response({"text": ""}, status_code=status.HTTP_400_BAD_REQUEST) + + def test_idp_does_not_exist(self): + """Test that get_identity fails if we cannot find a matching idp.""" + data = self.send_slack_message("", team_id="slack:2") + assert DISCONNECTED_MESSAGE in get_response_text(data) diff --git a/tests/sentry/integrations/slack/endpoints/test_commands.py b/tests/sentry/integrations/slack/endpoints/test_commands.py deleted file mode 100644 index 1195741d59c413..00000000000000 --- a/tests/sentry/integrations/slack/endpoints/test_commands.py +++ /dev/null @@ -1,565 +0,0 @@ -from typing import Any, Mapping, Optional -from urllib.parse import urlencode - -import responses -from django.urls import reverse -from requests import Response -from rest_framework import status - -from sentry import options -from sentry.integrations.slack.endpoints.base import NOT_LINKED_MESSAGE -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.integrations.slack.message_builder import SlackBody -from sentry.integrations.slack.message_builder.disconnected import DISCONNECTED_MESSAGE -from sentry.integrations.slack.utils import set_signing_secret -from sentry.integrations.slack.views.link_identity import SUCCESS_LINKED_MESSAGE, build_linking_url -from sentry.integrations.slack.views.link_team import build_team_linking_url -from sentry.integrations.slack.views.unlink_identity import ( - SUCCESS_UNLINKED_MESSAGE, - build_unlinking_url, -) -from sentry.integrations.slack.views.unlink_team import build_team_unlinking_url -from sentry.models import ( - ExternalActor, - Identity, - IdentityProvider, - IdentityStatus, - NotificationSetting, -) -from sentry.notifications.types import NotificationScopeType -from sentry.testutils import APITestCase, TestCase -from sentry.types.integrations import ExternalProviders -from sentry.utils import json -from tests.sentry.integrations.slack import install_slack - - -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 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 SlackCommandsTest(APITestCase, TestCase): - endpoint = "sentry-integration-slack-commands" - method = "post" - - 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": "UXXXXXXX1", - **kwargs, - } - ) - return json.loads(str(response.content.decode("utf-8"))) - - def find_identity(self) -> Optional[Identity]: - identities = Identity.objects.filter( - idp=self.idp, - user=self.user, - status=IdentityStatus.VALID, - ) - if not identities: - return None - return identities[0] - - def link_user(self) -> None: - Identity.objects.create( - external_id="UXXXXXXX1", - idp=self.idp, - user=self.user, - status=IdentityStatus.VALID, - scopes=[], - ) - - def link_team(self) -> None: - ExternalActor.objects.create( - actor_id=self.team.actor_id, - organization=self.organization, - integration=self.integration, - provider=ExternalProviders.SLACK.value, - external_name="general", - external_id="CXXXXXXX9", - ) - - 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 - - def setUp(self): - super().setUp() - self.external_id = "slack:1" - self.integration = install_slack(self.organization, self.external_id) - self.idp = IdentityProvider.objects.create(type="slack", external_id="slack:1", config={}) - self.login_as(self.user) - - -class SlackCommandsGetTest(SlackCommandsTest): - method = "get" - - def test_method_get_not_allowed(self): - self.get_error_response(status_code=status.HTTP_405_METHOD_NOT_ALLOWED) - - -class SlackCommandsPostTest(SlackCommandsTest): - def test_invalid_signature(self): - # The `get_error_response` method doesn't add a signature to the request. - self.get_error_response(status_code=status.HTTP_400_BAD_REQUEST) - - def test_missing_team(self): - self.get_slack_response({"text": ""}, status_code=status.HTTP_400_BAD_REQUEST) - - def test_idp_does_not_exist(self): - """Test that get_identity fails if we cannot find a matching idp.""" - data = self.send_slack_message("", team_id="slack:2") - assert DISCONNECTED_MESSAGE in get_response_text(data) - - -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) - - -class SlackCommandsLinkUserTest(SlackCommandsTest): - @responses.activate - def test_link_user_identity(self): - """Do the auth flow and assert that the identity was created.""" - # Assert that the identity does not exist. - assert not self.find_identity() - - linking_url = build_linking_url( - self.integration, "UXXXXXXX1", "CXXXXXXX9", "http://example.slack.com/response_url" - ) - - response = self.client.get(linking_url) - assert response.status_code == 200 - self.assertTemplateUsed(response, "sentry/auth-link-identity.html") - - response = self.client.post(linking_url) - assert response.status_code == 200 - self.assertTemplateUsed(response, "sentry/integrations/slack/linked.html") - - # Assert that the identity was created. - assert self.find_identity() - - assert len(responses.calls) >= 1 - data = json.loads(str(responses.calls[0].request.body.decode("utf-8"))) - assert SUCCESS_LINKED_MESSAGE in data["text"] - - @responses.activate - def test_unlink_user_identity(self): - self.link_user() - assert self.find_identity() - - unlinking_url = build_unlinking_url( - self.integration.id, - "UXXXXXXX1", - "CXXXXXXX9", - "http://example.slack.com/response_url", - ) - - response = self.client.get(unlinking_url) - assert response.status_code == 200 - self.assertTemplateUsed(response, "sentry/auth-unlink-identity.html") - - response = self.client.post(unlinking_url) - assert response.status_code == 200 - self.assertTemplateUsed(response, "sentry/integrations/slack/unlinked.html") - - # Assert that the identity was deleted. - assert not self.find_identity() - - assert len(responses.calls) >= 1 - data = json.loads(str(responses.calls[0].request.body.decode("utf-8"))) - assert SUCCESS_UNLINKED_MESSAGE in data["text"] - - def test_link_command(self): - data = self.send_slack_message("link") - assert "Link your Slack identity" in get_response_text(data) - - def test_unlink_command(self): - self.link_user() - data = self.send_slack_message("unlink") - assert "to unlink your identity" in get_response_text(data) - - def test_link_command_already_linked(self): - self.link_user() - data = self.send_slack_message("link") - assert "You are already linked as" in get_response_text(data) - - def test_unlink_command_already_unlinked(self): - data = self.send_slack_message("unlink") - assert NOT_LINKED_MESSAGE in get_response_text(data) - - -class SlackCommandsLinkTeamTest(SlackCommandsTest): - def setUp(self): - super().setUp() - self.link_user() - self.data = self.send_slack_message("link team") - responses.add( - method=responses.POST, - url="https://slack.com/api/chat.postMessage", - body='{"ok": true}', - status=200, - content_type="application/json", - ) - self.external_actor = ExternalActor.objects.filter( - actor_id=self.team.actor_id, - organization=self.organization, - integration=self.integration, - provider=ExternalProviders.SLACK.value, - external_name="general", - external_id="CXXXXXXX9", - ) - - @responses.activate - def test_link_team_command(self): - """Test that we successfully link a team to a Slack channel""" - assert "Link your Sentry team to this Slack channel!" in self.data["text"] - linking_url = build_team_linking_url( - self.integration, - "UXXXXXXX1", - "CXXXXXXX9", - "general", - "http://example.slack.com/response_url", - ) - - resp = self.client.get(linking_url) - assert resp.status_code == 200 - self.assertTemplateUsed(resp, "sentry/integrations/slack/link-team.html") - - data = urlencode({"team": self.team.id}) - resp = self.client.post(linking_url, data, content_type="application/x-www-form-urlencoded") - assert resp.status_code == 200 - self.assertTemplateUsed(resp, "sentry/integrations/slack/post-linked-team.html") - - assert len(self.external_actor) == 1 - assert self.external_actor[0].actor_id == self.team.actor_id - - assert len(responses.calls) >= 1 - data = json.loads(str(responses.calls[0].request.body.decode("utf-8"))) - assert ( - f"The {self.team.slug} team will now receive issue alert notifications in the {self.external_actor[0].external_name} channel." - in data["text"] - ) - - team_settings = NotificationSetting.objects.filter( - scope_type=NotificationScopeType.TEAM.value, target=self.team.actor.id - ) - assert len(team_settings) == 1 - - 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. - """ - ExternalActor.objects.create( - actor_id=self.team.actor_id, - organization=self.organization, - integration=self.integration, - provider=ExternalProviders.SLACK.value, - external_name="general", - external_id="CXXXXXXX9", - ) - - response = self.get_slack_response( - { - "text": "link team", - "team_id": self.external_id, - "user_id": "UXXXXXXX1", - "channel_name": "general", - "channel_id": "CXXXXXXX9", - } - ) - data = json.loads(str(response.content.decode("utf-8"))) - assert CHANNEL_ALREADY_LINKED_MESSAGE in data["text"] - - 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": "UXXXXXXX2", - "channel_name": "directmessage", - } - ) - data = json.loads(str(response.content.decode("utf-8"))) - assert LINK_FROM_CHANNEL_MESSAGE in data["text"] - - 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="UXXXXXXX2") - assert LINK_USER_FIRST_MESSAGE in data["text"] - - @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) - Identity.objects.create( - external_id="UXXXXXXX2", - idp=self.idp, - user=user2, - status=IdentityStatus.VALID, - scopes=[], - ) - data = self.send_slack_message("link team", user_id="UXXXXXXX2") - assert INSUFFICIENT_ROLE_MESSAGE in data["text"] - assert len(self.external_actor) == 0 - - @responses.activate - def test_link_team_already_linked(self): - """Test that if a team has already been linked to a Slack channel when a user tries - to link them again, we reject the attempt and reply with the ALREADY_LINKED_MESSAGE""" - ExternalActor.objects.create( - actor_id=self.team.actor_id, - organization=self.organization, - integration=self.integration, - provider=ExternalProviders.SLACK.value, - external_name="general", - external_id="CXXXXXXX9", - ) - assert "Link your Sentry team to this Slack channel!" in self.data["text"] - linking_url = build_team_linking_url( - self.integration, - "UXXXXXXX1", - "CXXXXXXX9", - "general", - "http://example.slack.com/response_url", - ) - - resp = self.client.get(linking_url) - - assert resp.status_code == 200 - self.assertTemplateUsed(resp, "sentry/integrations/slack/link-team.html") - - data = urlencode({"team": self.team.id}) - resp = self.client.post(linking_url, data, content_type="application/x-www-form-urlencoded") - assert resp.status_code == 200 - self.assertTemplateUsed(resp, "sentry/integrations/slack/post-linked-team.html") - assert len(responses.calls) >= 1 - data = json.loads(str(responses.calls[0].request.body.decode("utf-8"))) - assert ( - f"The {self.team.slug} team has already been linked to a Slack channel." in data["text"] - ) - - def test_error_page(self): - """Test that we successfully render an error page when bad form data is sent.""" - assert "Link your Sentry team to this Slack channel!" in self.data["text"] - linking_url = build_team_linking_url( - self.integration, - "UXXXXXXX1", - "CXXXXXXX9", - "general", - "http://example.slack.com/response_url", - ) - - resp = self.client.get(linking_url) - assert resp.status_code == 200 - self.assertTemplateUsed(resp, "sentry/integrations/slack/link-team.html") - - data = urlencode({"team": ["some", "garbage"]}) - resp = self.client.post(linking_url, data, content_type="application/x-www-form-urlencoded") - assert resp.status_code == 200 - self.assertTemplateUsed(resp, "sentry/integrations/slack/link-team-error.html") - - -class SlackCommandsUnlinkTeamTest(SlackCommandsTest): - def setUp(self): - super().setUp() - self.link_user() - self.link_team() - self.external_actor = ExternalActor.objects.filter( - actor_id=self.team.actor_id, - organization=self.organization, - integration=self.integration, - provider=ExternalProviders.SLACK.value, - external_name="general", - external_id="CXXXXXXX9", - ) - responses.add( - method=responses.POST, - url="https://slack.com/api/chat.postMessage", - body='{"ok": true}', - status=200, - content_type="application/json", - ) - self.team_unlinking_url = build_team_unlinking_url( - self.integration, - self.organization.id, - "UXXXXXXX1", - "CXXXXXXX9", - "general", - "http://example.slack.com/response_url", - ) - - @responses.activate - def test_unlink_team(self): - """Test that a team can be unlinked from a Slack channel""" - data = self.send_slack_message( - "unlink team", - channel_name="general", - channel_id="CXXXXXXX9", - ) - assert "Click here to unlink your team from this channel." in data["text"] - - resp = self.client.get(self.team_unlinking_url) - assert resp.status_code == 200 - self.assertTemplateUsed(resp, "sentry/integrations/slack/unlink-team.html") - - resp = self.client.post(self.team_unlinking_url) - assert resp.status_code == 200 - self.assertTemplateUsed(resp, "sentry/integrations/slack/unlinked-team.html") - - assert len(self.external_actor) == 0 - - assert len(responses.calls) >= 1 - data = json.loads(str(responses.calls[0].request.body.decode("utf-8"))) - assert ( - f"This channel will no longer receive issue alert notifications for the {self.team.slug} team." - in data["text"] - ) - - team_settings = NotificationSetting.objects.filter( - scope_type=NotificationScopeType.TEAM.value, target=self.team.actor.id - ) - assert len(team_settings) == 0 - - 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="CXXXXXXX8", - ) - assert TEAM_NOT_LINKED_MESSAGE in data["text"] - - @responses.activate - def test_unlink_multiple_teams(self): - """ - Test that if you have linked multiple teams to a single channel, when - you type `/sentry unlink team`, we unlink all teams from that channel. - This should only apply to the one organization who did this before we - blocked users from doing so. - """ - team2 = self.create_team(organization=self.organization, name="Team Hellboy") - ExternalActor.objects.create( - actor_id=team2.actor_id, - organization=self.organization, - integration=self.integration, - provider=ExternalProviders.SLACK.value, - external_name="general", - external_id="CXXXXXXX9", - ) - assert ( - ExternalActor.objects.filter( - actor_id__in=[self.team.actor_id, team2.actor_id], - organization=self.organization, - integration=self.integration, - provider=ExternalProviders.SLACK.value, - external_name="general", - external_id="CXXXXXXX9", - ).count() - == 2 - ) - - data = self.send_slack_message( - "unlink team", - channel_name="general", - channel_id="CXXXXXXX9", - ) - assert "Click here to unlink your team from this channel." in data["text"] - - resp = self.client.get(self.team_unlinking_url) - assert resp.status_code == 200 - self.assertTemplateUsed(resp, "sentry/integrations/slack/unlink-team.html") - - resp = self.client.post(self.team_unlinking_url) - assert resp.status_code == 200 - self.assertTemplateUsed(resp, "sentry/integrations/slack/unlinked-team.html") - - assert ( - ExternalActor.objects.filter( - actor_id__in=[self.team.actor_id, team2.actor_id], - organization=self.organization, - integration=self.integration, - provider=ExternalProviders.SLACK.value, - external_name="general", - external_id="CXXXXXXX9", - ).count() - == 0 - ) - - assert len(responses.calls) >= 1 - data = json.loads(str(responses.calls[0].request.body.decode("utf-8"))) - assert ( - f"This channel will no longer receive issue alert notifications for the {self.team.slug} team." - in data["text"] - ) - - team_settings = NotificationSetting.objects.filter( - scope_type=NotificationScopeType.TEAM.value, target=self.team.actor.id - ) - assert len(team_settings) == 0 diff --git a/tests/sentry/integrations/slack/endpoints/test_event.py b/tests/sentry/integrations/slack/endpoints/test_event.py index fd2210fd468ecb..c91d757311234f 100644 --- a/tests/sentry/integrations/slack/endpoints/test_event.py +++ b/tests/sentry/integrations/slack/endpoints/test_event.py @@ -8,7 +8,7 @@ from sentry.testutils import APITestCase from sentry.utils import json from sentry.utils.compat.mock import Mock, patch -from tests.sentry.integrations.slack import install_slack +from tests.sentry.integrations.slack import get_response_text, install_slack UNSET = object() @@ -288,7 +288,7 @@ def test_user_message_link(self): request = responses.calls[0].request assert request.headers["Authorization"] == "Bearer xoxb-xxxxxxxxx-xxxxxxxxxx-xxxxxxxxxxxx" data = json.loads(request.body) - assert "Link your Slack identity" in data["text"] + assert "Link your Slack identity" in get_response_text(data) @responses.activate def test_user_message_already_linked(self): @@ -310,7 +310,7 @@ def test_user_message_already_linked(self): request = responses.calls[0].request assert request.headers["Authorization"] == "Bearer xoxb-xxxxxxxxx-xxxxxxxxxx-xxxxxxxxxxxx" data = json.loads(request.body) - assert "You are already linked" in data["text"] + assert "You are already linked" in get_response_text(data) @responses.activate def test_user_message_unlink(self): @@ -332,7 +332,7 @@ def test_user_message_unlink(self): request = responses.calls[0].request assert request.headers["Authorization"] == "Bearer xoxb-xxxxxxxxx-xxxxxxxxxx-xxxxxxxxxxxx" data = json.loads(request.body) - assert "Click here to unlink your identity" in data["text"] + assert "Click here to unlink your identity" in get_response_text(data) @responses.activate def test_user_message_already_unlinked(self): @@ -347,7 +347,7 @@ def test_user_message_already_unlinked(self): request = responses.calls[0].request assert request.headers["Authorization"] == "Bearer xoxb-xxxxxxxxx-xxxxxxxxxx-xxxxxxxxxxxx" data = json.loads(request.body) - assert "You do not have a linked identity to unlink" in data["text"] + assert "You do not have a linked identity to unlink" in get_response_text(data) def test_bot_message_im(self): resp = self.post_webhook(event_data=json.loads(MESSAGE_IM_BOT_EVENT)) diff --git a/tests/sentry/integrations/slack/test_link_identity.py b/tests/sentry/integrations/slack/test_link_identity.py index 73ada862b37d69..6bbbbb69a45691 100644 --- a/tests/sentry/integrations/slack/test_link_identity.py +++ b/tests/sentry/integrations/slack/test_link_identity.py @@ -31,6 +31,7 @@ def setUp(self): class SlackIntegrationLinkIdentityTest(SlackIntegrationLinkIdentityTestBase): @responses.activate def test_basic_flow(self): + """Do the auth flow and assert that the identity was created.""" linking_url = build_linking_url( self.integration, self.external_id, self.channel_id, self.response_url ) @@ -92,7 +93,10 @@ def test_basic_flow(self): self.assertTemplateUsed(response, "sentry/auth-unlink-identity.html") # Unlink identity of user. - self.client.post(self.unlinking_url) + response = self.client.post(self.unlinking_url) + assert response.status_code == 200 + self.assertTemplateUsed(response, "sentry/integrations/slack/unlinked.html") + assert not Identity.objects.filter(external_id="new-slack-id", user=self.user).exists() assert len(responses.calls) == 1 diff --git a/tests/sentry/integrations/slack/test_link_team.py b/tests/sentry/integrations/slack/test_link_team.py new file mode 100644 index 00000000000000..3235d0d59a2ef5 --- /dev/null +++ b/tests/sentry/integrations/slack/test_link_team.py @@ -0,0 +1,254 @@ +from typing import Any, Mapping, Optional, Sequence +from urllib.parse import urlencode + +import responses +from django.db.models import QuerySet +from requests import Response +from rest_framework import status + +from sentry.integrations.slack.views.link_team import build_team_linking_url +from sentry.integrations.slack.views.unlink_team import build_team_unlinking_url +from sentry.models import ( + ExternalActor, + NotificationSetting, + Organization, + OrganizationIntegration, + Team, +) +from sentry.notifications.types import NotificationScopeType +from sentry.testutils import TestCase +from sentry.types.integrations import ExternalProviders +from sentry.utils import json +from tests.sentry.integrations.slack import ( + add_identity, + get_response_text, + install_slack, + link_team, +) + + +class SlackIntegrationLinkTeamTestBase(TestCase): + def setUp(self): + super().setUp() + self.login_as(self.user) + + 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.idp = add_identity(self.integration, self.user, self.external_id) + + responses.add( + method=responses.POST, + url=self.response_url, + body='{"ok": true}', + status=status.HTTP_200_OK, + content_type="application/json", + ) + responses.add( + method=responses.POST, + url="https://slack.com/api/chat.postMessage", + body='{"ok": true}', + status=status.HTTP_200_OK, + content_type="application/json", + ) + + def get_success_response(self, data: Optional[Mapping[str, Any]] = None) -> Response: + """This isn't in APITestCase so this isn't really an override.""" + kwargs = dict(content_type="application/x-www-form-urlencoded") + + if data is not None: + response = self.client.post(self.url, urlencode(data), **kwargs) + else: + response = self.client.get(self.url, **kwargs) + assert response.status_code == status.HTTP_200_OK + return response + + 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_linked_teams( + self, actor_ids: Optional[Sequence[int]] = None, organization: Optional[Organization] = None + ) -> QuerySet: + actor_ids = actor_ids or [self.team.actor_id] + organization = organization or self.organization + return ExternalActor.objects.filter( + actor_id__in=actor_ids, + organization=organization, + integration=self.integration, + provider=ExternalProviders.SLACK.value, + external_name=self.channel_name, + external_id=self.channel_id, + ) + + +class SlackIntegrationLinkTeamTest(SlackIntegrationLinkTeamTestBase): + def setUp(self): + super().setUp() + self.url = build_team_linking_url( + integration=self.integration, + slack_id=self.external_id, + channel_id=self.channel_id, + channel_name=self.channel_name, + response_url=self.response_url, + ) + + @responses.activate + def test_link_team(self): + """Test that we successfully link a team to a Slack channel""" + response = self.get_success_response() + self.assertTemplateUsed(response, "sentry/integrations/slack/link-team.html") + + response = self.get_success_response(data={"team": self.team.id}) + self.assertTemplateUsed(response, "sentry/integrations/slack/post-linked-team.html") + + external_actors = self.get_linked_teams() + assert len(external_actors) == 1 + assert external_actors[0].actor_id == self.team.actor_id + + assert len(responses.calls) >= 1 + data = json.loads(str(responses.calls[0].request.body.decode("utf-8"))) + assert ( + f"The {self.team.slug} team will now receive issue alert notifications in the {external_actors[0].external_name} channel." + in get_response_text(data) + ) + + team_settings = NotificationSetting.objects.filter( + scope_type=NotificationScopeType.TEAM.value, target=self.team.actor.id + ) + assert len(team_settings) == 1 + + @responses.activate + def test_link_team_already_linked(self): + """Test that if a team has already been linked to a Slack channel when a user tries + to link them again, we reject the attempt and reply with the ALREADY_LINKED_MESSAGE""" + self.link_team() + + response = self.get_success_response(data={"team": self.team.id}) + self.assertTemplateUsed(response, "sentry/integrations/slack/post-linked-team.html") + assert len(responses.calls) >= 1 + data = json.loads(str(responses.calls[0].request.body.decode("utf-8"))) + assert ( + f"The {self.team.slug} team has already been linked to a Slack channel." + in get_response_text(data) + ) + + def test_error_page(self): + """Test that we successfully render an error page when bad form data is sent.""" + response = self.get_success_response(data={"team": ["some", "garbage"]}) + self.assertTemplateUsed(response, "sentry/integrations/slack/link-team-error.html") + + +class SlackIntegrationUnlinkTeamTest(SlackIntegrationLinkTeamTestBase): + def setUp(self): + super().setUp() + + self.link_team() + self.url = build_team_unlinking_url( + integration=self.integration, + organization_id=self.organization.id, + slack_id=self.external_id, + channel_id=self.channel_id, + channel_name=self.channel_name, + response_url=self.response_url, + ) + + @responses.activate + def test_unlink_team(self): + """Test that a team can be unlinked from a Slack channel""" + response = self.get_success_response() + self.assertTemplateUsed(response, "sentry/integrations/slack/unlink-team.html") + + response = self.get_success_response(data={}) + self.assertTemplateUsed(response, "sentry/integrations/slack/unlinked-team.html") + + external_actors = self.get_linked_teams() + assert len(external_actors) == 0 + + assert len(responses.calls) >= 1 + data = json.loads(str(responses.calls[0].request.body.decode("utf-8"))) + assert ( + f"This channel will no longer receive issue alert notifications for the {self.team.slug} team." + in get_response_text(data) + ) + + team_settings = NotificationSetting.objects.filter( + scope_type=NotificationScopeType.TEAM.value, target=self.team.actor.id + ) + assert len(team_settings) == 0 + + @responses.activate + def test_unlink_multiple_teams(self): + """ + Test that if you have linked multiple teams to a single channel, when + you type `/sentry unlink team`, we unlink all teams from that channel. + This should only apply to the one organization who did this before we + blocked users from doing so. + """ + team2 = self.create_team(organization=self.organization, name="Team Hellboy") + self.link_team(team2) + + external_actors = self.get_linked_teams([self.team.actor_id, team2.actor_id]) + assert len(external_actors) == 2 + + response = self.get_success_response() + self.assertTemplateUsed(response, "sentry/integrations/slack/unlink-team.html") + + response = self.get_success_response(data={}) + self.assertTemplateUsed(response, "sentry/integrations/slack/unlinked-team.html") + + external_actors = self.get_linked_teams([self.team.actor_id, team2.actor_id]) + assert len(external_actors) == 0 + + assert len(responses.calls) >= 1 + data = json.loads(str(responses.calls[0].request.body.decode("utf-8"))) + assert ( + f"This channel will no longer receive issue alert notifications for the {self.team.slug} team." + in get_response_text(data) + ) + + team_settings = NotificationSetting.objects.filter( + scope_type=NotificationScopeType.TEAM.value, target=self.team.actor.id + ) + assert len(team_settings) == 0 + + @responses.activate + def test_unlink_team_multiple_organizations(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) + + # Team order should not matter. + for team in (self.team, team2): + external_actors = self.get_linked_teams( + organization=team.organization, actor_ids=[team.actor_id] + ) + assert len(external_actors) == 1 + + # Override the URL. + self.url = build_team_unlinking_url( + integration=self.integration, + organization_id=team.organization.id, + slack_id=self.external_id, + channel_id=self.channel_id, + channel_name=self.channel_name, + response_url=self.response_url, + ) + response = self.get_success_response(data={}) + self.assertTemplateUsed(response, "sentry/integrations/slack/unlinked-team.html") + + external_actors = self.get_linked_teams( + organization=team.organization, actor_ids=[team.actor_id] + ) + assert len(external_actors) == 0