from __future__ import annotations
from typing import Any, Protocol
from django.http.response import HttpResponse
from django.urls import reverse
from sentry.discover.models import MAX_TEAM_KEY_TRANSACTIONS, TeamKeyTransaction
from sentry.models.projectteam import ProjectTeam
from sentry.testutils.cases import APITestCase, SnubaTestCase
from sentry.testutils.helpers import parse_link_header
from sentry.testutils.helpers.pagination import override_pagination_limit
from sentry.utils.samples import load_data
class TeamKeyTransactionTestBase(APITestCase, SnubaTestCase):
def setUp(self):
super().setUp()
self.login_as(user=self.user, superuser=False)
self.org = self.create_organization(owner=self.user, name="foo")
self.project = self.create_project(name="baz", organization=self.org)
self.event_data = load_data("transaction")
self.features = ["organizations:performance-view"]
class ClientCallable(Protocol):
def __call__(self, url: str, data: dict[str, Any], format: str, **kwargs: Any) -> HttpResponse:
...
class TeamKeyTransactionTest(TeamKeyTransactionTestBase):
def setUp(self):
super().setUp()
self.url = reverse("sentry-api-0-organization-key-transactions", args=[self.org.slug])
def test_key_transaction_without_feature(self):
project = self.create_project(name="qux", organization=self.org)
data = {
"project": [self.project.id, project.id],
"transaction": self.event_data["transaction"],
"team": "myteams",
}
for response in (
self.client.get(self.url, data=data, format="json"),
self.client.post(self.url, data=data, format="json"),
self.client.delete(self.url, data=data, format="json"),
):
assert response.status_code == 404, response.content
def test_get_key_transaction_multiple_projects(self):
project = self.create_project(name="qux", organization=self.org)
with self.feature(self.features):
response = self.client.get(
self.url,
data={
"project": [self.project.id, project.id],
"transaction": self.event_data["transaction"],
},
format="json",
)
assert response.status_code == 400, response.content
assert response.data == {"detail": "Only 1 project per Key Transaction"}
def test_get_key_transaction_no_transaction_name(self):
with self.feature(self.features):
response = self.client.get(
self.url,
data={
"project": [self.project.id],
},
format="json",
)
assert response.status_code == 400, response.content
assert response.data == {"detail": "A transaction name is required"}
def test_get_no_key_transaction(self):
with self.feature(self.features):
response = self.client.get(
self.url,
data={
"project": [self.project.id],
"transaction": self.event_data["transaction"],
},
format="json",
)
assert response.status_code == 200, response.content
assert response.data == []
def test_get_key_transaction_my_teams(self):
team1 = self.create_team(organization=self.org, name="Team A")
team2 = self.create_team(organization=self.org, name="Team B")
team3 = self.create_team(organization=self.org, name="Team C")
# should not be in response because we never joined this team
self.create_team(organization=self.org, name="Team D")
# only join teams 1,2,3
for team in [team1, team2, team3]:
self.create_team_membership(team, user=self.user)
self.project.add_team(team)
TeamKeyTransaction.objects.bulk_create(
[
TeamKeyTransaction(
organization=self.org,
project_team=project_team,
transaction=self.event_data["transaction"],
)
for project_team in ProjectTeam.objects.filter(
project=self.project, team__in=[team1, team2]
)
]
+ [
TeamKeyTransaction(
organization=self.org,
project_team=project_team,
transaction="other-transaction",
)
for project_team in ProjectTeam.objects.filter(
project=self.project, team__in=[team2, team3]
)
]
)
with self.feature(self.features):
response = self.client.get(
self.url,
data={
"project": [self.project.id],
"transaction": self.event_data["transaction"],
"team": "myteams",
},
format="json",
)
assert response.status_code == 200, response.content
assert response.data == [
{
"team": str(team1.id),
},
{
"team": str(team2.id),
},
]
def test_post_key_transaction_more_than_1_project(self):
team = self.create_team(organization=self.org, name="Team Foo")
self.create_team_membership(team, user=self.user)
self.project.add_team(team)
project = self.create_project(name="bar", organization=self.org)
project.add_team(team)
with self.feature(self.features):
response = self.client.post(
self.url,
data={
"project": [self.project.id, project.id],
"transaction": self.event_data["transaction"],
"team": [team.id],
},
format="json",
)
assert response.status_code == 400, response.content
assert response.data == {"detail": "Only 1 project per Key Transaction"}
def test_post_key_transaction_no_team(self):
team = self.create_team(organization=self.org, name="Team Foo")
self.create_team_membership(team, user=self.user)
self.project.add_team(team)
with self.feature(self.features):
response = self.client.post(
self.url,
data={
"project": [self.project.id],
"transaction": self.event_data["transaction"],
},
format="json",
)
assert response.status_code == 400, response.content
assert response.data == {"team": ["This field is required."]}
def test_post_key_transaction_no_transaction_name(self):
team = self.create_team(organization=self.org, name="Team Foo")
self.create_team_membership(team, user=self.user)
self.project.add_team(team)
with self.feature(self.features):
response = self.client.post(
self.url,
data={
"project": [self.project.id],
"team": [team.id],
},
format="json",
)
assert response.status_code == 400, response.content
assert response.data == {"transaction": ["This field is required."]}
def test_post_key_transaction_no_access_team(self):
org = self.create_organization(
owner=self.user, # use other user as owner
name="foo",
flags=0, # disable default allow_joinleave
)
project = self.create_project(name="baz", organization=org)
user = self.create_user()
self.login_as(user=user, superuser=False)
team = self.create_team(organization=org, name="Team Foo")
self.create_team_membership(team, user=user)
project.add_team(team)
other_team = self.create_team(organization=org, name="Team Bar")
project.add_team(other_team)
with self.feature(self.features):
response = self.client.post(
reverse("sentry-api-0-organization-key-transactions", args=[org.slug]),
data={
"project": [project.id],
"transaction": self.event_data["transaction"],
"team": [other_team.id],
},
format="json",
)
assert response.status_code == 400, response.content
assert response.data == {
"team": [f"You do not have permission to access {other_team.name}"]
}
def test_post_key_transaction_no_access_project(self):
team1 = self.create_team(organization=self.org, name="Team Foo")
self.create_team_membership(team1, user=self.user)
self.project.add_team(team1)
team2 = self.create_team(organization=self.org, name="Team Bar")
self.create_team_membership(team2, user=self.user)
with self.feature(self.features):
response = self.client.post(
self.url,
data={
"project": [self.project.id],
"transaction": self.event_data["transaction"],
"team": [team2.id],
},
format="json",
)
assert response.status_code == 400, response.content
assert response.data == {"detail": "Team does not have access to project"}
def test_post_key_transactions_exceed_limit(self):
team = self.create_team(organization=self.org, name="Team Foo")
self.create_team_membership(team, user=self.user)
self.project.add_team(team)
project_team = ProjectTeam.objects.get(project=self.project, team=team)
TeamKeyTransaction.objects.bulk_create(
[
TeamKeyTransaction(
organization=self.org,
project_team=project_team,
transaction=f"{self.event_data['transaction']}-{i}",
)
for i in range(MAX_TEAM_KEY_TRANSACTIONS)
]
)
with self.feature(self.features):
response = self.client.post(
self.url,
data={
"project": [self.project.id],
"transaction": self.event_data["transaction"],
"team": [team.id],
},
format="json",
)
assert response.status_code == 400, response.content
assert response.data == {
"non_field_errors": [
f"At most {MAX_TEAM_KEY_TRANSACTIONS} Key Transactions can be added for a team"
]
}
def test_post_key_transaction_limit_is_per_team(self):
team1 = self.create_team(organization=self.org, name="Team Foo")
self.create_team_membership(team1, user=self.user)
self.project.add_team(team1)
team2 = self.create_team(organization=self.org, name="Team Bar")
self.create_team_membership(team2, user=self.user)
self.project.add_team(team2)
project_teams = ProjectTeam.objects.filter(project=self.project, team__in=[team1, team2])
TeamKeyTransaction.objects.bulk_create(
[
TeamKeyTransaction(
organization=self.org,
project_team=project_team,
transaction=f"{self.event_data['transaction']}-{i}",
)
for project_team in project_teams
for i in range(MAX_TEAM_KEY_TRANSACTIONS - 1)
]
)
with self.feature(self.features):
response = self.client.post(
self.url,
data={
"project": [self.project.id],
"transaction": self.event_data["transaction"],
"team": [team1.id, team2.id],
},
format="json",
)
assert response.status_code == 201, response.content
key_transactions = TeamKeyTransaction.objects.filter(project_team__team__in=[team1, team2])
assert len(key_transactions) == 2 * MAX_TEAM_KEY_TRANSACTIONS
def test_post_key_transactions(self):
team = self.create_team(organization=self.org, name="Team Foo")
self.create_team_membership(team, user=self.user)
self.project.add_team(team)
with self.feature(self.features):
response = self.client.post(
self.url,
data={
"project": [self.project.id],
"transaction": self.event_data["transaction"],
"team": [team.id],
},
format="json",
)
assert response.status_code == 201, response.content
key_transactions = TeamKeyTransaction.objects.filter(project_team__team=team)
assert len(key_transactions) == 1
def test_post_key_transactions_duplicate(self):
team = self.create_team(organization=self.org, name="Team Foo")
self.create_team_membership(team, user=self.user)
self.project.add_team(team)
project_team = ProjectTeam.objects.get(project=self.project, team=team)
TeamKeyTransaction.objects.create(
organization=self.org,
project_team=project_team,
transaction=self.event_data["transaction"],
)
with self.feature(self.features):
response = self.client.post(
self.url,
data={
"project": [self.project.id],
"transaction": self.event_data["transaction"],
"team": [team.id],
},
format="json",
)
assert response.status_code == 204, response.content
key_transactions = TeamKeyTransaction.objects.filter(
project_team=project_team, transaction=self.event_data["transaction"]
)
assert len(key_transactions) == 1
def test_post_key_transaction_multiple_team(self):
team1 = self.create_team(organization=self.org, name="Team Foo")
self.create_team_membership(team1, user=self.user)
self.project.add_team(team1)
team2 = self.create_team(organization=self.org, name="Team Bar")
self.create_team_membership(team2, user=self.user)
self.project.add_team(team2)
with self.feature(self.features):
response = self.client.post(
self.url,
data={
"project": [self.project.id],
"transaction": self.event_data["transaction"],
"team": [team1.id, team2.id],
},
format="json",
)
assert response.status_code == 201, response.content
key_transactions = TeamKeyTransaction.objects.filter(
project_team__in=ProjectTeam.objects.filter(
project=self.project, team__in=[team1, team2]
),
transaction=self.event_data["transaction"],
)
assert len(key_transactions) == 2
def test_post_key_transaction_partially_existing_teams(self):
team1 = self.create_team(organization=self.org, name="Team Foo")
self.create_team_membership(team1, user=self.user)
self.project.add_team(team1)
team2 = self.create_team(organization=self.org, name="Team Bar")
self.create_team_membership(team2, user=self.user)
self.project.add_team(team2)
TeamKeyTransaction.objects.create(
organization=self.org,
project_team=ProjectTeam.objects.get(project=self.project, team=team1),
transaction=self.event_data["transaction"],
)
with self.feature(self.features):
response = self.client.post(
self.url,
data={
"project": [self.project.id],
"transaction": self.event_data["transaction"],
"team": [team1.id, team2.id],
},
format="json",
)
assert response.status_code == 201, response.content
key_transactions = TeamKeyTransaction.objects.filter(
project_team__in=ProjectTeam.objects.filter(
project=self.project, team__in=[team1, team2]
),
transaction=self.event_data["transaction"],
)
assert len(key_transactions) == 2
def test_post_key_transaction_multiple_users(self):
team = self.create_team(organization=self.org, name="Team Foo")
self.create_team_membership(team, user=self.user)
self.project.add_team(team)
with self.feature(self.features):
response = self.client.post(
self.url,
data={
"project": [self.project.id],
"transaction": self.event_data["transaction"],
"team": [team.id],
},
format="json",
)
assert response.status_code == 201, response.content
with self.feature(self.features):
response = self.client.post(
self.url,
data={
"project": [self.project.id],
"transaction": self.event_data["transaction"],
"team": [team.id],
},
format="json",
)
# already created by this user, so it's 204 this time
assert response.status_code == 204, response.content
user = self.create_user()
self.create_member(user=user, organization=self.org, role="member")
self.create_team_membership(team, user=user)
self.login_as(user=user, superuser=False)
with self.feature(self.features):
response = self.client.post(
self.url,
data={
"project": [self.project.id],
"transaction": self.event_data["transaction"],
"team": [team.id],
},
format="json",
)
# already created by another user, so it's 204 this time
assert response.status_code == 204, response.content
# should only create 1 team key transaction
key_transactions = TeamKeyTransaction.objects.filter(project_team__team=team)
assert len(key_transactions) == 1
def test_post_key_transaction_overly_long_transaction(self):
team = self.create_team(organization=self.org, name="Team Foo")
self.create_team_membership(team, user=self.user)
self.project.add_team(team)
with self.feature(self.features):
response = self.client.post(
self.url,
data={
"project": [self.project.id],
"transaction": "a" * 500,
"team": [team.id],
},
format="json",
)
assert response.status_code == 400, response.content
assert response.data == {
"transaction": ["Ensure this field has no more than 200 characters."]
}
def test_delete_key_transaction_no_transaction_name(self):
team = self.create_team(organization=self.org, name="Team Foo")
self.create_team_membership(team, user=self.user)
self.project.add_team(team)
with self.feature(self.features):
response = self.client.delete(
self.url,
data={
"project": [self.project.id],
"team": [team.id],
},
format="json",
)
assert response.status_code == 400, response.content
assert response.data == {"transaction": ["This field is required."]}
def test_delete_key_transaction_no_team(self):
team = self.create_team(organization=self.org, name="Team Foo")
self.create_team_membership(team, user=self.user)
self.project.add_team(team)
with self.feature(self.features):
response = self.client.delete(
self.url,
data={
"project": [self.project.id],
"transaction": self.event_data["transaction"],
},
format="json",
)
assert response.status_code == 400, response.content
assert response.data == {"team": ["This field is required."]}
def test_delete_key_transactions_no_exist(self):
team = self.create_team(organization=self.org, name="Team Foo")
self.create_team_membership(team, user=self.user)
self.project.add_team(team)
with self.feature(self.features):
response = self.client.delete(
self.url,
data={
"project": [self.project.id],
"transaction": self.event_data["transaction"],
"team": [team.id],
},
format="json",
)
assert response.status_code == 204, response.content
key_transactions = TeamKeyTransaction.objects.filter(project_team__team=team)
assert len(key_transactions) == 0
def test_delete_key_transaction_no_access_team(self):
org = self.create_organization(
owner=self.user, # use other user as owner
name="foo",
flags=0, # disable default allow_joinleave
)
project = self.create_project(name="baz", organization=org)
user = self.create_user()
self.login_as(user=user, superuser=False)
team = self.create_team(organization=org, name="Team Foo")
self.create_team_membership(team, user=user)
project.add_team(team)
other_team = self.create_team(organization=org, name="Team Bar")
project.add_team(other_team)
TeamKeyTransaction.objects.create(
organization=org,
project_team=ProjectTeam.objects.get(project=project, team=team),
transaction=self.event_data["transaction"],
)
with self.feature(self.features):
response = self.client.delete(
reverse("sentry-api-0-organization-key-transactions", args=[org.slug]),
data={
"project": [project.id],
"transaction": self.event_data["transaction"],
"team": [other_team.id],
},
format="json",
)
assert response.status_code == 400, response.content
assert response.data == {
"team": [f"You do not have permission to access {other_team.name}"]
}
def test_delete_key_transactions(self):
team = self.create_team(organization=self.org, name="Team Foo")
self.create_team_membership(team, user=self.user)
self.project.add_team(team)
TeamKeyTransaction.objects.create(
organization=self.org,
project_team=ProjectTeam.objects.get(project=self.project, team=team),
transaction=self.event_data["transaction"],
)
with self.feature(self.features):
response = self.client.delete(
self.url,
data={
"project": [self.project.id],
"transaction": self.event_data["transaction"],
"team": [team.id],
},
format="json",
)
assert response.status_code == 204, response.content
key_transactions = TeamKeyTransaction.objects.filter(project_team__team=team)
assert len(key_transactions) == 0
def test_delete_key_transaction_partially_existing_teams(self):
team1 = self.create_team(organization=self.org, name="Team Foo")
self.create_team_membership(team1, user=self.user)
self.project.add_team(team1)
team2 = self.create_team(organization=self.org, name="Team Bar")
self.create_team_membership(team2, user=self.user)
self.project.add_team(team2)
TeamKeyTransaction.objects.create(
organization=self.org,
project_team=ProjectTeam.objects.get(project=self.project, team=team1),
transaction=self.event_data["transaction"],
)
with self.feature(self.features):
response = self.client.delete(
self.url,
data={
"project": [self.project.id],
"transaction": self.event_data["transaction"],
"team": [team1.id, team2.id],
},
format="json",
)
assert response.status_code == 204, response.content
class TeamKeyTransactionListTest(TeamKeyTransactionTestBase):
def setUp(self):
super().setUp()
self.url = reverse("sentry-api-0-organization-key-transactions-list", args=[self.org.slug])
self.team1 = self.create_team(organization=self.org, name="Team A")
self.team2 = self.create_team(organization=self.org, name="Team B")
self.team3 = self.create_team(organization=self.org, name="Team C")
self.team4 = self.create_team(organization=self.org, name="Team D")
self.team5 = self.create_team(organization=self.org, name="Team E")
for team in [self.team1, self.team2, self.team3, self.team4, self.team5]:
self.project.add_team(team)
# only join teams 1,2,3
for team in [self.team1, self.team2, self.team3]:
self.create_team_membership(team, user=self.user)
TeamKeyTransaction.objects.bulk_create(
[
TeamKeyTransaction(
organization=self.org,
project_team=project_team,
transaction=self.event_data["transaction"],
)
for project_team in ProjectTeam.objects.filter(
project=self.project, team__in=[self.team2, self.team3]
)
]
+ [
TeamKeyTransaction(
organization=self.org,
project_team=project_team,
transaction="other-transaction",
)
for project_team in ProjectTeam.objects.filter(
project=self.project, team__in=[self.team3, self.team4]
)
]
)
def test_get_key_transaction_list_no_permissions(self):
org = self.create_organization(
owner=self.user, # use other user as owner
name="foo",
flags=0, # disable default allow_joinleave
)
project = self.create_project(name="baz", organization=org)
user = self.create_user()
self.login_as(user=user, superuser=False)
team = self.create_team(organization=org, name="Team Foo")
self.create_team_membership(team, user=user)
project.add_team(team)
other_team = self.create_team(organization=org, name="Team Bar")
project.add_team(other_team)
with self.feature(self.features):
response = self.client.get(
reverse("sentry-api-0-organization-key-transactions-list", args=[org.slug]),
data={
"project": [self.project.id],
"team": ["myteams", other_team.id],
},
format="json",
)
assert response.status_code == 403, response.content
assert response.data == {
"detail": f"Error: You do not have permission to access {other_team.name}"
}
def test_get_key_transaction_list_my_teams(self):
with self.feature(self.features):
response = self.client.get(
self.url,
data={
"project": [self.project.id],
"team": ["myteams"],
},
format="json",
)
assert response.status_code == 200, response.content
assert response.data == [
{
"team": str(self.team1.id),
"count": 0,
"keyed": [],
},
{
"team": str(self.team2.id),
"count": 1,
"keyed": [
{
"project_id": str(self.project.id),
"transaction": self.event_data["transaction"],
},
],
},
{
"team": str(self.team3.id),
"count": 2,
"keyed": [
{
"project_id": str(self.project.id),
"transaction": self.event_data["transaction"],
},
{"project_id": str(self.project.id), "transaction": "other-transaction"},
],
},
]
def test_get_key_transaction_list_other_teams(self):
with self.feature(self.features):
response = self.client.get(
self.url,
data={
"project": [self.project.id],
"team": [self.team4.id, self.team5.id],
},
format="json",
)
assert response.status_code == 200, response.content
assert response.data == [
{
"team": str(self.team4.id),
"count": 1,
"keyed": [
{"project_id": str(self.project.id), "transaction": "other-transaction"},
],
},
{
"team": str(self.team5.id),
"count": 0,
"keyed": [],
},
]
def test_get_key_transaction_list_mixed_my_and_other_teams(self):
with self.feature(self.features):
response = self.client.get(
self.url,
data={
"project": [self.project.id],
"team": ["myteams", self.team4.id, self.team5.id],
},
format="json",
)
assert response.status_code == 200, response.content
assert response.data == [
{
"team": str(self.team1.id),
"count": 0,
"keyed": [],
},
{
"team": str(self.team2.id),
"count": 1,
"keyed": [
{
"project_id": str(self.project.id),
"transaction": self.event_data["transaction"],
},
],
},
{
"team": str(self.team3.id),
"count": 2,
"keyed": [
{
"project_id": str(self.project.id),
"transaction": self.event_data["transaction"],
},
{"project_id": str(self.project.id), "transaction": "other-transaction"},
],
},
{
"team": str(self.team4.id),
"count": 1,
"keyed": [
{"project_id": str(self.project.id), "transaction": "other-transaction"},
],
},
{
"team": str(self.team5.id),
"count": 0,
"keyed": [],
},
]
@override_pagination_limit(5)
def test_get_key_transaction_list_pagination(self):
user = self.create_user()
self.login_as(user=user)
org = self.create_organization(owner=user, name="foo")
project = self.create_project(name="baz", organization=org)
teams = []
for i in range(8):
team = self.create_team(organization=org, name=f"Team {i:02d}")
self.create_team_membership(team, user=user)
project.add_team(team)
teams.append(team)
# get the first page
with self.feature(self.features):
response = self.client.get(
reverse("sentry-api-0-organization-key-transactions-list", args=[org.slug]),
data={
"project": [project.id],
"team": ["myteams"],
},
format="json",
)
assert response.status_code == 200, response.content
assert len(response.data) == 5
links = {
link["rel"]: {"url": url, **link}
for url, link in parse_link_header(response["Link"]).items()
}
assert links["previous"]["results"] == "false"
assert links["next"]["results"] == "true"
# get the second page
with self.feature(self.features):
assert links["next"]["cursor"] is not None
response = self.client.get(
reverse("sentry-api-0-organization-key-transactions-list", args=[org.slug]),
data={
"project": [project.id],
"team": ["myteams"],
"cursor": links["next"]["cursor"],
},
format="json",
)
assert response.status_code == 200, response.content
assert len(response.data) == 3
links = {
link["rel"]: {"url": url, **link}
for url, link in parse_link_header(response["Link"]).items()
}
assert links["previous"]["results"] == "true"
assert links["next"]["results"] == "false"
def test_get_key_transaction_list_partial_project(self):
another_project = self.create_project(organization=self.org)
another_project.add_team(self.team2)
TeamKeyTransaction.objects.create(
organization=self.org,
project_team=ProjectTeam.objects.get(project=another_project, team=self.team2),
transaction="another-transaction",
)
with self.feature(self.features):
response = self.client.get(
self.url,
data={
"project": [another_project.id],
"team": [self.team2.id],
},
format="json",
)
assert response.status_code == 200, response.content
assert response.data == [
{
"team": str(self.team2.id),
# the key transaction in self.project is counted but not in
# the list because self.project is not in the project param
"count": 2,
"keyed": [
{
"project_id": str(another_project.id),
"transaction": "another-transaction",
},
],
},
]