fix(hc): Improve test mode detection (#52605)

src/sentry/db/postgres/roles.py

@@ -16,7 +16,7 @@ def in_test_psql_role_override(role_name: str, using: str | None = None):
     back to its original.  Has no effect in production.
     """
 
-    if "pytest" not in sys.modules or os.environ.get("DB", "postgres") != "postgres":
+    if "pytest" not in sys.argv[0] or os.environ.get("DB", "postgres") != "postgres":
         yield
         return
 

src/sentry/db/postgres/transactions.py

@@ -19,7 +19,7 @@ def django_test_transaction_water_mark(using: str | None = None):
 
     This method has no effect in production.
     """
-    if "pytest" not in sys.modules:
+    if "pytest" not in sys.argv[0]:
         yield
         return
 
@@ -61,7 +61,7 @@ def in_test_hide_transaction_boundary():
     In tests, it hides 'in_test_assert_no_transaction' invocations against problematic code paths.
     Using this function is a huge code smell, often masking some other code smell, but not always possible to avoid.
     """
-    if "pytest" not in sys.modules:
+    if "pytest" not in sys.argv[0]:
         yield
         return
 
@@ -82,7 +82,7 @@ def in_test_assert_no_transaction(msg: str):
     execution time can have cause major performance issues by holding transactional resources open for long periods
     of time.
     """
-    if "pytest" not in sys.modules or not in_test_transaction_enforcement.enabled:
+    if "pytest" not in sys.argv[0] or not in_test_transaction_enforcement.enabled:
         return
 
     from sentry.testutils import hybrid_cloud

src/sentry/db/router.py

@@ -75,7 +75,7 @@ class SiloRouter:
             self.__is_simulated = False
 
     def use_simulated(self, value: bool):
-        if "pytest" not in sys.modules:
+        if "pytest" not in sys.argv[0]:
             raise ValueError("Cannot mutate simulation mode outside of tests")
         self.__is_simulated = value
 
@@ -95,7 +95,7 @@ class SiloRouter:
 
             # If we're in tests raise an error, otherwise return 'no decision'
             # so that django skips migration operations that won't work.
-            if "pytest" in sys.modules:
+            if "pytest" in sys.argv[0]:
                 raise SiloConnectionUnavailableError(
                     f"Cannot resolve table {table} in {silo_mode}. "
                     f"Application silo mode is {active_mode} and simulated silos are not enabled."

src/sentry/services/hybrid_cloud/util.py

@@ -19,7 +19,7 @@ class FunctionSiloLimit(SiloLimit):
         current_mode: SiloMode,
         available_modes: Iterable[SiloMode],
     ) -> Callable[..., Any]:
-        if "pytest" in sys.modules:
+        if "pytest" in sys.argv[0]:
             mode_str = ", ".join(str(m) for m in available_modes)
             message = (
                 f"Called {original_method.__name__} in "

src/sentry/shared_integrations/client/proxy.py

@@ -76,7 +76,7 @@ class IntegrationProxyClient(ApiClient):
         is_region_silo = SiloMode.get_current_mode() == SiloMode.REGION
         subnet_secret = getattr(settings, "SENTRY_SUBNET_SECRET", None)
         control_address = getattr(settings, "SENTRY_CONTROL_ADDRESS", None)
-        is_test_environment = "pytest" in sys.modules
+        is_test_environment = "pytest" in sys.argv[0]
 
         if is_region_silo and subnet_secret and control_address:
             self._should_proxy_to_control = True

src/sentry/signals.py

@@ -91,7 +91,7 @@ class BetterSignal(Signal):
             try:
                 response = receiver(signal=self, sender=sender, **named)
             except Exception as err:
-                if "pytest" in sys.modules:
+                if "pytest" in sys.argv[0]:
                     if (
                         _receivers_that_raise is _AllReceivers.ALL
                         or receiver in _receivers_that_raise

src/sentry/silo/base.py

@@ -49,7 +49,7 @@ class SiloMode(Enum):
         cases unless the exit_single_process_silo_context is explicitly embedded, ensuring that this single process
         silo mode simulates the boundaries explicitly between what would be separate processes in deployment.
         """
-        if "pytest" in sys.modules:
+        if "pytest" in sys.argv[0]:
             assert (
                 single_process_silo_mode_state.mode is None
             ), "Re-entrant invariant broken! Use exit_single_process_silo_context to explicit pass 'fake' RPC boundaries."

src/sentry/tasks/deletion/scheduled.py

@@ -153,5 +153,5 @@ def run_deletion(deletion_id, first_pass=True, silo_mode="CONTROL"):
             },
         )
         sentry_sdk.capture_exception(err)
-        if "pytest" in sys.modules:
+        if "pytest" in sys.argv[0]:
             raise err

src/sentry/testutils/silo.py

@@ -255,7 +255,7 @@ _role_privileges_created: MutableMapping[str, bool] = {}
 
 def create_model_role_guards(app_config: Any, using: str, **kwargs: Any):
     global _role_created
-    if "pytest" not in sys.modules:
+    if "pytest" not in sys.argv[0]:
         return
 
     from sentry.db.models.fields.hybrid_cloud_foreign_key import HybridCloudForeignKey