Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
Expand Up @@ -74,7 +74,7 @@ def can_edit_detector_workflow_connections(detector: Detector, request: Request)
Anyone with alert write access to the project can connect/disconnect detectors of any type,
which is slightly different from full edit access which differs by detector type.
"""
return can_edit_user_created_detectors(request, detector.project)
return request.access.has_scope("alerts:write")


def validate_detectors_exist_and_have_permissions(
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -3,7 +3,6 @@

from sentry import audit_log
from sentry.deletions.tasks.scheduled import run_scheduled_deletions
from sentry.grouping.grouptype import ErrorGroupType
from sentry.testutils.cases import APITestCase
from sentry.testutils.silo import region_silo_test
from sentry.workflow_engine.models.detector_workflow import DetectorWorkflow
Expand Down Expand Up @@ -195,40 +194,10 @@ def test_missing_body_params(self) -> None:
self.organization.slug,
)

def test_team_admin_can_connect_detectors(self) -> None:
self.login_as(user=self.team_admin_user)
self.organization.update_option("sentry:alerts_member_write", False)

detector = self.create_detector(
project=self.create_project(organization=self.organization, teams=[self.team]),
created_by_id=self.user.id,
)
body_params = {
"detectorId": detector.id,
"workflowId": self.unconnected_workflow.id,
}
self.get_success_response(
self.organization.slug,
**body_params,
)

def test_team_admin_cannot_connect_detectors_for_other_projects(self) -> None:
self.login_as(user=self.team_admin_user)
self.organization.update_option("sentry:alerts_member_write", False)

other_detector = self.create_detector(
project=self.create_project(organization=self.organization, teams=[]),
created_by_id=self.user.id,
)
body_params = {
"detectorId": other_detector.id,
"workflowId": self.unconnected_workflow.id,
}
self.get_error_response(self.organization.slug, **body_params, status_code=403)

def test_member_can_connect_user_created_detectors(self) -> None:
def test_member_can_connect_detectors_with_alerts_write(self) -> None:
self.organization.flags.allow_joinleave = False
self.organization.save()
self.organization.update_option("sentry:alerts_member_write", True)
self.login_as(user=self.member_user)

detector = self.create_detector(
Expand All @@ -244,28 +213,10 @@ def test_member_can_connect_user_created_detectors(self) -> None:
**body_params,
)

def test_member_can_connect_system_created_detectors(self) -> None:
self.organization.flags.allow_joinleave = False
self.organization.save()
self.login_as(user=self.member_user)

detector = self.create_detector(
project=self.create_project(organization=self.organization),
created_by_id=None,
type=ErrorGroupType.slug,
)
body_params = {
"detectorId": detector.id,
"workflowId": self.unconnected_workflow.id,
}
self.get_success_response(
self.organization.slug,
**body_params,
)

def test_member_cannot_connect_detectors_for_other_projects(self) -> None:
def test_member_can_connect_detectors_for_other_projects_with_alerts_write(self) -> None:
self.organization.flags.allow_joinleave = False
self.organization.save()
self.organization.update_option("sentry:alerts_member_write", True)
self.login_as(user=self.member_user)

other_detector = self.create_detector(
Expand All @@ -276,9 +227,9 @@ def test_member_cannot_connect_detectors_for_other_projects(self) -> None:
"detectorId": other_detector.id,
"workflowId": self.unconnected_workflow.id,
}
self.get_error_response(self.organization.slug, **body_params, status_code=403)
self.get_success_response(self.organization.slug, **body_params)

def test_member_cannot_connect_detectors_when_alerts_member_write_disabled(self) -> None:
def test_member_cannot_connect_detectors_without_alerts_write(self) -> None:
self.organization.update_option("sentry:alerts_member_write", False)
self.organization.flags.allow_joinleave = True
self.organization.save()
Expand Down Expand Up @@ -371,75 +322,10 @@ def test_missing_ids(self) -> None:
self.organization.slug,
)

def test_team_admin_can_disconnect_user_detectors(self) -> None:
self.login_as(user=self.team_admin_user)

detector = self.create_detector(
project=self.create_project(organization=self.organization),
created_by_id=self.user.id,
)
self.create_detector_workflow(
detector=detector,
workflow=self.workflow_1,
)
self.get_success_response(
self.organization.slug,
qs_params={"detector_id": detector.id, "workflow_id": self.workflow_1.id},
)

def test_team_admin_can_disconnect_sentry_detectors(self) -> None:
self.login_as(user=self.team_admin_user)

sentry_detector = self.create_detector(
project=self.create_project(organization=self.organization),
)
self.create_detector_workflow(
detector=sentry_detector,
workflow=self.workflow_1,
)
self.get_success_response(
self.organization.slug,
qs_params={"detector_id": sentry_detector.id, "workflow_id": self.workflow_1.id},
)

def test_team_admin_can_disconnect_detectors_for_accessible_projects(self) -> None:
self.login_as(user=self.team_admin_user)
self.organization.update_option("sentry:alerts_member_write", False)

project_detector = self.create_detector(
project=self.create_project(organization=self.organization, teams=[self.team]),
created_by_id=self.user.id,
)
self.create_detector_workflow(
detector=project_detector,
workflow=self.workflow_1,
)
self.get_success_response(
self.organization.slug,
qs_params={"detector_id": project_detector.id, "workflow_id": self.workflow_1.id},
)

def test_team_admin_cannot_disconnect_detectors_for_other_projects(self) -> None:
self.login_as(user=self.team_admin_user)
self.organization.update_option("sentry:alerts_member_write", False)

other_detector = self.create_detector(
project=self.create_project(organization=self.organization, teams=[]),
created_by_id=self.user.id,
)
self.create_detector_workflow(
detector=other_detector,
workflow=self.workflow_1,
)
self.get_error_response(
self.organization.slug,
qs_params={"detector_id": other_detector.id, "workflow_id": self.workflow_1.id},
status_code=403,
)

def test_member_can_disconnect_user_detectors(self) -> None:
def test_member_can_disconnect_detectors_with_alerts_write(self) -> None:
self.organization.flags.allow_joinleave = False
self.organization.save()
self.organization.update_option("sentry:alerts_member_write", True)
self.login_as(user=self.member_user)

detector = self.create_detector(
Expand All @@ -455,9 +341,10 @@ def test_member_can_disconnect_user_detectors(self) -> None:
qs_params={"detector_id": detector.id, "workflow_id": self.workflow_1.id},
)

def test_member_cannot_disconnect_detectors_for_other_projects(self) -> None:
def test_member_cannot_disconnect_detectors_without_alerts_write(self) -> None:
self.organization.flags.allow_joinleave = False
self.organization.save()
self.organization.update_option("sentry:alerts_member_write", False)
self.login_as(user=self.member_user)

other_detector = self.create_detector(
Expand All @@ -474,45 +361,6 @@ def test_member_cannot_disconnect_detectors_for_other_projects(self) -> None:
status_code=403,
)

def test_member_can_disconnect_system_created_detectors(self) -> None:
self.organization.flags.allow_joinleave = False
self.organization.save()
self.login_as(user=self.member_user)

sentry_detector = self.create_detector(
project=self.create_project(organization=self.organization),
type=ErrorGroupType.slug,
created_by_id=None,
)
self.create_detector_workflow(
detector=sentry_detector,
workflow=self.workflow_1,
)
self.get_success_response(
self.organization.slug,
qs_params={"detector_id": sentry_detector.id, "workflow_id": self.workflow_1.id},
)

def test_member_cannot_disconnect_detectors_when_alerts_member_write_disabled(self) -> None:
self.organization.update_option("sentry:alerts_member_write", False)
self.organization.flags.allow_joinleave = True
self.organization.save()
self.login_as(user=self.member_user)

detector = self.create_detector(
project=self.create_project(organization=self.organization, teams=[self.team]),
created_by_id=self.user.id,
)
self.create_detector_workflow(
detector=detector,
workflow=self.workflow_1,
)
self.get_error_response(
self.organization.slug,
qs_params={"detector_id": detector.id, "workflow_id": self.workflow_1.id},
status_code=403,
)

def test_batch_delete_no_permission(self) -> None:
self.organization.update_option("sentry:alerts_member_write", False)
self.login_as(user=self.member_user)
Expand All @@ -526,6 +374,7 @@ def test_batch_delete_no_permission(self) -> None:
detector=detector,
workflow=self.workflow_1,
)
# Member lacks alerts:write because alerts_member_write is disabled
self.get_error_response(
self.organization.slug,
qs_params={"workflow_id": self.workflow_1.id},
Expand Down