Skip to content

fix(tests): prevent tests from producing real Kafka messages to taskbroker #107729

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
dashed merged 1 commit into from
Feb 6, 2026
Merged

fix(tests): prevent tests from producing real Kafka messages to taskbroker #107729

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
20 changes: 20 additions & 0 deletions src/sentry/testutils/pytest/sentry.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -336,6 +336,26 @@ def register_extensions() -> None:
)


def pytest_sessionstart(session: pytest.Session) -> None:
from sentry.taskworker.registry import TaskNamespace

# Store original send_task so tests that need it can restore it
TaskNamespace._original_send_task = TaskNamespace.send_task # type: ignore[attr-defined]

# Prevent tests from producing real Kafka messages via the taskworker pipeline.
# Tests use TaskRunner (TASKWORKER_ALWAYS_EAGER=True) or BurstTaskRunner
# (_signal_send hook) which both operate before send_task in the call chain.
TaskNamespace.send_task = lambda self, *args, **kwargs: None # type: ignore[method-assign]
Comment on lines +339 to +348
Copy link
Member

@markstory markstory Feb 5, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Should we unwind this monkeypatch with a pytest_sessionfinish?



def pytest_sessionfinish(session: pytest.Session, exitstatus: int) -> None:
from sentry.taskworker.registry import TaskNamespace

if hasattr(TaskNamespace, "_original_send_task"):
TaskNamespace.send_task = TaskNamespace._original_send_task # type: ignore[method-assign]
del TaskNamespace._original_send_task


def pytest_runtest_setup(item: pytest.Item) -> None:
if item.config.getvalue("nomigrations") and any(
mark for mark in item.iter_markers(name="migrations")
Expand Down
17 changes: 13 additions & 4 deletions tests/sentry/taskworker/test_registry.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,6 +19,15 @@
from sentry.taskworker.task import Task


@pytest.fixture
def real_send_task():
"""Restore real send_task for tests that directly test send_task behavior."""
original = TaskNamespace._original_send_task # type: ignore[attr-defined]
TaskNamespace.send_task = original # type: ignore[method-assign]
yield
TaskNamespace.send_task = lambda self, *args, **kwargs: None # type: ignore[method-assign]


def test_namespace_register_task() -> None:
namespace = TaskNamespace(
name="tests",
Expand Down Expand Up @@ -112,7 +121,7 @@ def test_namespace_get_unknown() -> None:


@pytest.mark.django_db
def test_namespace_send_task_no_retry() -> None:
def test_namespace_send_task_no_retry(real_send_task: None) -> None:
namespace = TaskNamespace(
name="tests",
application="sentry",
Expand Down Expand Up @@ -200,7 +209,7 @@ def simple_task_with_compression(param: str) -> None:


@pytest.mark.django_db
def test_namespace_send_task_with_retry() -> None:
def test_namespace_send_task_with_retry(real_send_task: None) -> None:
namespace = TaskNamespace(
name="tests",
application="sentry",
Expand Down Expand Up @@ -231,7 +240,7 @@ def simple_task() -> None:


@pytest.mark.django_db
def test_namespace_with_retry_send_task() -> None:
def test_namespace_with_retry_send_task(real_send_task: None) -> None:
namespace = TaskNamespace(
name="tests",
application="sentry",
Expand Down Expand Up @@ -262,7 +271,7 @@ def simple_task() -> None:


@pytest.mark.django_db
def test_namespace_with_wait_for_delivery_send_task() -> None:
def test_namespace_with_wait_for_delivery_send_task(real_send_task: None) -> None:
namespace = TaskNamespace(
name="tests",
application="sentry",
Expand Down
Loading