feat(relocation): Create pause/unpause endpoints (#61194)

Issue: getsentry/team-ospo#222

src/sentry/api/endpoints/relocations/__init__.py

@@ -1 +1,8 @@
+from string import Template
+
 ERR_FEATURE_DISABLED = "This feature is not yet enabled"
+ERR_UNKNOWN_RELOCATION_STEP = Template("`$step` is not a valid relocation step.")
+ERR_COULD_NOT_PAUSE_RELOCATION_AT_STEP = Template(
+    """Could not pause relocation at step `$step`; this is likely because this step has already
+    started."""
+)

src/sentry/api/endpoints/relocations/details.py

@@ -25,6 +25,8 @@ class RelocationDetailsEndpoint(Endpoint):
         Get a single relocation.
         ``````````````````````````````````````````````````
 
+        :pparam string relocation_uuid: a UUID identifying the relocation.
+
         :auth: required
         """
 

src/sentry/api/endpoints/relocations/index.py

@@ -31,7 +31,7 @@ from sentry.tasks.relocation import uploading_complete
 from sentry.utils.db import atomic_transaction
 from sentry.utils.relocation import RELOCATION_BLOB_SIZE, RELOCATION_FILE_TYPE
 
-ERR_DUPLICATE_RELOCATION = "An in-progress relocation already exists for this owner"
+ERR_DUPLICATE_RELOCATION = "An in-progress relocation already exists for this owner."
 ERR_THROTTLED_RELOCATION = (
     "We've reached our daily limit of relocations - please try again tomorrow or contact support."
 )

src/sentry/api/endpoints/relocations/pause.py

@@ -0,0 +1,117 @@
+from string import Template
+
+from django.db import DatabaseError
+from django.db.models import F
+from rest_framework.request import Request
+from rest_framework.response import Response
+
+from sentry.api.api_owners import ApiOwner
+from sentry.api.api_publish_status import ApiPublishStatus
+from sentry.api.base import Endpoint, region_silo_endpoint
+from sentry.api.endpoints.relocations import (
+    ERR_COULD_NOT_PAUSE_RELOCATION_AT_STEP,
+    ERR_UNKNOWN_RELOCATION_STEP,
+)
+from sentry.api.exceptions import ResourceDoesNotExist
+from sentry.api.permissions import SuperuserPermission
+from sentry.api.serializers import serialize
+from sentry.models.relocation import Relocation
+
+ERR_NOT_PAUSABLE_STATUS = Template(
+    """Relocations can only be paused if they are currently in progress; this relocation is
+    `$status`."""
+)
+ERR_COULD_NOT_PAUSE_RELOCATION = (
+    "Could not pause relocation, perhaps because it is no longer in-progress."
+)
+
+
+@region_silo_endpoint
+class RelocationPauseEndpoint(Endpoint):
+    owner = ApiOwner.RELOCATION
+    publish_status = {
+        # TODO(getsentry/team-ospo#214): Stabilize before GA.
+        "PUT": ApiPublishStatus.EXPERIMENTAL,
+    }
+    permission_classes = (SuperuserPermission,)
+
+    def put(self, request: Request, relocation_uuid: str) -> Response:
+        """
+        Pause an in-progress relocation.
+        ``````````````````````````````````````````````````
+
+        This command accepts a single optional parameter, which specifies the step BEFORE which the
+        pause should occur. If no such parameter is specified, the pause is scheduled for the step
+        immediately following the currently active one, if possible.
+
+        :pparam string relocation_uuid: a UUID identifying the relocation.
+        :param string atStep: an optional string identifying the step to pause at; must be greater
+                               than the currently active step, and one of: `PREPROCESSING`,
+                               `VALIDATING`, `IMPORTING`, `POSTPROCESSING`, `NOTIFYING`.
+
+        :auth: required
+        """
+
+        try:
+            relocation: Relocation = Relocation.objects.get(uuid=relocation_uuid)
+        except Relocation.DoesNotExist:
+            raise ResourceDoesNotExist
+
+        if relocation.status not in {
+            Relocation.Status.IN_PROGRESS.value,
+            Relocation.Status.PAUSE.value,
+        }:
+            return Response(
+                {
+                    "detail": ERR_NOT_PAUSABLE_STATUS.substitute(
+                        status=Relocation.Status(relocation.status).name
+                    )
+                },
+                status=400,
+            )
+
+        at_step = request.data.get("atStep", None)
+        if at_step is not None:
+            try:
+                step = Relocation.Step[at_step.upper()]
+            except KeyError:
+                return Response(
+                    {"detail": ERR_UNKNOWN_RELOCATION_STEP.substitute(step=at_step)},
+                    status=400,
+                )
+
+            if step in {
+                Relocation.Step.UNKNOWN,
+                Relocation.Step.UPLOADING,
+                Relocation.Step.COMPLETED,
+            }:
+                return Response(
+                    {"detail": ERR_COULD_NOT_PAUSE_RELOCATION_AT_STEP.substitute(step=step.name)},
+                    status=400,
+                )
+
+            try:
+                relocation.scheduled_pause_at_step = step.value
+                relocation.save()
+            except DatabaseError:
+                return Response(
+                    {"detail": ERR_COULD_NOT_PAUSE_RELOCATION_AT_STEP.substitute(step=step.name)},
+                    status=400,
+                )
+            pass
+        else:
+            try:
+                updated = Relocation.objects.filter(
+                    uuid=relocation.uuid, step__lt=Relocation.Step.COMPLETED.value - 1
+                ).update(scheduled_pause_at_step=F("step") + 1)
+                if not updated:
+                    raise DatabaseError("Cannot set `scheduled_pause_at_step` to `COMPLETED`")
+
+                relocation.refresh_from_db()
+            except DatabaseError:
+                return Response(
+                    {"detail": ERR_COULD_NOT_PAUSE_RELOCATION},
+                    status=400,
+                )
+
+        return self.respond(serialize(relocation))

src/sentry/api/endpoints/relocations/unpause.py

@@ -0,0 +1,118 @@
+from string import Template
+
+from django.db import DatabaseError, router, transaction
+from rest_framework.request import Request
+from rest_framework.response import Response
+
+from sentry.api.api_owners import ApiOwner
+from sentry.api.api_publish_status import ApiPublishStatus
+from sentry.api.base import Endpoint, region_silo_endpoint
+from sentry.api.endpoints.relocations import (
+    ERR_COULD_NOT_PAUSE_RELOCATION_AT_STEP,
+    ERR_UNKNOWN_RELOCATION_STEP,
+)
+from sentry.api.exceptions import ResourceDoesNotExist
+from sentry.api.permissions import SuperuserPermission
+from sentry.api.serializers import serialize
+from sentry.models.relocation import Relocation
+from sentry.tasks.relocation import get_first_task_for_step
+
+ERR_NOT_UNPAUSABLE_STATUS = Template(
+    """Relocations can only be unpaused if they are already paused; this relocation is
+    `$status`."""
+)
+ERR_COULD_NOT_UNPAUSE_RELOCATION = (
+    "Could not unpause relocation, perhaps because it is no longer in-progress."
+)
+
+
+@region_silo_endpoint
+class RelocationUnpauseEndpoint(Endpoint):
+    owner = ApiOwner.RELOCATION
+    publish_status = {
+        # TODO(getsentry/team-ospo#214): Stabilize before GA.
+        "PUT": ApiPublishStatus.EXPERIMENTAL,
+    }
+    permission_classes = (SuperuserPermission,)
+
+    def put(self, request: Request, relocation_uuid: str) -> Response:
+        """
+        Unpause an in-progress relocation.
+        ``````````````````````````````````````````````````
+
+        This command accepts a single optional parameter, which specifies the step BEFORE which the
+        next pause should occur. If no such parameter is specified, no future pauses are scheduled.
+
+        :pparam string relocation_uuid: a UUID identifying the relocation.
+        :param string untilStep: an optional string identifying the next step to pause before; must
+                                 be greater than the currently active step, and one of:
+                                 `PREPROCESSING`, `VALIDATING`, `IMPORTING`, `POSTPROCESSING`,
+                                 `NOTIFYING`.
+
+        :auth: required
+        """
+
+        # Use a `select_for_update` transaction to prevent duplicate tasks from being started by
+        # racing unpause calls.
+        with transaction.atomic(using=router.db_for_write(Relocation)):
+            try:
+                relocation: Relocation = Relocation.objects.select_for_update().get(
+                    uuid=relocation_uuid
+                )
+            except Relocation.DoesNotExist:
+                raise ResourceDoesNotExist
+
+            if relocation.status != Relocation.Status.PAUSE.value:
+                return Response(
+                    {
+                        "detail": ERR_NOT_UNPAUSABLE_STATUS.substitute(
+                            status=Relocation.Status(relocation.status).name
+                        )
+                    },
+                    status=400,
+                )
+
+            relocation.status = Relocation.Status.IN_PROGRESS.value
+            relocation.latest_task_attempts = 0
+
+            until_step = request.data.get("untilStep", None)
+            if until_step is not None:
+                try:
+                    step = Relocation.Step[until_step.upper()]
+                except KeyError:
+                    return Response(
+                        {"detail": ERR_UNKNOWN_RELOCATION_STEP.substitute(step=until_step)},
+                        status=400,
+                    )
+
+                if step in {
+                    Relocation.Step.UNKNOWN,
+                    Relocation.Step.UPLOADING,
+                    Relocation.Step.COMPLETED,
+                }:
+                    return Response(
+                        {
+                            "detail": ERR_COULD_NOT_PAUSE_RELOCATION_AT_STEP.substitute(
+                                step=step.name
+                            )
+                        },
+                        status=400,
+                    )
+
+                relocation.scheduled_pause_at_step = step.value
+
+            task = get_first_task_for_step(Relocation.Step(relocation.step))
+            if task is None:
+                raise RuntimeError("Unknown relocation task")
+
+            # Save the model first, since we can do so multiple times if the task scheduling fails.
+            try:
+                relocation.save()
+            except DatabaseError:
+                return Response(
+                    {"detail": ERR_COULD_NOT_PAUSE_RELOCATION_AT_STEP.substitute(step=step.name)},
+                    status=400,
+                )
+
+            task.delay(str(relocation.uuid))
+            return self.respond(serialize(relocation))

src/sentry/api/urls.py

@@ -40,7 +40,9 @@ from sentry.api.endpoints.release_thresholds.release_threshold_status_index impo
 )
 from sentry.api.endpoints.relocations.details import RelocationDetailsEndpoint
 from sentry.api.endpoints.relocations.index import RelocationIndexEndpoint
+from sentry.api.endpoints.relocations.pause import RelocationPauseEndpoint
 from sentry.api.endpoints.relocations.public_key import RelocationPublicKeyEndpoint
+from sentry.api.endpoints.relocations.unpause import RelocationUnpauseEndpoint
 from sentry.api.endpoints.source_map_debug_blue_thunder_edition import (
     SourceMapDebugBlueThunderEditionEndpoint,
 )
@@ -755,6 +757,16 @@ RELOCATION_URLS = [
         RelocationDetailsEndpoint.as_view(),
         name="sentry-api-0-relocations-details",
     ),
+    re_path(
+        r"^(?P<relocation_uuid>[^\/]+)/pause/$",
+        RelocationPauseEndpoint.as_view(),
+        name="sentry-api-0-relocations-pause",
+    ),
+    re_path(
+        r"^(?P<relocation_uuid>[^\/]+)/unpause/$",
+        RelocationUnpauseEndpoint.as_view(),
+        name="sentry-api-0-relocations-unpause",
+    ),
 ]
 
 RELAY_URLS = [

src/sentry/tasks/relocation.py

@@ -9,6 +9,7 @@ from typing import Optional
 from zipfile import ZipFile
 
 import yaml
+from celery.app.task import Task
 from cryptography.fernet import Fernet
 from django.db import router, transaction
 from google.cloud.devtools.cloudbuild_v1 import Build
@@ -51,6 +52,7 @@ from sentry.utils.env import gcp_project_id, log_gcp_credentials_details
 from sentry.utils.relocation import (
     RELOCATION_BLOB_SIZE,
     RELOCATION_FILE_TYPE,
+    TASK_TO_STEP,
     LoggingPrinter,
     OrderedTask,
     create_cloudbuild_yaml,
@@ -1171,3 +1173,36 @@ def completed(uuid: str) -> None:
     ):
         relocation.status = Relocation.Status.SUCCESS.value
         relocation.save()
+
+
+TASK_MAP: dict[OrderedTask, Task] = {
+    OrderedTask.NONE: Task(),
+    OrderedTask.UPLOADING_COMPLETE: uploading_complete,
+    OrderedTask.PREPROCESSING_SCAN: preprocessing_scan,
+    OrderedTask.PREPROCESSING_BASELINE_CONFIG: preprocessing_baseline_config,
+    OrderedTask.PREPROCESSING_COLLIDING_USERS: preprocessing_colliding_users,
+    OrderedTask.PREPROCESSING_COMPLETE: preprocessing_complete,
+    OrderedTask.VALIDATING_START: validating_start,
+    OrderedTask.VALIDATING_POLL: validating_poll,
+    OrderedTask.VALIDATING_COMPLETE: validating_complete,
+    OrderedTask.IMPORTING: importing,
+    OrderedTask.POSTPROCESSING: postprocessing,
+    OrderedTask.NOTIFYING_USERS: notifying_users,
+    OrderedTask.NOTIFYING_OWNER: notifying_owner,
+    OrderedTask.COMPLETED: completed,
+}
+
+assert list(OrderedTask._member_map_.keys()) == [k.name for k in TASK_MAP.keys()]
+
+
+def get_first_task_for_step(target_step: Relocation.Step) -> Task | None:
+    min_task: OrderedTask | None = None
+    for ordered_task, step in TASK_TO_STEP.items():
+        if step == target_step:
+            if min_task is None or ordered_task.value < min_task.value:
+                min_task = ordered_task
+
+    if min_task is None or min_task == OrderedTask.NONE:
+        return None
+
+    return TASK_MAP.get(min_task, None)

tests/sentry/api/endpoints/relocations/test_pause.py

@@ -0,0 +1,195 @@
+from datetime import datetime, timezone
+from uuid import uuid4
+
+from sentry.api.endpoints.relocations import (
+    ERR_COULD_NOT_PAUSE_RELOCATION_AT_STEP,
+    ERR_UNKNOWN_RELOCATION_STEP,
+)
+from sentry.api.endpoints.relocations.pause import (
+    ERR_COULD_NOT_PAUSE_RELOCATION,
+    ERR_NOT_PAUSABLE_STATUS,
+)
+from sentry.models.relocation import Relocation
+from sentry.testutils.cases import APITestCase
+from sentry.testutils.silo import region_silo_test
+from sentry.utils.relocation import OrderedTask
+
+TEST_DATE_ADDED = datetime(2023, 1, 23, 1, 23, 45, tzinfo=timezone.utc)
+
+
+@region_silo_test
+class PauseRelocationTest(APITestCase):
+    endpoint = "sentry-api-0-relocations-pause"
+
+    def setUp(self):
+        super().setUp()
+        self.owner = self.create_user(
+            email="owner", is_superuser=False, is_staff=True, is_active=True
+        )
+        self.superuser = self.create_user(
+            "superuser", is_superuser=True, is_staff=True, is_active=True
+        )
+        self.relocation: Relocation = Relocation.objects.create(
+            date_added=TEST_DATE_ADDED,
+            creator_id=self.superuser.id,
+            owner_id=self.owner.id,
+            status=Relocation.Status.IN_PROGRESS.value,
+            step=Relocation.Step.PREPROCESSING.value,
+            want_org_slugs='["foo"]',
+            want_usernames='["alice", "bob"]',
+            latest_notified=Relocation.EmailKind.STARTED.value,
+            latest_task=OrderedTask.PREPROCESSING_SCAN.name,
+            latest_task_attempts=1,
+        )
+
+    def test_good_pause_asap(self):
+        self.login_as(user=self.superuser, superuser=True)
+        response = self.client.put(f"/api/0/relocations/{str(self.relocation.uuid)}/pause/")
+
+        assert response.status_code == 200
+        assert response.data["status"] == Relocation.Status.IN_PROGRESS.name
+        assert response.data["step"] == Relocation.Step.PREPROCESSING.name
+        assert response.data["scheduledPauseAtStep"] == Relocation.Step.VALIDATING.name
+
+    def test_good_pause_at_next_step(self):
+        self.login_as(user=self.superuser, superuser=True)
+        response = self.client.put(
+            f"/api/0/relocations/{str(self.relocation.uuid)}/pause/",
+            {"atStep": Relocation.Step.VALIDATING.name},
+        )
+
+        assert response.status_code == 200
+        assert response.data["status"] == Relocation.Status.IN_PROGRESS.name
+        assert response.data["step"] == Relocation.Step.PREPROCESSING.name
+        assert response.data["scheduledPauseAtStep"] == Relocation.Step.VALIDATING.name
+
+    def test_good_pause_at_future_step(self):
+        self.login_as(user=self.superuser, superuser=True)
+        response = self.client.put(
+            f"/api/0/relocations/{str(self.relocation.uuid)}/pause/",
+            {"atStep": Relocation.Step.NOTIFYING.name},
+        )
+
+        assert response.status_code == 200
+        assert response.data["status"] == Relocation.Status.IN_PROGRESS.name
+        assert response.data["step"] == Relocation.Step.PREPROCESSING.name
+        assert response.data["scheduledPauseAtStep"] == Relocation.Step.NOTIFYING.name
+
+    def test_good_already_paused(self):
+        self.login_as(user=self.superuser, superuser=True)
+        self.relocation.status = Relocation.Status.PAUSE.value
+        self.relocation.save()
+        response = self.client.put(
+            f"/api/0/relocations/{str(self.relocation.uuid)}/pause/",
+            {"atStep": Relocation.Step.IMPORTING.name},
+        )
+
+        assert response.status_code == 200
+        assert response.data["status"] == Relocation.Status.PAUSE.name
+        assert response.data["step"] == Relocation.Step.PREPROCESSING.name
+        assert response.data["scheduledPauseAtStep"] == Relocation.Step.IMPORTING.name
+
+    def test_bad_not_found(self):
+        self.login_as(user=self.superuser, superuser=True)
+        does_not_exist_uuid = uuid4().hex
+        response = self.client.put(f"/api/0/relocations/{str(does_not_exist_uuid)}/pause/")
+
+        assert response.status_code == 404
+
+    def test_bad_already_completed(self):
+        self.login_as(user=self.superuser, superuser=True)
+        self.relocation.status = Relocation.Status.FAILURE.value
+        self.relocation.save()
+        response = self.client.put(f"/api/0/relocations/{str(self.relocation.uuid)}/pause/")
+
+        assert response.status_code == 400
+        assert response.data.get("detail") is not None
+        assert response.data.get("detail") == ERR_NOT_PAUSABLE_STATUS.substitute(
+            status=Relocation.Status.FAILURE.name
+        )
+
+    def test_bad_invalid_step(self):
+        self.login_as(user=self.superuser, superuser=True)
+        response = self.client.put(
+            f"/api/0/relocations/{str(self.relocation.uuid)}/pause/",
+            {"atStep": "nonexistent"},
+        )
+
+        assert response.status_code == 400
+        assert response.data.get("detail") is not None
+        assert response.data.get("detail") == ERR_UNKNOWN_RELOCATION_STEP.substitute(
+            step="nonexistent"
+        )
+
+    def test_bad_unknown_step(self):
+        self.login_as(user=self.superuser, superuser=True)
+        response = self.client.put(
+            f"/api/0/relocations/{str(self.relocation.uuid)}/pause/",
+            {"atStep": Relocation.Step.UNKNOWN.name},
+        )
+
+        assert response.status_code == 400
+        assert response.data.get("detail") is not None
+        assert response.data.get("detail") == ERR_COULD_NOT_PAUSE_RELOCATION_AT_STEP.substitute(
+            step=Relocation.Step.UNKNOWN.name
+        )
+
+    def test_bad_current_step(self):
+        self.login_as(user=self.superuser, superuser=True)
+        response = self.client.put(
+            f"/api/0/relocations/{str(self.relocation.uuid)}/pause/",
+            {"atStep": Relocation.Step.PREPROCESSING.name},
+        )
+
+        assert response.status_code == 400
+        assert response.data.get("detail") is not None
+        assert response.data.get("detail") == ERR_COULD_NOT_PAUSE_RELOCATION_AT_STEP.substitute(
+            step=Relocation.Step.PREPROCESSING.name
+        )
+
+    def test_bad_past_step(self):
+        self.login_as(user=self.superuser, superuser=True)
+        response = self.client.put(
+            f"/api/0/relocations/{str(self.relocation.uuid)}/pause/",
+            {"atStep": Relocation.Step.UPLOADING.name},
+        )
+
+        assert response.status_code == 400
+        assert response.data.get("detail") is not None
+        assert response.data.get("detail") == ERR_COULD_NOT_PAUSE_RELOCATION_AT_STEP.substitute(
+            step=Relocation.Step.UPLOADING.name
+        )
+
+    def test_bad_last_step_specified(self):
+        self.login_as(user=self.superuser, superuser=True)
+        response = self.client.put(
+            f"/api/0/relocations/{str(self.relocation.uuid)}/pause/",
+            {"atStep": Relocation.Step.COMPLETED.name},
+        )
+
+        assert response.status_code == 400
+        assert response.data.get("detail") is not None
+        assert response.data.get("detail") == ERR_COULD_NOT_PAUSE_RELOCATION_AT_STEP.substitute(
+            step=Relocation.Step.COMPLETED.name
+        )
+
+    def test_bad_last_step_automatic(self):
+        self.login_as(user=self.superuser, superuser=True)
+        self.relocation.step = Relocation.Step.NOTIFYING.value
+        self.relocation.save()
+        response = self.client.put(f"/api/0/relocations/{str(self.relocation.uuid)}/pause/")
+
+        assert response.status_code == 400
+        assert response.data.get("detail") is not None
+        assert response.data.get("detail") == ERR_COULD_NOT_PAUSE_RELOCATION
+
+    def test_bad_no_auth(self):
+        response = self.client.put(f"/api/0/relocations/{str(self.relocation.uuid)}/pause/")
+
+        assert response.status_code == 401
+
+    def test_bad_no_superuser(self):
+        self.login_as(user=self.superuser, superuser=False)
+        response = self.client.put(f"/api/0/relocations/{str(self.relocation.uuid)}/pause/")
+
+        assert response.status_code == 403

tests/sentry/api/endpoints/relocations/test_unpause.py

@@ -0,0 +1,259 @@
+from datetime import datetime, timezone
+from unittest.mock import Mock, patch
+from uuid import uuid4
+
+from sentry.api.endpoints.relocations import (
+    ERR_COULD_NOT_PAUSE_RELOCATION_AT_STEP,
+    ERR_UNKNOWN_RELOCATION_STEP,
+)
+from sentry.api.endpoints.relocations.unpause import ERR_NOT_UNPAUSABLE_STATUS
+from sentry.models.relocation import Relocation
+from sentry.testutils.cases import APITestCase
+from sentry.testutils.silo import region_silo_test
+from sentry.utils.relocation import OrderedTask
+
+TEST_DATE_ADDED = datetime(2023, 1, 23, 1, 23, 45, tzinfo=timezone.utc)
+
+
+@region_silo_test
+class UnpauseRelocationTest(APITestCase):
+    endpoint = "sentry-api-0-relocations-unpause"
+
+    def setUp(self):
+        super().setUp()
+        self.owner = self.create_user(
+            email="owner", is_superuser=False, is_staff=True, is_active=True
+        )
+        self.superuser = self.create_user(
+            "superuser", is_superuser=True, is_staff=True, is_active=True
+        )
+        self.relocation: Relocation = Relocation.objects.create(
+            date_added=TEST_DATE_ADDED,
+            creator_id=self.superuser.id,
+            owner_id=self.owner.id,
+            status=Relocation.Status.PAUSE.value,
+            step=Relocation.Step.PREPROCESSING.value,
+            want_org_slugs='["foo"]',
+            want_usernames='["alice", "bob"]',
+            latest_notified=Relocation.EmailKind.STARTED.value,
+            latest_task=OrderedTask.PREPROCESSING_SCAN.name,
+            latest_task_attempts=1,
+        )
+
+    @patch("sentry.tasks.relocation.preprocessing_scan.delay")
+    def test_good_unpause_until_validating(self, async_task_scheduled: Mock):
+        self.login_as(user=self.superuser, superuser=True)
+        response = self.client.put(
+            f"/api/0/relocations/{str(self.relocation.uuid)}/unpause/",
+            {"untilStep": Relocation.Step.VALIDATING.name},
+        )
+
+        assert response.status_code == 200
+        assert response.data["status"] == Relocation.Status.IN_PROGRESS.name
+        assert response.data["step"] == Relocation.Step.PREPROCESSING.name
+        assert response.data["scheduledPauseAtStep"] == Relocation.Step.VALIDATING.name
+
+        assert async_task_scheduled.call_count == 1
+        assert async_task_scheduled.call_args.args == (str(self.relocation.uuid),)
+
+    @patch("sentry.tasks.relocation.validating_start.delay")
+    def test_good_unpause_until_importing(self, async_task_scheduled: Mock):
+        self.login_as(user=self.superuser, superuser=True)
+        self.relocation.step = Relocation.Step.VALIDATING.value
+        self.relocation.save()
+        response = self.client.put(
+            f"/api/0/relocations/{str(self.relocation.uuid)}/unpause/",
+            {"untilStep": Relocation.Step.IMPORTING.name},
+        )
+
+        assert response.status_code == 200
+        assert response.data["status"] == Relocation.Status.IN_PROGRESS.name
+        assert response.data["step"] == Relocation.Step.VALIDATING.name
+        assert response.data["scheduledPauseAtStep"] == Relocation.Step.IMPORTING.name
+
+        assert async_task_scheduled.call_count == 1
+        assert async_task_scheduled.call_args.args == (str(self.relocation.uuid),)
+
+    @patch("sentry.tasks.relocation.importing.delay")
+    def test_good_unpause_until_postprocessing(self, async_task_scheduled: Mock):
+        self.login_as(user=self.superuser, superuser=True)
+        self.relocation.step = Relocation.Step.IMPORTING.value
+        self.relocation.save()
+        response = self.client.put(
+            f"/api/0/relocations/{str(self.relocation.uuid)}/unpause/",
+            {"untilStep": Relocation.Step.POSTPROCESSING.name},
+        )
+
+        assert response.status_code == 200
+        assert response.data["status"] == Relocation.Status.IN_PROGRESS.name
+        assert response.data["step"] == Relocation.Step.IMPORTING.name
+        assert response.data["scheduledPauseAtStep"] == Relocation.Step.POSTPROCESSING.name
+
+        assert async_task_scheduled.call_count == 1
+        assert async_task_scheduled.call_args.args == (str(self.relocation.uuid),)
+
+    @patch("sentry.tasks.relocation.postprocessing.delay")
+    def test_good_unpause_until_notifying(self, async_task_scheduled: Mock):
+        self.login_as(user=self.superuser, superuser=True)
+        self.relocation.step = Relocation.Step.POSTPROCESSING.value
+        self.relocation.save()
+        response = self.client.put(
+            f"/api/0/relocations/{str(self.relocation.uuid)}/unpause/",
+            {"untilStep": Relocation.Step.NOTIFYING.name},
+        )
+
+        assert response.status_code == 200
+        assert response.data["status"] == Relocation.Status.IN_PROGRESS.name
+        assert response.data["step"] == Relocation.Step.POSTPROCESSING.name
+        assert response.data["scheduledPauseAtStep"] == Relocation.Step.NOTIFYING.name
+
+        assert async_task_scheduled.call_count == 1
+        assert async_task_scheduled.call_args.args == (str(self.relocation.uuid),)
+
+    @patch("sentry.tasks.relocation.notifying_users.delay")
+    def test_good_unpause_no_follow_up_step(self, async_task_scheduled: Mock):
+        self.login_as(user=self.superuser, superuser=True)
+        self.relocation.step = Relocation.Step.NOTIFYING.value
+        self.relocation.save()
+        response = self.client.put(f"/api/0/relocations/{str(self.relocation.uuid)}/unpause/")
+
+        assert response.status_code == 200
+        assert response.data["status"] == Relocation.Status.IN_PROGRESS.name
+        assert response.data["step"] == Relocation.Step.NOTIFYING.name
+        assert not response.data["scheduledPauseAtStep"]
+
+        assert async_task_scheduled.call_count == 1
+        assert async_task_scheduled.call_args.args == (str(self.relocation.uuid),)
+
+    @patch("sentry.tasks.relocation.preprocessing_scan.delay")
+    def test_bad_not_found(self, async_task_scheduled: Mock):
+        self.login_as(user=self.superuser, superuser=True)
+        does_not_exist_uuid = uuid4().hex
+        response = self.client.put(f"/api/0/relocations/{str(does_not_exist_uuid)}/unpause/")
+
+        assert response.status_code == 404
+
+    @patch("sentry.tasks.relocation.preprocessing_scan.delay")
+    def test_bad_already_completed(self, async_task_scheduled: Mock):
+        self.login_as(user=self.superuser, superuser=True)
+        self.relocation.status = Relocation.Status.FAILURE.value
+        self.relocation.save()
+        response = self.client.put(f"/api/0/relocations/{str(self.relocation.uuid)}/unpause/")
+
+        assert response.status_code == 400
+        assert response.data.get("detail") is not None
+        assert response.data.get("detail") == ERR_NOT_UNPAUSABLE_STATUS.substitute(
+            status=Relocation.Status.FAILURE.name
+        )
+
+        assert async_task_scheduled.call_count == 0
+
+    @patch("sentry.tasks.relocation.preprocessing_scan.delay")
+    def test_bad_already_paused(self, async_task_scheduled: Mock):
+        self.login_as(user=self.superuser, superuser=True)
+        self.relocation.status = Relocation.Status.IN_PROGRESS.value
+        self.relocation.save()
+        response = self.client.put(f"/api/0/relocations/{str(self.relocation.uuid)}/unpause/")
+
+        assert response.status_code == 400
+        assert response.data.get("detail") is not None
+        assert response.data.get("detail") == ERR_NOT_UNPAUSABLE_STATUS.substitute(
+            status=Relocation.Status.IN_PROGRESS.name
+        )
+
+        assert async_task_scheduled.call_count == 0
+
+    @patch("sentry.tasks.relocation.preprocessing_scan.delay")
+    def test_bad_invalid_step(self, async_task_scheduled: Mock):
+        self.login_as(user=self.superuser, superuser=True)
+        response = self.client.put(
+            f"/api/0/relocations/{str(self.relocation.uuid)}/unpause/",
+            {"untilStep": "nonexistent"},
+        )
+
+        assert response.status_code == 400
+        assert response.data.get("detail") is not None
+        assert response.data.get("detail") == ERR_UNKNOWN_RELOCATION_STEP.substitute(
+            step="nonexistent"
+        )
+
+        assert async_task_scheduled.call_count == 0
+
+    @patch("sentry.tasks.relocation.preprocessing_scan.delay")
+    def test_bad_unknown_step(self, async_task_scheduled: Mock):
+        self.login_as(user=self.superuser, superuser=True)
+        response = self.client.put(
+            f"/api/0/relocations/{str(self.relocation.uuid)}/unpause/",
+            {"untilStep": Relocation.Step.UNKNOWN.name},
+        )
+
+        assert response.status_code == 400
+        assert response.data.get("detail") is not None
+        assert response.data.get("detail") == ERR_COULD_NOT_PAUSE_RELOCATION_AT_STEP.substitute(
+            step=Relocation.Step.UNKNOWN.name
+        )
+
+        assert async_task_scheduled.call_count == 0
+
+    @patch("sentry.tasks.relocation.preprocessing_scan.delay")
+    def test_bad_current_step(self, async_task_scheduled: Mock):
+        self.login_as(user=self.superuser, superuser=True)
+        response = self.client.put(
+            f"/api/0/relocations/{str(self.relocation.uuid)}/unpause/",
+            {"untilStep": Relocation.Step.PREPROCESSING.name},
+        )
+
+        assert response.status_code == 400
+        assert response.data.get("detail") is not None
+        assert response.data.get("detail") == ERR_COULD_NOT_PAUSE_RELOCATION_AT_STEP.substitute(
+            step=Relocation.Step.PREPROCESSING.name
+        )
+
+        assert async_task_scheduled.call_count == 0
+
+    @patch("sentry.tasks.relocation.preprocessing_scan.delay")
+    def test_bad_past_step(self, async_task_scheduled: Mock):
+        self.login_as(user=self.superuser, superuser=True)
+        response = self.client.put(
+            f"/api/0/relocations/{str(self.relocation.uuid)}/unpause/",
+            {"untilStep": Relocation.Step.UPLOADING.name},
+        )
+
+        assert response.status_code == 400
+        assert response.data.get("detail") is not None
+        assert response.data.get("detail") == ERR_COULD_NOT_PAUSE_RELOCATION_AT_STEP.substitute(
+            step=Relocation.Step.UPLOADING.name
+        )
+
+        assert async_task_scheduled.call_count == 0
+
+    @patch("sentry.tasks.relocation.preprocessing_scan.delay")
+    def test_bad_last_step(self, async_task_scheduled: Mock):
+        self.login_as(user=self.superuser, superuser=True)
+        response = self.client.put(
+            f"/api/0/relocations/{str(self.relocation.uuid)}/unpause/",
+            {"untilStep": Relocation.Step.COMPLETED.name},
+        )
+
+        assert response.status_code == 400
+        assert response.data.get("detail") is not None
+        assert response.data.get("detail") == ERR_COULD_NOT_PAUSE_RELOCATION_AT_STEP.substitute(
+            step=Relocation.Step.COMPLETED.name
+        )
+
+        assert async_task_scheduled.call_count == 0
+
+    @patch("sentry.tasks.relocation.preprocessing_scan.delay")
+    def test_bad_no_auth(self, async_task_scheduled: Mock):
+        response = self.client.put(f"/api/0/relocations/{str(self.relocation.uuid)}/unpause/")
+
+        assert response.status_code == 401
+        assert async_task_scheduled.call_count == 0
+
+    @patch("sentry.tasks.relocation.preprocessing_scan.delay")
+    def test_bad_no_superuser(self, async_task_scheduled: Mock):
+        self.login_as(user=self.superuser, superuser=False)
+        response = self.client.put(f"/api/0/relocations/{str(self.relocation.uuid)}/unpause/")
+
+        assert response.status_code == 403
+        assert async_task_scheduled.call_count == 0