Remove fifo regression (microsoft#24685)

fixes regression in
microsoft#24656 by reverting
problem

noxfile.py

@@ -53,7 +53,6 @@ def install_python_libs(session: nox.Session):
         )
 
     session.install("packaging")
-    session.install("debugpy")
 
     # Download get-pip script
     session.run(

python_files/testing_tools/socket_manager.py

@@ -20,24 +20,39 @@ def __exit__(self, *_):
         self.close()
 
     def connect(self):
-        self._writer = open(self.name, "w", encoding="utf-8")  # noqa: SIM115, PTH123
-        # reader created in read method
+        if sys.platform == "win32":
+            self._writer = open(self.name, "w", encoding="utf-8")  # noqa: SIM115, PTH123
+            # reader created in read method
+        else:
+            self._socket = _SOCKET(socket.AF_UNIX, socket.SOCK_STREAM)
+            self._socket.connect(self.name)
         return self
 
     def close(self):
-        self._writer.close()
-        if hasattr(self, "_reader"):
-            self._reader.close()
+        if sys.platform == "win32":
+            self._writer.close()
+        else:
+            # add exception catch
+            self._socket.close()
 
     def write(self, data: str):
-        try:
-            # for windows, is should only use \n\n
-            request = f"""content-length: {len(data)}\ncontent-type: application/json\n\n{data}"""
-            self._writer.write(request)
-            self._writer.flush()
-        except Exception as e:
-            print("error attempting to write to pipe", e)
-            raise (e)
+        if sys.platform == "win32":
+            try:
+                # for windows, is should only use \n\n
+                request = (
+                    f"""content-length: {len(data)}\ncontent-type: application/json\n\n{data}"""
+                )
+                self._writer.write(request)
+                self._writer.flush()
+            except Exception as e:
+                print("error attempting to write to pipe", e)
+                raise (e)
+        else:
+            # must include the carriage-return defined (as \r\n) for unix systems
+            request = (
+                f"""content-length: {len(data)}\r\ncontent-type: application/json\r\n\r\n{data}"""
+            )
+            self._socket.send(request.encode("utf-8"))
 
     def read(self, bufsize=1024) -> str:
         """Read data from the socket.
@@ -48,10 +63,17 @@ def read(self, bufsize=1024) -> str:
         Returns:
             data (str): Data received from the socket.
         """
-        # returns a string automatically from read
-        if not hasattr(self, "_reader"):
-            self._reader = open(self.name, encoding="utf-8")  # noqa: SIM115, PTH123
-        return self._reader.read(bufsize)
+        if sys.platform == "win32":
+            # returns a string automatically from read
+            if not hasattr(self, "_reader"):
+                self._reader = open(self.name, encoding="utf-8")  # noqa: SIM115, PTH123
+            return self._reader.read(bufsize)
+        else:
+            # receive bytes and convert to string
+            while True:
+                part: bytes = self._socket.recv(bufsize)
+                data: str = part.decode("utf-8")
+                return data
 
 
 class SocketManager:

python_files/tests/pytestadapter/helpers.py

@@ -128,22 +128,6 @@ def parse_rpc_message(data: str) -> Tuple[Dict[str, str], str]:
             print("json decode error")
 
 
-def _listen_on_fifo(pipe_name: str, result: List[str], completed: threading.Event):
-    # Open the FIFO for reading
-    fifo_path = pathlib.Path(pipe_name)
-    with fifo_path.open() as fifo:
-        print("Waiting for data...")
-        while True:
-            if completed.is_set():
-                break  # Exit loop if completed event is set
-            data = fifo.read()  # This will block until data is available
-            if len(data) == 0:
-                # If data is empty, assume EOF
-                break
-            print(f"Received: {data}")
-            result.append(data)
-
-
 def _listen_on_pipe_new(listener, result: List[str], completed: threading.Event):
     """Listen on the named pipe or Unix domain socket for JSON data from the server.
 
@@ -323,19 +307,14 @@ def runner_with_cwd_env(
         # if additional environment variables are passed, add them to the environment
         if env_add:
             env.update(env_add)
-        # server = UnixPipeServer(pipe_name)
-        # server.start()
-        #################
-        # Create the FIFO (named pipe) if it doesn't exist
-        # if not pathlib.Path.exists(pipe_name):
-        os.mkfifo(pipe_name)
-        #################
+        server = UnixPipeServer(pipe_name)
+        server.start()
 
         completed = threading.Event()
 
         result = []  # result is a string array to store the data during threading
         t1: threading.Thread = threading.Thread(
-            target=_listen_on_fifo, args=(pipe_name, result, completed)
+            target=_listen_on_pipe_new, args=(server, result, completed)
         )
         t1.start()
 
@@ -385,14 +364,14 @@ def generate_random_pipe_name(prefix=""):
 
     # For Windows, named pipes have a specific naming convention.
     if sys.platform == "win32":
-        return f"\\\\.\\pipe\\{prefix}-{random_suffix}"
+        return f"\\\\.\\pipe\\{prefix}-{random_suffix}-sock"
 
     # For Unix-like systems, use either the XDG_RUNTIME_DIR or a temporary directory.
     xdg_runtime_dir = os.getenv("XDG_RUNTIME_DIR")
     if xdg_runtime_dir:
-        return os.path.join(xdg_runtime_dir, f"{prefix}-{random_suffix}")  # noqa: PTH118
+        return os.path.join(xdg_runtime_dir, f"{prefix}-{random_suffix}.sock")  # noqa: PTH118
     else:
-        return os.path.join(tempfile.gettempdir(), f"{prefix}-{random_suffix}")  # noqa: PTH118
+        return os.path.join(tempfile.gettempdir(), f"{prefix}-{random_suffix}.sock")  # noqa: PTH118
 
 
 class UnixPipeServer:

python_files/unittestadapter/pvsc_utils.py

@@ -18,6 +18,8 @@
 
 from typing_extensions import NotRequired  # noqa: E402
 
+from testing_tools import socket_manager  # noqa: E402
+
 # Types
 
 
@@ -329,10 +331,10 @@ def send_post_request(
 
     if __writer is None:
         try:
-            __writer = open(test_run_pipe, "w", encoding="utf-8", newline="\r\n")  # noqa: SIM115, PTH123
+            __writer = socket_manager.PipeManager(test_run_pipe)
+            __writer.connect()
         except Exception as error:
             error_msg = f"Error attempting to connect to extension named pipe {test_run_pipe}[vscode-unittest]: {error}"
-            print(error_msg, file=sys.stderr)
             __writer = None
             raise VSCodeUnittestError(error_msg) from error
 
@@ -341,11 +343,10 @@ def send_post_request(
         "params": payload,
     }
     data = json.dumps(rpc)
+
     try:
         if __writer:
-            request = f"""content-length: {len(data)}\ncontent-type: application/json\n\n{data}"""
-            __writer.write(request)
-            __writer.flush()
+            __writer.write(data)
         else:
             print(
                 f"Connection error[vscode-unittest], writer is None \n[vscode-unittest] data: \n{data} \n",

python_files/vscode_pytest/__init__.py

@@ -21,6 +21,11 @@
 
 import pytest
 
+script_dir = pathlib.Path(__file__).parent.parent
+sys.path.append(os.fspath(script_dir))
+sys.path.append(os.fspath(script_dir / "lib" / "python"))
+from testing_tools import socket_manager  # noqa: E402
+
 if TYPE_CHECKING:
     from pluggy import Result
 
@@ -166,7 +171,7 @@ def pytest_exception_interact(node, call, report):
             collected_test = TestRunResultDict()
             collected_test[node_id] = item_result
             cwd = pathlib.Path.cwd()
-            send_execution_message(
+            execution_post(
                 os.fsdecode(cwd),
                 "success",
                 collected_test if collected_test else None,
@@ -290,7 +295,7 @@ def pytest_report_teststatus(report, config):  # noqa: ARG001
             )
             collected_test = TestRunResultDict()
             collected_test[absolute_node_id] = item_result
-            send_execution_message(
+            execution_post(
                 os.fsdecode(cwd),
                 "success",
                 collected_test if collected_test else None,
@@ -324,7 +329,7 @@ def pytest_runtest_protocol(item, nextitem):  # noqa: ARG001
             )
             collected_test = TestRunResultDict()
             collected_test[absolute_node_id] = item_result
-            send_execution_message(
+            execution_post(
                 os.fsdecode(cwd),
                 "success",
                 collected_test if collected_test else None,
@@ -400,15 +405,15 @@ def pytest_sessionfinish(session, exitstatus):
                 "children": [],
                 "id_": "",
             }
-            send_discovery_message(os.fsdecode(cwd), error_node)
+            post_response(os.fsdecode(cwd), error_node)
         try:
             session_node: TestNode | None = build_test_tree(session)
             if not session_node:
                 raise VSCodePytestError(
                     "Something went wrong following pytest finish, \
                         no session node was created"
                 )
-            send_discovery_message(os.fsdecode(cwd), session_node)
+            post_response(os.fsdecode(cwd), session_node)
         except Exception as e:
             ERRORS.append(
                 f"Error Occurred, traceback: {(traceback.format_exc() if e.__traceback__ else '')}"
@@ -420,7 +425,7 @@ def pytest_sessionfinish(session, exitstatus):
                 "children": [],
                 "id_": "",
             }
-            send_discovery_message(os.fsdecode(cwd), error_node)
+            post_response(os.fsdecode(cwd), error_node)
     else:
         if exitstatus == 0 or exitstatus == 1:
             exitstatus_bool = "success"
@@ -430,7 +435,7 @@ def pytest_sessionfinish(session, exitstatus):
             )
             exitstatus_bool = "error"
 
-            send_execution_message(
+            execution_post(
                 os.fsdecode(cwd),
                 exitstatus_bool,
                 None,
@@ -484,7 +489,7 @@ def pytest_sessionfinish(session, exitstatus):
             result=file_coverage_map,
             error=None,
         )
-        send_message(payload)
+        send_post_request(payload)
 
 
 def build_test_tree(session: pytest.Session) -> TestNode:
@@ -852,10 +857,8 @@ def get_node_path(node: Any) -> pathlib.Path:
 atexit.register(lambda: __writer.close() if __writer else None)
 
 
-def send_execution_message(
-    cwd: str, status: Literal["success", "error"], tests: TestRunResultDict | None
-):
-    """Sends message execution payload details.
+def execution_post(cwd: str, status: Literal["success", "error"], tests: TestRunResultDict | None):
+    """Sends a POST request with execution payload details.
 
     Args:
         cwd (str): Current working directory.
@@ -867,10 +870,10 @@ def send_execution_message(
     )
     if ERRORS:
         payload["error"] = ERRORS
-    send_message(payload)
+    send_post_request(payload)
 
 
-def send_discovery_message(cwd: str, session_node: TestNode) -> None:
+def post_response(cwd: str, session_node: TestNode) -> None:
     """
     Sends a POST request with test session details in payload.
 
@@ -886,7 +889,7 @@ def send_discovery_message(cwd: str, session_node: TestNode) -> None:
     }
     if ERRORS is not None:
         payload["error"] = ERRORS
