support pytest-forked plugin

Author: eleanorjboyd

python_files/tests/pytestadapter/.data/test_forked.py

@@ -0,0 +1,17 @@
+# Copyright (c) Microsoft Corporation.
+# Licensed under the MIT License.
+
+import pytest
+import os
+
+
+@pytest.mark.forked
+def test_forked_process_p():
+    pid = os.getpid()
+    print(f"Running in process with PID: {pid}")
+    assert True
+@pytest.mark.forked
+def test_forked_process_f():
+    pid = os.getpid()
+    print(f"Running in process with PID: {pid}")
+    assert False

python_files/vscode_pytest/__init__.py

@@ -68,6 +68,8 @@ def __init__(self, message):
 collected_tests_so_far = []
 TEST_RUN_PIPE = os.getenv("TEST_RUN_PIPE")
 SYMLINK_PATH = None
+# Get variable set in the `run_pytest_script.py` for the parent process to check if its forked
+ROOT_PROCESS_PID: int = os.getenv("PROCESS_ID")
 
 
 def pytest_load_initial_conftests(early_config, parser, args):  # noqa: ARG001
@@ -92,6 +94,24 @@ def pytest_load_initial_conftests(early_config, parser, args):  # noqa: ARG001
         global IS_DISCOVERY
         IS_DISCOVERY = True
 
+    global ROOT_PROCESS_PID
+    print(early_config.pluginmanager.list_name_plugin())
+    if early_config.pluginmanager.has_plugin("pytest_forked"):
+        if ROOT_PROCESS_PID and int(os.getpid()) == int(ROOT_PROCESS_PID):
+            print(
+                "Plugin info[vscode-pytest]: Forked plugin detected but NOT running in forked process."
+            )
+        elif ROOT_PROCESS_PID:
+            print(
+                f"Plugin info[vscode-pytest]: Forked plugin detected and running in forked process. Root PID: {ROOT_PROCESS_PID}, Current PID: {os.getpid()}"
+            )
+        else:
+            print("Plugin info[vscode-pytest]: No root PID defined, setting to 0.")
+            ROOT_PROCESS_PID = 0
+    else:
+        # If the plugin is not detected, then the root process is the current process.
+        ROOT_PROCESS_PID = 0
+
     # check if --rootdir is in the args
     for arg in args:
         if "--rootdir=" in arg:
@@ -858,12 +878,19 @@ def send_execution_message(
         status (Literal["success", "error"]): Execution status indicating success or error.
         tests (Union[testRunResultDict, None]): Test run results, if available.
     """
-    payload: ExecutionPayloadDict = ExecutionPayloadDict(
-        cwd=cwd, status=status, result=tests, not_found=None, error=None
-    )
-    if ERRORS:
-        payload["error"] = ERRORS
-    send_message(payload)
+    current_pid = os.getpid()
+    # if ROOT_PROCESS_PID is 0 then forked plugin is not enabled
+    if ROOT_PROCESS_PID != 0 and int(current_pid) != int(ROOT_PROCESS_PID):
+        print("PID NOT equal to the root proccess, no return", current_pid, ROOT_PROCESS_PID)
+        return
+    else:
+        print("PID equal to the root of forked proccess", current_pid, ROOT_PROCESS_PID)
+        payload: ExecutionPayloadDict = ExecutionPayloadDict(
+            cwd=cwd, status=status, result=tests, not_found=None, error=None
+        )
+        if ERRORS:
+            payload["error"] = ERRORS
+        send_message(payload)
 
 
 def send_discovery_message(cwd: str, session_node: TestNode) -> None:

python_files/vscode_pytest/run_pytest_script.py

@@ -28,6 +28,7 @@ def run_pytest(args):
 # args through sys.argv. It then runs pytest.main() with the args and test_ids.
 
 if __name__ == "__main__":
+    os.environ["PROCESS_ID"] = str(os.getpid())
     # Add the root directory to the path so that we can import the plugin.
     directory_path = pathlib.Path(__file__).parent.parent
     sys.path.append(os.fspath(directory_path))