sentry/tests/snuba/api/endpoints/test_project_event_details.py

from django.urls import reverse

from sentry.testutils import APITestCase, SnubaTestCase
from sentry.testutils.helpers.datetime import before_now, iso_format
from sentry.testutils.silo import region_silo_test
from sentry.types.issues import GroupType


@region_silo_test
class ProjectEventDetailsTest(APITestCase, SnubaTestCase):
    def setUp(self):
        super().setUp()
        self.login_as(user=self.user)
        project = self.create_project()

        one_min_ago = iso_format(before_now(minutes=1))
        two_min_ago = iso_format(before_now(minutes=2))
        three_min_ago = iso_format(before_now(minutes=3))
        four_min_ago = iso_format(before_now(minutes=4))

        self.prev_event = self.store_event(
            data={"event_id": "a" * 32, "timestamp": four_min_ago, "fingerprint": ["group-1"]},
            project_id=project.id,
        )
        self.cur_event = self.store_event(
            data={"event_id": "b" * 32, "timestamp": three_min_ago, "fingerprint": ["group-1"]},
            project_id=project.id,
        )
        self.next_event = self.store_event(
            data={
                "event_id": "c" * 32,
                "timestamp": two_min_ago,
                "fingerprint": ["group-1"],
                "environment": "production",
                "tags": {"environment": "production"},
            },
            project_id=project.id,
        )

        # Event in different group
        self.store_event(
            data={
                "event_id": "d" * 32,
                "timestamp": one_min_ago,
                "fingerprint": ["group-2"],
                "environment": "production",
                "tags": {"environment": "production"},
            },
            project_id=project.id,
        )

    def test_simple(self):
        url = reverse(
            "sentry-api-0-project-event-details",
            kwargs={
                "event_id": self.cur_event.event_id,
                "project_slug": self.cur_event.project.slug,
                "organization_slug": self.cur_event.project.organization.slug,
            },
        )
        response = self.client.get(url, format="json")

        assert response.status_code == 200, response.content
        assert response.data["id"] == str(self.cur_event.event_id)
        assert response.data["nextEventID"] == str(self.next_event.event_id)
        assert response.data["previousEventID"] == str(self.prev_event.event_id)
        assert response.data["groupID"] == str(self.cur_event.group.id)

    def test_snuba_no_prev(self):
        url = reverse(
            "sentry-api-0-project-event-details",
            kwargs={
                "event_id": self.prev_event.event_id,
                "project_slug": self.prev_event.project.slug,
                "organization_slug": self.prev_event.project.organization.slug,
            },
        )
        response = self.client.get(url, format="json")

        assert response.status_code == 200, response.content
        assert response.data["id"] == str(self.prev_event.event_id)
        assert response.data["previousEventID"] is None
        assert response.data["nextEventID"] == self.cur_event.event_id
        assert response.data["groupID"] == str(self.prev_event.group.id)

    def test_snuba_with_environment(self):
        url = reverse(
            "sentry-api-0-project-event-details",
            kwargs={
                "event_id": self.cur_event.event_id,
                "project_slug": self.cur_event.project.slug,
                "organization_slug": self.cur_event.project.organization.slug,
            },
        )
        response = self.client.get(
            url, format="json", data={"environment": ["production", "staging"]}
        )

        assert response.status_code == 200, response.content
        assert response.data["id"] == str(self.cur_event.event_id)
        assert response.data["previousEventID"] is None
        assert response.data["nextEventID"] == self.next_event.event_id
        assert response.data["groupID"] == str(self.prev_event.group.id)

    def test_ignores_different_group(self):
        url = reverse(
            "sentry-api-0-project-event-details",
            kwargs={
                "event_id": self.next_event.event_id,
                "project_slug": self.next_event.project.slug,
                "organization_slug": self.next_event.project.organization.slug,
            },
        )
        response = self.client.get(url, format="json")

        assert response.status_code == 200, response.content
        assert response.data["id"] == str(self.next_event.event_id)
        assert response.data["nextEventID"] is None


