fix(tests): Fix outstanding flaky tests (#128264)

Fixes the root causes of backend tests that still fail on master, based
on the per-attempt failure events in the `sentry-tests` project over the
last 30 days. There is one commit per cause.

All fixes are test-only except one: the uptime alert index now orders
its pagination by `id`. The customer-domain `(?!new)` bug was moved to
its own PRs, #128274 (backend) and #128275 (frontend).

### Production change: unordered pagination in the uptime alert index
(`test_owner_filter` 11, `test` 2)

`OrganizationUptimeAlertIndexEndpoint` paginates with `OffsetPaginator`
and no `order_by`, so offset and limit are applied to an unordered
queryset. The order within a page, and which rows land on which page,
follows Postgres row order. That is a real correctness bug: rows can be
duplicated or skipped across pages. It now orders by `id`.

`test_paginates_in_id_order` is a new regression test. It creates five
detectors, deletes and reinserts their rows newest first so the physical
order is the reverse of id order, then walks three pages with
`per_page=2` and the cursor. It checks that each page has the expected
ids in ascending order, with no gaps or duplicates. Without the fix it
gets the rows in reverse.

Refs #108890
Refs #108884

### Test-only fixes

**Unordered queries in the deploy notification:
`SlackDeployNotificationTest::test_deploy_block` (12).** The buttons
come from `release.projects.all()` and the footer from
`release.projects.all()[0]`. These are two unordered queries, and the
`LIMIT 1` one can use a different join plan, so the footer can name a
project other than the first button. The production code does not
promise any order between them. The test still checks every button's
slug, URL, and value, and now checks that the footer names one of the
release's projects instead of the first button's.

Refs #108927

**Postgres row order:
`test_redirect_for_logged_in_user_with_different_active_org` (19).** The
session had no active org, so the channel login view fell back to the
first of the user's orgs from an unordered query. When Postgres returned
the second org first, the user counted as already logged in to the
target org. The test now sets `activeorg` in the session, which is the
state its comment describes.

Refs #108900
Refs #108881

**Snuba rows left over from an earlier test: feedback categories (17),
label query (13 + 3), `EventTest::test_snuba_data` (3).** These classes
write to and read from Snuba but never reset it. Snowflake ids are built
from the wall-clock second plus a Redis sequence, and Redis is flushed
and Postgres rolled back after every test. So two tests that start in
the same second get the same org and project ids, and the second one
reads the first one's rows. The classes now use the `reset_snuba`
fixture through `@pytest.mark.usefixtures`, not `SnubaTestCase`.
`SnubaTestCase` also overrides `create_project()` and `store_event()`,
which would change the fixtures and ingestion these tests run against,
and none of them needs that.

Snowflake id reuse across tests is broader than these classes. Any test
that reads Snuba data without resetting it can hit it. This PR fixes
only the tests that fail.

Refs #108654
Refs #109451
Refs #109450
Refs #109445
Refs #109435
Refs #109433
Refs #109432
Refs #108888
Refs #108770
Refs #108935
Refs #108873
Refs #109056
Refs #108982
Refs #108930
Refs #108902
Refs #108838

**Global `time.sleep` patches: `StreamDemoDataTest` (12).**
`patch("sentry.conduit.tasks.time.sleep")` replaces `time.sleep` for
every thread in the process, so any background thread that sleeps adds
to the count (CI saw 451 to 5332 calls instead of 100). Both stream
tests now patch the module-local `time` references instead:
`sentry.conduit.tasks.time`, and in the retry test also
`sentry.utils.retries.time`, where the retry policy sleeps. The real
`time.time()` stays in place. The retry test now also checks both sleep
counts. No other test in the file patches `time.sleep`.

Refs #108842

**Batch deadline:
`TestBatchedOccurrenceConsumer::test_issue_multiple_status_changes`
(4).** Arroyo starts the batch deadline when the strategy is built, so
with `max_batch_time=1` a slow run flushed the batch before all six
messages arrived. The test now uses `sys.maxsize`, like its sibling
tests since #88955.

Refs #108997

**1 second rate limit windows:
`test_impersonation_enforces_rate_limits_when_disabled` (1).**
Impersonation limits use 1 second windows, so three calls that straddle
a second boundary are counted in different windows. The calls now run
under `freeze_time`, like the other tests in the file.
`test_impersonation_with_custom_endpoint_config` has the same pattern
and gets the same fix.

Refs #109043

**Random collision: `test_ttl_jitter` (1).** Two random TTLs round to
the same second about once in 1800 runs, and the test asserted they
differ. It now checks the bounds with real draws and the spread with
fixed draws.

Refs #108952

### Reproductions

Each fix was checked with a deterministic reproduction, run against the
parent commit (fails) and against the fix (passes). The uptime
pagination test and the TTL jitter draws are committed. The others are
not, because they force the failure by rewriting physical row order,
pinning the snowflake clock, or adding a sleeping background thread.
Those depend on planner behavior or test order, or test the test harness
rather than the product.

| Fix | Reproduction | Parent | Fixed |
|---|---|---|---|
| Uptime pagination | `test_paginates_in_id_order` | fail | pass |
| Deploy footer | release_project rows in the opposite physical order to
project rows | fail | pass |
| Channel login | first org's mapping row rewritten to the end of the
heap | fail | pass |
| Snuba reset (3 classes) | snowflake clock pinned to one second,
preceded by a test that writes feedback or events | fail | pass |
| Conduit sleeps | background thread calling `time.sleep` during the
test | fail | pass |
| Batch deadline | arroyo clock advanced 2s after the strategy is built
| fail | pass |
| Impersonation windows | wall clock advanced 1s per rate limit check |
fail | pass |
| TTL jitter | `random.seed(1095)` | fail | pass |

The ten changed test files (139 tests) also passed 20 consecutive runs
with `-n 2 --dist=loadfile`, per-worker Snuba, and `PYTHONHASHSEED=0`,
as in CI.

<!-- junior-request-attribution:start -->
via @joshuarli.
<!-- junior-request-attribution:end -->

<!-- junior-session-footer:start -->
<!-- junior-conversation-id:agent%3A3f29e2341c39dfa3441598ed78b058a7 -->

--

[View Junior
Session](https://junior-prod.sentry.dev/conversations/agent%3A3f29e2341c39dfa3441598ed78b058a7)
[[Sentry]](https://sentry.sentry.io/explore/conversations/agent%3A3f29e2341c39dfa3441598ed78b058a7/?project=4510944073809921)

<!-- junior-session-footer:end -->

---------

Co-authored-by: sentry-junior[bot] <264270552+sentry-junior[bot]@users.noreply.github.com>
Co-authored-by: Joshua Li <14209781+joshuarli@users.noreply.github.com>

Author: sentry-junior[bot]

src/sentry/uptime/endpoints/organiation_uptime_alert_index.py

@@ -141,4 +141,5 @@ def get(
             queryset=queryset,
             on_results=lambda x: serialize(x, request.user, UptimeDetectorSerializer()),
             paginator_cls=OffsetPaginator,
+            order_by="id",
         )

tests/sentry/conduit/test_tasks.py

@@ -1,5 +1,6 @@
 import datetime
 import re
+import time
 from unittest.mock import patch
 from uuid import uuid4
 
@@ -256,9 +257,12 @@ class StreamDemoDataTest(TestCase):
         CONDUIT_PUBLISH_JWT_AUDIENCE="conduit",
         CONDUIT_PUBLISH_URL="http://localhost:9093",
     )
-    @patch("sentry.conduit.tasks.time.sleep")
-    def test_stream_demo_data_sends_all_phases(self, mock_sleep):
+    # Patch the module's own `time` reference. Patching `time.sleep` would also count the sleeps of
+    # every other thread in the test process.
+    @patch("sentry.conduit.tasks.time")
+    def test_stream_demo_data_sends_all_phases(self, mock_time):
         """Test that stream_demo_data sends START, DELTA, and END phases."""
+        mock_time.time.side_effect = time.time
         org_id = 123
         channel_id = str(uuid4())
 
@@ -271,7 +275,7 @@ def test_stream_demo_data_sends_all_phases(self, mock_sleep):
         stream_demo_data(org_id=org_id, channel_id=channel_id)
 
         assert len(responses.calls) == NUM_DELTAS + 2
-        assert mock_sleep.call_count == NUM_DELTAS
+        assert mock_time.sleep.call_count == NUM_DELTAS
 
     @responses.activate
     @override_settings(
@@ -280,9 +284,13 @@ def test_stream_demo_data_sends_all_phases(self, mock_sleep):
         CONDUIT_PUBLISH_JWT_AUDIENCE="conduit",
         CONDUIT_PUBLISH_URL="http://localhost:9093",
     )
-    @patch("sentry.conduit.tasks.time.sleep")
-    def test_stream_demo_data_stops_on_exhausted_retries(self, mock_sleep):
+    # Patch the module-local `time` references (the task's and the retry policy's) rather than
+    # `time.sleep`, which would replace sleep for every thread in the test process.
+    @patch("sentry.utils.retries.time")
+    @patch("sentry.conduit.tasks.time")
+    def test_stream_demo_data_stops_on_exhausted_retries(self, mock_time, mock_retries_time):
         """Test that stream_demo_data stops streaming when a publish fails after max retries."""
+        mock_time.time.side_effect = time.time
         org_id = 123
         channel_id = str(uuid4())
 
@@ -305,3 +313,6 @@ def test_stream_demo_data_stops_on_exhausted_retries(self, mock_sleep):
             stream_demo_data(org_id=org_id, channel_id=channel_id)
 
         assert len(responses.calls) == 10 + PUBLISH_REQUEST_MAX_RETRIES
+        # START plus 9 deltas succeed, and the task sleeps after each delta.
+        assert mock_time.sleep.call_count == 9
+        assert mock_retries_time.sleep.call_count == PUBLISH_REQUEST_MAX_RETRIES - 1

tests/sentry/feedback/endpoints/test_organization_feedback_categories.py

@@ -1,6 +1,7 @@
 from datetime import datetime
 from unittest.mock import patch
 
+import pytest
 from django.urls import reverse
 
 from sentry.feedback.lib.utils import FeedbackCreationSource
@@ -13,6 +14,9 @@
 
 
 @cell_silo_test
+# Snowflake org and project ids repeat when two tests start in the same second, so reset Snuba
+# rather than read back a previous test's rows for "our" project.
+@pytest.mark.usefixtures("reset_snuba")
 class OrganizationFeedbackCategoriesTest(APITestCase):
     endpoint = "sentry-api-0-organization-user-feedback-categories"
 

tests/sentry/feedback/lib/test_label_query.py

@@ -20,6 +20,9 @@
 from tests.sentry.feedback import mock_feedback_event
 
 
+# Snowflake org and project ids repeat when two tests start in the same second, so reset Snuba
+# rather than read back a previous test's rows for "our" project.
+@pytest.mark.usefixtures("reset_snuba")
 class TestLabelQuery(APITestCase):
     def setUp(self) -> None:
         super().setUp()

tests/sentry/integrations/slack/notifications/test_deploy.py

@@ -50,21 +50,22 @@ def test_deploy_block(self) -> None:
         )
         assert blocks[0]["text"]["text"] == fallback_text
 
-        first_project = None
-        for i in range(len(projects)):
-            project = SLUGS_TO_PROJECT[blocks[2]["elements"][i]["text"]["text"]]
-            if not first_project:
-                first_project = project
+        actions = blocks[2]["elements"]
+        assert sorted(action["text"]["text"] for action in actions) == sorted(SLUGS_TO_PROJECT)
+        for action in actions:
+            project = SLUGS_TO_PROJECT[action["text"]["text"]]
             assert (
-                blocks[2]["elements"][i]["url"]
+                action["url"]
                 == f"http://testserver/organizations/{self.organization.slug}/releases/"
                 f"{release.version}/?project={project.id}&unselectedSeries=Healthy&referrer=release_activity&notification_uuid={notification.notification_uuid}"
             )
-            assert blocks[2]["elements"][i]["value"] == "link_clicked"
-        assert first_project is not None
+            assert action["value"] == "link_clicked"
 
-        # footer project is the first project in the actions list
+        # The footer and the actions come from separate unordered queries, so the footer can
+        # name any of the release's projects, not necessarily the first action's.
+        footer_slug, footer_link = blocks[1]["elements"][0]["text"].split(" | ")
+        assert footer_slug in SLUGS_TO_PROJECT
         assert (
-            blocks[1]["elements"][0]["text"]
-            == f"{first_project.slug} | <http://testserver/settings/account/notifications/deploy/?referrer=release_activity-slack-user&notification_uuid={notification.notification_uuid}|Notification Settings>"
+            footer_link
+            == f"<http://testserver/settings/account/notifications/deploy/?referrer=release_activity-slack-user&notification_uuid={notification.notification_uuid}|Notification Settings>"
         )

tests/sentry/issues/test_run.py

@@ -465,7 +465,9 @@ def test_issue_multiple_status_changes(
         strategy = OccurrenceStrategyFactory(
             mode="batched-parallel",
             max_batch_size=6,
-            max_batch_time=1,
+            # The batch deadline starts when the strategy is built, so a short one can split the
+            # batch on a slow run.
+            max_batch_time=sys.maxsize,
         ).create_with_partitions(
             commit=mock_commit,
             partitions={},

tests/sentry/middleware/test_ratelimit_middleware.py

@@ -235,10 +235,14 @@ def test_impersonation_enforces_rate_limits_when_disabled(self) -> None:
         impersonator = self.create_user(email="impersonator@example.com")
         request.actual_user = impersonator
 
-        # Call this endpoint multiple times get hit by rate limit
-        self.middleware.process_view(request, self._test_endpoint_no_rate_limits, [], {})
-        self.middleware.process_view(request, self._test_endpoint_no_rate_limits, [], {})
-        response = self.middleware.process_view(request, self._test_endpoint_no_rate_limits, [], {})
+        # Call this endpoint multiple times get hit by rate limit. Freeze time so all calls land
+        # in the same 1s impersonation window.
+        with freeze_time("2000-01-01"):
+            self.middleware.process_view(request, self._test_endpoint_no_rate_limits, [], {})
+            self.middleware.process_view(request, self._test_endpoint_no_rate_limits, [], {})
+            response = self.middleware.process_view(
+                request, self._test_endpoint_no_rate_limits, [], {}
+            )
 
         assert response is not None
         assert isinstance(response, HttpResponse)
@@ -289,10 +293,12 @@ def get(self):
         impersonator = self.create_user(email="impersonator@example.com")
         request.actual_user = impersonator
 
-        # Call with an endpoint that normally doesn't enforce rate limits multiple times to get hit by rate limit
-        self.middleware.process_view(request, high_limit_view, [], {})
-        self.middleware.process_view(request, high_limit_view, [], {})
-        response = self.middleware.process_view(request, high_limit_view, [], {})
+        # Call with an endpoint that normally doesn't enforce rate limits multiple times to get hit by rate limit.
+        # Freeze time so all calls land in the same 1s impersonation window.
+        with freeze_time("2000-01-01"):
+            self.middleware.process_view(request, high_limit_view, [], {})
+            self.middleware.process_view(request, high_limit_view, [], {})
+            response = self.middleware.process_view(request, high_limit_view, [], {})
 
         # During impersonation, should get a 429 response
         assert response is not None

tests/sentry/sentry_metrics/test_indexer_cache.py

@@ -1,4 +1,5 @@
 from datetime import timedelta
+from unittest.mock import patch
 
 import pytest
 from django.conf import settings
@@ -216,7 +217,14 @@ def test_ttl_jitter() -> None:
     ttl_2 = indexer_cache.randomized_ttl
     assert base_ttl <= ttl_2 <= max_ttl
 
-    assert not ttl_1 == ttl_2
+    # Two real draws land on the same whole second about once in 1800 runs, so check the spread
+    # with fixed draws instead.
+    with patch("sentry.sentry_metrics.indexer.cache.random") as mock_random:
+        mock_random.uniform.side_effect = [0.0, 0.1, 0.25]
+        assert indexer_cache.randomized_ttl == base_ttl
+        assert indexer_cache.randomized_ttl == base_ttl + 720
+        assert indexer_cache.randomized_ttl == max_ttl
+    mock_random.uniform.assert_called_with(0, 0.25)
 
 
 def test_separate_namespacing() -> None:

tests/sentry/services/eventstore/test_models.py

@@ -44,6 +44,9 @@ def timezone_context(tz: str):
         time.tzset()
 
 
+# Snowflake org and project ids repeat when two tests start in the same second, so reset Snuba
+# rather than read back a previous test's rows for "our" project.
+@pytest.mark.usefixtures("reset_snuba")
 class EventTest(TestCase, PerformanceIssueTestCase):
     def test_pickling_compat(self) -> None:
         event = self.store_event(

tests/sentry/uptime/endpoints/test_organization_uptime_alert_index.py

@@ -1,6 +1,9 @@
+from django.db import connections, router
+
 from sentry.api.serializers import serialize
 from sentry.constants import ObjectStatus
 from sentry.uptime.endpoints.serializers import UptimeDetectorSerializer
+from sentry.workflow_engine.models import Detector
 from tests.sentry.uptime.endpoints import UptimeAlertBaseEndpointTest
 
 
@@ -23,6 +26,30 @@ def test(self) -> None:
         resp = self.get_success_response(self.organization.slug)
         self.check_valid_response(resp, [alert_1, alert_2])
 
+    def test_paginates_in_id_order(self) -> None:
+        detectors = [self.create_uptime_detector(name=f"test{i}") for i in range(5)]
+        # Rewrite the rows newest first, so their physical order, which an unordered query
+        # follows, no longer matches id order.
+        with connections[router.db_for_write(Detector)].cursor() as cursor:
+            for detector in reversed(detectors):
+                cursor.execute(
+                    "WITH moved AS (DELETE FROM workflow_engine_detector WHERE id = %s RETURNING *) "
+                    "INSERT INTO workflow_engine_detector SELECT * FROM moved",
+                    [detector.id],
+                )
+
+        response = self.get_success_response(self.organization.slug, per_page=2)
+        self.check_valid_response(response, detectors[0:2])
+        response = self.get_success_response(
+            self.organization.slug, per_page=2, cursor=self.get_cursor_headers(response)[1]
+        )
+        self.check_valid_response(response, detectors[2:4])
+        response = self.get_success_response(
+            self.organization.slug, per_page=2, cursor=self.get_cursor_headers(response)[1]
+        )
+        self.check_valid_response(response, detectors[4:5])
+        assert 'rel="next"; results="false"' in response["Link"]
+
     def test_search_by_url(self) -> None:
         self.create_uptime_detector()
         santry_monitor = self.create_uptime_detector(