review

Author: dubloom

tests/debugging/test_encoding.py

@@ -271,50 +271,33 @@ def test_process_tags_are_not_included_by_default():
 
 
 def test_process_tags_are_included():
-    from unittest.mock import patch
-
-    from ddtrace.internal.process_tags import _process_tag_reload
-    from ddtrace.internal.process_tags.constants import ENTRYPOINT_BASEDIR_TAG
-    from ddtrace.internal.process_tags.constants import ENTRYPOINT_NAME_TAG
-    from ddtrace.internal.process_tags.constants import ENTRYPOINT_TYPE_SCRIPT
-    from ddtrace.internal.process_tags.constants import ENTRYPOINT_TYPE_TAG
-    from ddtrace.internal.process_tags.constants import ENTRYPOINT_WORKDIR_TAG
-    from ddtrace.settings._config import config
+    from ddtrace.internal.settings._config import config
+    from tests.utils import process_tag_reload
 
     try:
-        with patch("sys.argv", ["/path/to/test_script.py"]), patch("os.getcwd", return_value="/path/to/workdir"):
-            config._process_tags_enabled = True
-            _process_tag_reload()
-            s = Snapshot(
-                probe=create_snapshot_line_probe(probe_id="batch-test", source_file="foo.py", line=42),
-                frame=inspect.currentframe(),
-                thread=threading.current_thread(),
-            )
-            buffer_size = 30 * (1 << 20)
-            queue = SignalQueue(encoder=LogSignalJsonEncoder(None), buffer_size=buffer_size)
+        config._process_tags_enabled = True
+        process_tag_reload()
+        s = Snapshot(
+            probe=create_snapshot_line_probe(probe_id="batch-test", source_file="foo.py", line=42),
+            frame=inspect.currentframe(),
+            thread=threading.current_thread(),
+        )
+        buffer_size = 30 * (1 << 20)
+        queue = SignalQueue(encoder=LogSignalJsonEncoder(None), buffer_size=buffer_size)
 
-            s.line({})
+        s.line({})
 
-            queue = SignalQueue(encoder=LogSignalJsonEncoder("test-service"))
-            queue.put(s)
-            data = queue.flush()
-            assert data is not None
-            payload, _ = data
-            decoded = json.loads(payload.decode())
-
-            assert "process_tags" in decoded[0]
-
-            expected_raw = (
-                f"{ENTRYPOINT_BASEDIR_TAG}:to,"
-                f"{ENTRYPOINT_NAME_TAG}:test_script,"
-                f"{ENTRYPOINT_TYPE_TAG}:{ENTRYPOINT_TYPE_SCRIPT},"
-                f"{ENTRYPOINT_WORKDIR_TAG}:workdir"
-            )
+        queue = SignalQueue(encoder=LogSignalJsonEncoder("test-service"))
+        queue.put(s)
+        data = queue.flush()
+        assert data is not None
+        payload, _ = data
+        decoded = json.loads(payload.decode())
 
-            assert decoded[0]["process_tags"] == expected_raw
+        assert "process_tags" in decoded[0]
     finally:
         config._process_tags_enabled = False
-        _process_tag_reload()
+        process_tag_reload()
 
 
 def test_batch_flush_reencode():