Switch to using named pipes for test communication

build/test-requirements.txt

@@ -19,3 +19,6 @@ torch-tb-profiler
 
 # extension build tests
 freezegun
+
+# testing custom pytest plugin require the use of named pipes
+namedpipe; platform_system == "Windows"

python_files/testing_tools/process_json_util.py

@@ -2,12 +2,12 @@
 # Licensed under the MIT License.
 import io
 import json
-from typing import List
+from typing import List, Dict
 
 CONTENT_LENGTH: str = "Content-Length:"
 
 
-def process_rpc_json(data: str) -> List[str]:
+def process_rpc_json(data: str) -> Dict[str, List[str]]:
     """Process the JSON data which comes from the server."""
     str_stream: io.StringIO = io.StringIO(data)
 
@@ -22,7 +22,7 @@ def process_rpc_json(data: str) -> List[str]:
         if not line or line.isspace():
             raise ValueError("Header does not contain Content-Length")
 
-    while True:
+    while True:  # keep reading until the number of bytes is the CONTENT_LENGTH
         line: str = str_stream.readline()
         if not line or line.isspace():
             break

python_files/testing_tools/socket_manager.py

@@ -4,6 +4,76 @@
 import socket
 import sys
 
+# set the socket before it gets blocked or overwritten by a user tests
+_SOCKET = socket.socket
+
+
+class PipeManager:
+    def __init__(self, name):
+        self.name = name
+
+    def __enter__(self):
+        return self.connect()
+
+    def __exit__(self, *_):
+        self.close()
+
+    def connect(self):
+        if sys.platform == "win32":
+            self._writer = open(self.name, "wt", encoding="utf-8")
+            # 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):
+        if sys.platform == "win32":
+            self._writer.close()
+        else:
+            # add exception catch
+            self._socket.close()
+
+    def write(self, data: str):
+        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.
+
+        Args:
+            bufsize (int): Number of bytes to read from the socket.
+
+        Returns:
+            data (str): Data received from the socket.
+        """
+        if sys.platform == "win32":
+            # returns a string automatically from read
+            if not hasattr(self, "_reader"):
+                self._reader = open(self.name, "rt", encoding="utf-8")
+            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(object):
     """Create a socket and connect to the given address.

python_files/tests/pytestadapter/.data/pytest.ini

@@ -0,0 +1,5 @@
+# Copyright (c) Microsoft Corporation. All rights reserved.
+# Licensed under the MIT License.
+
+# pytest.ini is specified here so the root directory of the tests is kept at .data instead of referencing
+# the parent python_files/pyproject.toml for test_discovery.py and test_execution.py for pytest-adapter tests.