-    send_message(payload, cls_encoder=PathEncoder)
+    send_post_request(payload, cls_encoder=PathEncoder)
 
 
 class PathEncoder(json.JSONEncoder):
@@ -898,7 +901,7 @@ def default(self, o):
         return super().default(o)
 
 
-def send_message(
+def send_post_request(
     payload: ExecutionPayloadDict | DiscoveryPayloadDict | CoveragePayloadDict,
     cls_encoder=None,
 ):
@@ -923,7 +926,8 @@ def send_message(
 
     if __writer is None:
         try:
-            __writer = open(TEST_RUN_PIPE, "w", encoding="utf-8", newline="\r\n")  # noqa: SIM115, PTH123
+            __writer = socket_manager.PipeManager(TEST_RUN_PIPE)
+            __writer.connect()
         except Exception as error:
             error_msg = f"Error attempting to connect to extension named pipe {TEST_RUN_PIPE}[vscode-pytest]: {error}"
             print(error_msg, file=sys.stderr)
@@ -941,11 +945,10 @@ def send_message(
         "params": payload,
     }
     data = json.dumps(rpc, cls=cls_encoder)
+
     try:
         if __writer:
-            request = f"""content-length: {len(data)}\ncontent-type: application/json\n\n{data}"""
-            __writer.write(request)
-            __writer.flush()
+            __writer.write(data)
         else:
             print(
                 f"Plugin error connection error[vscode-pytest], writer is None \n[vscode-pytest] data: \n{data} \n",