ref: fix usage_accountant-related test pollution (#88113)

previously failing with:

```console
$ pytest tests/sentry/ingest/ingest_consumer/test_ingest_consumer_processing.py::test_accountant_transaction tests/sentry/usage_accountant/test_accountant.py::test_accountant
============================= test session starts ==============================
platform darwin -- Python 3.13.1, pytest-8.1.2, pluggy-1.5.0
django: version: 5.1.7
rootdir: /Users/asottile/workspace/sentry
configfile: pyproject.toml
plugins: fail-slow-0.3.0, time-machine-2.16.0, json-report-1.5.0, metadata-3.1.1, xdist-3.0.2, django-4.9.0, pytest_sentry-0.3.0, anyio-3.7.1, rerunfailures-15.0, cov-4.0.0
collected 2 items                                                              

tests/sentry/ingest/ingest_consumer/test_ingest_consumer_processing.py . [ 50%]
                                                                         [ 50%]
tests/sentry/usage_accountant/test_accountant.py F                       [100%]

=================================== FAILURES ===================================
_______________________________ test_accountant ________________________________
tests/sentry/usage_accountant/test_accountant.py:47: in test_accountant
    accountant.init_backend(producer)
src/sentry/usage_accountant/accountant.py:33: in init_backend
    assert _accountant_backend is None, "Accountant already initialized once."
E   AssertionError: Accountant already initialized once.
=========================== short test summary info ============================
FAILED tests/sentry/usage_accountant/test_accountant.py::test_accountant - AssertionError: Accountant already initialized once.
========================= 1 failed, 1 passed in 5.12s ==========================
```

<!-- Describe your PR here. -->

Author: asottile-sentry

pyproject.toml

@@ -406,6 +406,7 @@ module = [
     "sentry.tempest.endpoints.*",
     "sentry.tempest.migrations.*",
     "sentry.testutils.helpers.task_runner",
+    "sentry.testutils.helpers.usage_accountant",
     "sentry.testutils.pytest.json_report_reruns",
     "sentry.testutils.pytest.show_flaky_failures",
     "sentry.testutils.skips",

src/sentry/testutils/helpers/usage_accountant.py

@@ -0,0 +1,21 @@
+from __future__ import annotations
+
+import contextlib
+from collections.abc import Generator
+
+from arroyo.backends.abstract import Producer
+from arroyo.backends.kafka import KafkaPayload
+from usageaccountant import UsageAccumulator
+
+from sentry.usage_accountant import accountant
+
+
+@contextlib.contextmanager
+def usage_accountant_backend(producer: Producer[KafkaPayload]) -> Generator[None]:
+    assert accountant._accountant_backend is None, accountant._accountant_backend
+    accountant._accountant_backend = UsageAccumulator(producer=producer)
+    try:
+        yield
+    finally:
+        accountant._shutdown()
+        accountant._accountant_backend = None

src/sentry/usage_accountant/accountant.py

@@ -10,8 +10,7 @@
 import atexit
 import logging
 
-from arroyo.backends.abstract import Producer
-from arroyo.backends.kafka import KafkaPayload, KafkaProducer, build_kafka_configuration
+from arroyo.backends.kafka import KafkaProducer, build_kafka_configuration
 from usageaccountant import UsageAccumulator, UsageUnit
 
 from sentry.conf.types.kafka_definition import Topic
@@ -23,29 +22,7 @@
 _accountant_backend: UsageAccumulator | None = None
 
 
-def init_backend(producer: Producer[KafkaPayload]) -> None:
-    """
-    This method should be used externally only in tests to fit a
-    mock producer.
-    """
-    global _accountant_backend
-
-    assert _accountant_backend is None, "Accountant already initialized once."
-    _accountant_backend = UsageAccumulator(producer=producer)
-    atexit.register(_shutdown)
-
-
-def reset_backend() -> None:
-    """
-    This method should be used externally only in tests to reset
-    the accountant backend.
-    """
-    global _accountant_backend
-    _accountant_backend = None
-
-
 def _shutdown() -> None:
-    global _accountant_backend
     if _accountant_backend is not None:
         _accountant_backend.flush()
         _accountant_backend.close()

tests/sentry/event_manager/test_event_manager.py

@@ -65,7 +65,6 @@
 from sentry.models.release import Release
 from sentry.models.releasecommit import ReleaseCommit
 from sentry.models.releaseprojectenvironment import ReleaseProjectEnvironment
-from sentry.options import set
 from sentry.projectoptions.defaults import DEFAULT_GROUPING_CONFIG
 from sentry.spans.grouping.utils import hash_values
 from sentry.testutils.asserts import assert_mock_called_once_with_partial
@@ -77,13 +76,13 @@
 )
 from sentry.testutils.helpers import apply_feature_flag_on_cls, override_options
 from sentry.testutils.helpers.datetime import before_now, freeze_time
+from sentry.testutils.helpers.usage_accountant import usage_accountant_backend
 from sentry.testutils.performance_issues.event_generators import get_event
 from sentry.testutils.pytest.fixtures import django_db_all
 from sentry.testutils.skips import requires_snuba
 from sentry.tsdb.base import TSDBModel
 from sentry.types.activity import ActivityType
 from sentry.types.group import PriorityLevel
-from sentry.usage_accountant import accountant
 from sentry.utils import json
 from sentry.utils.cache import cache_key_for_event
 from sentry.utils.eventuser import EventUser
@@ -3517,21 +3516,21 @@ def test_cogs_event_manager(
     broker.create_topic(topic, 1)
     producer = broker.get_producer()
 
-    set("shared_resources_accounting_enabled", [settings.COGS_EVENT_STORE_LABEL])
-
-    accountant.init_backend(producer)
-
-    raw_event_params = make_event(**event_data)
+    with (
+        override_options(
+            {"shared_resources_accounting_enabled": [settings.COGS_EVENT_STORE_LABEL]}
+        ),
+        usage_accountant_backend(producer),
+    ):
+        raw_event_params = make_event(**event_data)
 
-    manager = EventManager(raw_event_params)
-    manager.normalize()
-    normalized_data = dict(manager.get_data())
-    _ = manager.save(default_project)
+        manager = EventManager(raw_event_params)
+        manager.normalize()
+        normalized_data = dict(manager.get_data())
+        _ = manager.save(default_project)
 
-    expected_len = len(json.dumps(normalized_data))
+        expected_len = len(json.dumps(normalized_data))
 
-    accountant._shutdown()
-    accountant.reset_backend()
     msg1 = broker.consume(Partition(topic, 0), 0)
     assert msg1 is not None
     payload = msg1.payload

tests/sentry/ingest/ingest_consumer/test_ingest_consumer_processing.py

@@ -29,11 +29,11 @@
 from sentry.models.debugfile import create_files_from_dif_zip
 from sentry.models.eventattachment import EventAttachment
 from sentry.models.userreport import UserReport
-from sentry.options import set
 from sentry.testutils.helpers.features import Feature
+from sentry.testutils.helpers.options import override_options
+from sentry.testutils.helpers.usage_accountant import usage_accountant_backend
 from sentry.testutils.pytest.fixtures import django_db_all
 from sentry.testutils.skips import requires_snuba, requires_symbolicator
-from sentry.usage_accountant import accountant
 from sentry.utils.eventuser import EventUser
 from sentry.utils.json import loads
 
@@ -169,42 +169,43 @@ def test_accountant_transaction(default_project):
     broker.create_topic(topic, 1)
     producer = broker.get_producer()
 
-    set("shared_resources_accounting_enabled", [settings.EVENT_PROCESSING_STORE])
-
-    accountant.init_backend(producer)
-
-    now = datetime.datetime.now()
-    event = {
-        "type": "transaction",
-        "timestamp": now.isoformat(),
-        "start_timestamp": now.isoformat(),
-        "spans": [],
-        "contexts": {
-            "trace": {
-                "parent_span_id": "8988cec7cc0779c1",
-                "type": "trace",
-                "op": "foobar",
-                "trace_id": "a7d67cf796774551a95be6543cacd459",
-                "span_id": "babaae0d4b7512d9",
-                "status": "ok",
-            }
-        },
-    }
-    payload = get_normalized_event(event, default_project)
-    serialized = orjson.dumps(payload).decode()
-    process_event(
-        ConsumerType.Events,
-        {
-            "payload": serialized,
-            "start_time": time.time() - 3600,
-            "event_id": payload["event_id"],
-            "project_id": default_project.id,
-            "remote_addr": "127.0.0.1",
-        },
-        project=default_project,
-    )
+    with (
+        override_options(
+            {"shared_resources_accounting_enabled": [settings.EVENT_PROCESSING_STORE]}
+        ),
+        usage_accountant_backend(producer),
+    ):
+        now = datetime.datetime.now()
+        event = {
+            "type": "transaction",
+            "timestamp": now.isoformat(),
+            "start_timestamp": now.isoformat(),
+            "spans": [],
+            "contexts": {
+                "trace": {
+                    "parent_span_id": "8988cec7cc0779c1",
+                    "type": "trace",
+                    "op": "foobar",
+                    "trace_id": "a7d67cf796774551a95be6543cacd459",
+                    "span_id": "babaae0d4b7512d9",
+                    "status": "ok",
+                }
+            },
+        }
+        payload = get_normalized_event(event, default_project)
+        serialized = orjson.dumps(payload).decode()
+        process_event(
+            ConsumerType.Events,
+            {
+                "payload": serialized,
+                "start_time": time.time() - 3600,
+                "event_id": payload["event_id"],
+                "project_id": default_project.id,
+                "remote_addr": "127.0.0.1",
+            },
+            project=default_project,
+        )
 
-    accountant._shutdown()
     msg1 = broker.consume(Partition(topic, 0), 0)
     assert msg1 is not None
     payload = msg1.payload

tests/sentry/usage_accountant/test_accountant.py

@@ -6,9 +6,10 @@
 from arroyo.types import BrokerValue, Partition, Topic
 from usageaccountant import UsageUnit
 
-from sentry.options import set
+from sentry.testutils.helpers.options import override_options
+from sentry.testutils.helpers.usage_accountant import usage_accountant_backend
 from sentry.testutils.pytest.fixtures import django_db_all
-from sentry.usage_accountant import accountant, record
+from sentry.usage_accountant import record
 from sentry.utils.json import loads
 
 
@@ -42,15 +43,13 @@ def test_accountant(mock_time: mock.Mock) -> None:
     broker.create_topic(topic, 1)
     producer = broker.get_producer()
 
-    set("shared_resources_accounting_enabled", ["resource_1"])
-
-    accountant.init_backend(producer)
-
-    mock_time.return_value = 1594839910.1
-    record("resource_1", "feature_1", 100, UsageUnit.BYTES)
-    record("resource_1", "feature_2", 100, UsageUnit.BYTES)
-
-    accountant._shutdown()
+    with (
+        override_options({"shared_resources_accounting_enabled": ["resource_1"]}),
+        usage_accountant_backend(producer),
+    ):
+        mock_time.return_value = 1594839910.1
+        record("resource_1", "feature_1", 100, UsageUnit.BYTES)
+        record("resource_1", "feature_2", 100, UsageUnit.BYTES)
 
     msg1 = broker.consume(Partition(topic, 0), 0)
     assert_msg(