@region_silo_test
class ProjectEventDetailsTransactionTest(APITestCase, SnubaTestCase):
    def setUp(self):
        super().setUp()
        self.login_as(user=self.user)
        project = self.create_project()

        one_min_ago = iso_format(before_now(minutes=1))
        two_min_ago = iso_format(before_now(minutes=2))
        three_min_ago = iso_format(before_now(minutes=3))
        four_min_ago = iso_format(before_now(minutes=4))

        transaction_event_data = {
            "level": "info",
            "message": "ayoo",
            "type": "transaction",
            "culprit": "app/components/events/eventEntries in map",
            "contexts": {"trace": {"trace_id": "b" * 32, "span_id": "c" * 16, "op": ""}},
        }

        self.prev_transaction_event = self.store_event(
            data={
                **transaction_event_data,
                "event_id": "a" * 32,
                "timestamp": four_min_ago,
                "start_timestamp": four_min_ago,
                "fingerprint": [f"{GroupType.PERFORMANCE_RENDER_BLOCKING_ASSET_SPAN.value}-group1"],
            },
            project_id=project.id,
        )

        self.cur_transaction_event = self.store_event(
            data={
                **transaction_event_data,
                "event_id": "b" * 32,
                "timestamp": three_min_ago,
                "start_timestamp": three_min_ago,
                "fingerprint": [f"{GroupType.PERFORMANCE_RENDER_BLOCKING_ASSET_SPAN.value}-group1"],
            },
            project_id=project.id,
        )

        self.next_transaction_event = self.store_event(
            data={
                **transaction_event_data,
                "event_id": "c" * 32,
                "timestamp": two_min_ago,
                "start_timestamp": two_min_ago,
                "environment": "production",
                "tags": {"environment": "production"},
                "fingerprint": [f"{GroupType.PERFORMANCE_RENDER_BLOCKING_ASSET_SPAN.value}-group1"],
            },
            project_id=project.id,
        )

        self.group = self.prev_transaction_event.groups[0]

        # Event in different group
        self.store_event(
            data={
                **transaction_event_data,
                "event_id": "d" * 32,
                "timestamp": one_min_ago,
                "start_timestamp": one_min_ago,
                "environment": "production",
                "tags": {"environment": "production"},
            },
            project_id=project.id,
        )

    def test_transaction_event(self):
        """Test that you can look up a transaction event w/ a prev and next event"""
        url = reverse(
            "sentry-api-0-project-event-details",
            kwargs={
                "event_id": self.cur_transaction_event.event_id,
                "project_slug": self.cur_transaction_event.project.slug,
                "organization_slug": self.cur_transaction_event.project.organization.slug,
            },
        )
        response = self.client.get(url, format="json", data={"group_id": self.group.id})

        assert response.status_code == 200, response.content
        assert response.data["id"] == str(self.cur_transaction_event.event_id)
        assert response.data["nextEventID"] == str(self.next_transaction_event.event_id)
        assert response.data["previousEventID"] == str(self.prev_transaction_event.event_id)
        assert response.data["groupID"] == str(self.cur_transaction_event.groups[0].id)

    def test_no_previous_event(self):
        """Test the case in which there is no previous event"""
        url = reverse(
            "sentry-api-0-project-event-details",
            kwargs={
                "event_id": self.prev_transaction_event.event_id,
                "project_slug": self.prev_transaction_event.project.slug,
                "organization_slug": self.prev_transaction_event.project.organization.slug,
            },
        )
        response = self.client.get(url, format="json", data={"group_id": self.group.id})

        assert response.status_code == 200, response.content
        assert response.data["id"] == str(self.prev_transaction_event.event_id)
        assert response.data["previousEventID"] is None
        assert response.data["nextEventID"] == self.cur_transaction_event.event_id
        assert response.data["groupID"] == str(self.prev_transaction_event.groups[0].id)

    def test_ignores_different_group(self):
        """Test that a different group's events aren't attributed to the one that was passed"""
        url = reverse(
            "sentry-api-0-project-event-details",
            kwargs={
                "event_id": self.next_transaction_event.event_id,
                "project_slug": self.next_transaction_event.project.slug,
                "organization_slug": self.next_transaction_event.project.organization.slug,
            },
        )
        response = self.client.get(url, format="json", data={"group_id": self.group.id})

        assert response.status_code == 200, response.content
        assert response.data["id"] == str(self.next_transaction_event.event_id)
        assert response.data["nextEventID"] is None

    def test_no_group_id(self):
        """Test the case where a group_id was not passed"""
        url = reverse(
            "sentry-api-0-project-event-details",
            kwargs={
                "event_id": self.cur_transaction_event.event_id,
                "project_slug": self.cur_transaction_event.project.slug,
                "organization_slug": self.cur_transaction_event.project.organization.slug,
            },
        )
        response = self.client.get(url, format="json")

        assert response.status_code == 200, response.content
        assert response.data["id"] == str(self.cur_transaction_event.event_id)
        assert response.data["previousEventID"] is None
        assert response.data["nextEventID"] is None
        assert response.data["groupID"] is None


@region_silo_test
class ProjectEventJsonEndpointTest(APITestCase, SnubaTestCase):
    def setUp(self):
        super().setUp()
        self.login_as(user=self.user)
        self.event_id = "c" * 32
        self.fingerprint = ["group_2"]
        self.min_ago = iso_format(before_now(minutes=1))
        self.event = self.store_event(
            data={
                "event_id": self.event_id,
                "timestamp": self.min_ago,
                "fingerprint": self.fingerprint,
                "user": {"email": self.user.email},
            },
            project_id=self.project.id,
        )
        self.url = reverse(
            "sentry-api-0-event-json",
            kwargs={
                "organization_slug": self.organization.slug,
                "project_slug": self.project.slug,
                "event_id": self.event_id,
            },
        )

    def assert_event(self, data):
        assert data["event_id"] == self.event_id
        assert data["user"]["email"] == self.user.email
        assert data["datetime"][:19] == self.min_ago
        assert data["fingerprint"] == self.fingerprint

    def test_simple(self):
        response = self.client.get(self.url, format="json")
        assert response.status_code == 200, response.content
        self.assert_event(response.data)

    def test_event_does_not_exist(self):
        self.url = reverse(
            "sentry-api-0-event-json",
            kwargs={
                "organization_slug": self.organization.slug,
                "project_slug": self.project.slug,
                "event_id": "no" * 16,
            },
        )
        response = self.client.get(self.url, format="json")
        assert response.status_code == 404, response.content
        assert response.data == {"detail": "Event not found"}

    def test_user_unauthorized(self):
        user = self.create_user()
        self.login_as(user)

        response = self.client.get(self.url, format="json")
        assert response.status_code == 403, response.content
        assert response.data == {"detail": "You do not have permission to perform this action."}

    def test_project_not_associated_with_event(self):
        project2 = self.create_project(organization=self.organization)
        url = reverse(
            "sentry-api-0-event-json",
            kwargs={
                "organization_slug": self.organization.slug,
                "project_slug": project2.slug,
                "event_id": self.event_id,
            },
        )
        response = self.client.get(url, format="json")
        assert response.status_code == 404, response.content
        assert response.data == {"detail": "Event not found"}