ci: fix flaky internal core tests (#17745)

## Description

There are a few scenarios where we may not properly clean up the core event hub between tests, and where we will fail because of a Python 3.12+ deprecation warning for `os.fork()`

## Testing

<!-- Describe your testing strategy or note what tests are included -->

## Risks

<!-- Note any risks associated with this change, or "None" if no risks -->

## Additional Notes

<!-- Any other information that would be helpful for reviewers -->


Co-authored-by: brett.langdon <brett.langdon@datadoghq.com>

Author: brettlangdon

ddtrace/internal/core/__init__.py

@@ -188,7 +188,20 @@ def __init__(
     def __enter__(self) -> "ExecutionContext[EventType]":
         if "_CURRENT_CONTEXT" in globals():
             self._token = _CURRENT_CONTEXT.set(self)
-        dispatch("context.started.%s" % self.identifier, (self,))
+        try:
+            dispatch("context.started.%s" % self.identifier, (self,))
+        except BaseException:
+            # If dispatch raises, __exit__ won't be called — reset the context ourselves
+            # to avoid leaving _CURRENT_CONTEXT pointing at this partially-entered context.
+            if self._token is not None:
+                try:
+                    _CURRENT_CONTEXT.reset(self._token)
+                except ValueError:
+                    log.debug("Encountered ValueError resetting context in __enter__ error path for %s", self)
+                except LookupError:
+                    log.debug("Encountered LookupError resetting context in __enter__ error path for %s", self)
+                self._token = None
+            raise
         return self
 
     def __repr__(self) -> str:

tests/internal/remoteconfig/test_remoteconfig.py

@@ -182,11 +182,9 @@ def test_remote_config_enable_validate_rc_disabled(remote_config_worker):
         assert remoteconfig_poller.status == ServiceStatus.STOPPED
 
 
-@pytest.mark.skipif(
-    sys.version_info >= (3, 12, 0),
-    reason="Python 3.12 subprocess will raise deprecation warning for forking in a multi-threaded process",
+@pytest.mark.subprocess(
+    ddtrace_run=True, env=dict(DD_REMOTE_CONFIGURATION_ENABLED="true", PYTHONWARNINGS="ignore::DeprecationWarning:os")
 )
-@pytest.mark.subprocess(ddtrace_run=True, env=dict(DD_REMOTE_CONFIGURATION_ENABLED="true"))
 def test_remote_config_forksafe():
     import os
 

tests/internal/test_context_events_api.py

@@ -26,6 +26,10 @@ def _inner(*args, **kwargs):
 
 
 class TestContextEventsApi(unittest.TestCase):
+    def setUp(self):
+        core.reset_listeners()
+        core._reset_context()
+
     def tearDown(self):
         core.reset_listeners()
         core._reset_context()

tests/internal/test_forksafe.py

@@ -5,7 +5,7 @@
 from ddtrace.internal import forksafe
 
 
-@pytest.mark.subprocess()
+@pytest.mark.subprocess(env={"PYTHONWARNINGS": "ignore::DeprecationWarning:os"})
 def test_forksafe():
     import os
 
@@ -34,7 +34,7 @@ def my_func():
     assert exit_code == 12
 
 
-@pytest.mark.subprocess()
+@pytest.mark.subprocess(env={"PYTHONWARNINGS": "ignore::DeprecationWarning:os"})
 def test_registry():
     """This verifies that registered hooks are called after a fork.
 
@@ -82,7 +82,7 @@ def before_fork():
     assert exit_code == 12
 
 
-@pytest.mark.subprocess()
+@pytest.mark.subprocess(env={"PYTHONWARNINGS": "ignore::DeprecationWarning:os"})
 def test_duplicates():
     import os
 
@@ -117,7 +117,7 @@ def f3():
     assert exit_code == 12
 
 
-@pytest.mark.subprocess()
+@pytest.mark.subprocess(env={"PYTHONWARNINGS": "ignore::DeprecationWarning:os"})
 def test_method_usage():
     import os
 
@@ -201,7 +201,7 @@ def test_event_basic():
     event.clear()
 
 
-@pytest.mark.subprocess()
+@pytest.mark.subprocess(env={"PYTHONWARNINGS": "ignore::DeprecationWarning:os"})
 def test_event_fork():
     """Check that a forksafe.Event is reset after a fork().
 
@@ -228,7 +228,7 @@ def test_event_fork():
     assert exit_code == 12
 
 
-@pytest.mark.subprocess()
+@pytest.mark.subprocess(env={"PYTHONWARNINGS": "ignore::DeprecationWarning:os"})
 def test_double_fork():
     import os
 

tests/internal/test_manual_context_events_api.py

@@ -8,6 +8,8 @@
 
 @pytest.fixture(autouse=True)
 def reset_context_and_listeners():
+    core.reset_listeners()
+    core._reset_context()
     yield
     core.reset_listeners()
     core._reset_context()

tests/internal/test_periodic.py

@@ -156,7 +156,7 @@ def periodic(self):
     assert queue == list(range(n + 1))
 
 
-@pytest.mark.subprocess()
+@pytest.mark.subprocess(env={"PYTHONWARNINGS": "ignore::DeprecationWarning:os"})
 def test_forksafe_awakeable_periodic_service():
     import os
     from threading import Event
@@ -199,7 +199,7 @@ def periodic(self):
 
 
 @pytest.mark.skipif(not hasattr(os, "fork"), reason="requires fork")
-@pytest.mark.subprocess()
+@pytest.mark.subprocess(env={"PYTHONWARNINGS": "ignore::DeprecationWarning:os"})
 def test_autorestart_false_service_restarts_in_parent_after_fork():
     """A PeriodicService with autorestart=False must keep running in the parent
     process after a fork. The flag means 'do not restart in the child', not
@@ -402,7 +402,7 @@ def _get_native_thread_name():
     return None
 
 
-@pytest.mark.subprocess()
+@pytest.mark.subprocess(env={"PYTHONWARNINGS": "ignore::DeprecationWarning:os"})
 def test_periodic_thread_stop_without_join_forksafe():
     """
     Dropping a PeriodicThread that was stop()'d without join() in a forked child

tests/internal/test_products.py

@@ -103,7 +103,7 @@ def test_product_manager_start():
     assert a.started
 
 
-@pytest.mark.subprocess()
+@pytest.mark.subprocess(env={"PYTHONWARNINGS": "ignore::DeprecationWarning:os"})
 def test_product_manager_restart():
     import os
 

tests/internal/test_subscribers.py

@@ -22,9 +22,11 @@
 @pytest.fixture(autouse=True)
 def reset_event_hub():
     """Reset event hub after each test to prevent listener leakage between tests."""
+    core._reset_context()
     yield
     event_hub.reset()
     called.clear()
+    core._reset_context()
 
 
 @dataclass