Add dynamic result

pythonFiles/tests/pytestadapter/helpers.py

@@ -1,5 +1,6 @@
 # Copyright (c) Microsoft Corporation. All rights reserved.
 # Licensed under the MIT License.
+
 import io
 import json
 import os
@@ -9,7 +10,7 @@
 import sys
 import threading
 import uuid
-from typing import Any, Dict, List, Optional, Tuple
+from typing import Any, Dict, List, Optional, Tuple, Union
 
 TEST_DATA_PATH = pathlib.Path(__file__).parent / ".data"
 from typing_extensions import TypedDict
@@ -72,21 +73,34 @@ def process_rpc_message(data: str) -> Tuple[Dict[str, Any], str]:
     str_stream: io.StringIO = io.StringIO(data)
 
     length: int = 0
+
     while True:
         line: str = str_stream.readline()
         if CONTENT_LENGTH.lower() in line.lower():
             length = int(line[len(CONTENT_LENGTH) :])
             break
+
         if not line or line.isspace():
             raise ValueError("Header does not contain Content-Length")
+
     while True:
         line: str = str_stream.readline()
         if not line or line.isspace():
             break
 
     raw_json: str = str_stream.read(length)
-    dict_json: Dict[str, Any] = json.loads(raw_json)
-    return dict_json, str_stream.read()
+    return json.loads(raw_json), str_stream.read()
+
+
+def process_rpc_json(data: str) -> List[Dict[str, Any]]:
+    """Process the JSON data which comes from the server which runs the pytest discovery."""
+    json_messages = []
+    remaining = data
+    while remaining:
+        json_data, remaining = process_rpc_message(remaining)
+        json_messages.append(json_data)
+
+    return json_messages
 
 
 def runner(args: List[str]) -> Optional[List[Dict[str, Any]]]:
@@ -110,53 +124,58 @@ def runner(args: List[str]) -> Optional[List[Dict[str, Any]]]:
             "PYTHONPATH": os.fspath(pathlib.Path(__file__).parent.parent.parent),
         }
     )
+    completed = threading.Event()
 
-    result: list = []
+    result = []
     t1: threading.Thread = threading.Thread(
-        target=_listen_on_socket, args=(listener, result)
+        target=_listen_on_socket, args=(listener, result, completed)
     )
     t1.start()
 
     t2 = threading.Thread(
-        target=lambda proc_args, proc_env, proc_cwd: subprocess.run(
-            proc_args, env=proc_env, cwd=proc_cwd
-        ),
-        args=(process_args, env, TEST_DATA_PATH),
+        target=_run_test_code,
+        args=(process_args, env, TEST_DATA_PATH, completed),
     )
     t2.start()
 
     t1.join()
     t2.join()
 
-    a = process_rpc_json(result[0])
-    return a if result else None
+    return process_rpc_json(result[0]) if result else None
 
 
-def process_rpc_json(data: str) -> List[Dict[str, Any]]:
-    """Process the JSON data which comes from the server which runs the pytest discovery."""
-    json_messages = []
-    remaining = data
-    while remaining:
-        json_data, remaining = process_rpc_message(remaining)
-        json_messages.append(json_data)
-
-    return json_messages
-
-
-def _listen_on_socket(listener: socket.socket, result: List[str]):
+def _listen_on_socket(
+    listener: socket.socket, result: List[str], completed: threading.Event
+):
     """Listen on the socket for the JSON data from the server.
-    Created as a seperate function for clarity in threading.
+    Created as a separate function for clarity in threading.
     """
     sock, (other_host, other_port) = listener.accept()
+    listener.settimeout(1)
     all_data: list = []
     while True:
         data: bytes = sock.recv(1024 * 1024)
         if not data:
-            break
+            if completed.is_set():
+                break
+            else:
+                try:
+                    sock, (other_host, other_port) = listener.accept()
+                except socket.timeout:
+                    result.append("".join(all_data))
+                    return
         all_data.append(data.decode("utf-8"))
     result.append("".join(all_data))
 
 
+def _run_test_code(
+    proc_args: List[str], proc_env, proc_cwd: str, completed: threading.Event
+):
+    result = subprocess.run(proc_args, env=proc_env, cwd=proc_cwd)
+    completed.set()
+    return result
+
+
 def find_test_line_number(test_name: str, test_file_path) -> str:
     """Function which finds the correct line number for a test by looking for the "test_marker--[test_name]" string.
 

pythonFiles/tests/pytestadapter/test_discovery.py

@@ -59,11 +59,10 @@ def test_syntax_error(tmp_path):
     temp_dir.mkdir()
     p = temp_dir / "error_syntax_discovery.py"
     shutil.copyfile(file_path, p)
-    actual_list: Optional[List[Dict[str, Any]]] = runner(
-        ["--collect-only", os.fspath(p)]
-    )
-    assert actual_list
-    for actual in actual_list:
+    actual = runner(["--collect-only", os.fspath(p)])
+    if actual:
+        actual = actual[0]
+        assert actual
         assert all(item in actual for item in ("status", "cwd", "error"))
         assert actual["status"] == "error"
         assert actual["cwd"] == os.fspath(TEST_DATA_PATH)
@@ -76,11 +75,9 @@ def test_parameterized_error_collect():
     The json should still be returned but the errors list should be present.
     """
     file_path_str = "error_parametrize_discovery.py"
-    actual_list: Optional[List[Dict[str, Any]]] = runner(
-        ["--collect-only", file_path_str]
-    )
-    assert actual_list
-    for actual in actual_list:
+    actual = runner(["--collect-only", file_path_str])
+    if actual:
+        actual = actual[0]
         assert all(item in actual for item in ("status", "cwd", "error"))
         assert actual["status"] == "error"
         assert actual["cwd"] == os.fspath(TEST_DATA_PATH)
@@ -135,14 +132,15 @@ def test_pytest_collect(file, expected_const):
     file -- a string with the file or folder to run pytest discovery on.
     expected_const -- the expected output from running pytest discovery on the file.
     """
-    actual_list: Optional[List[Dict[str, Any]]] = runner(
+    actual = runner(
         [
             "--collect-only",
             os.fspath(TEST_DATA_PATH / file),
         ]
     )
-    assert actual_list
-    for actual in actual_list:
+    if actual:
+        actual = actual[0]
+        assert actual
         assert all(item in actual for item in ("status", "cwd", "tests"))
         assert actual["status"] == "success"
         assert actual["cwd"] == os.fspath(TEST_DATA_PATH)

pythonFiles/tests/pytestadapter/test_execution.py

@@ -2,7 +2,6 @@
 # Licensed under the MIT License.
 import os
 import shutil
-from typing import Any, Dict, List, Optional
 
 import pytest
 from tests.pytestadapter import expected_execution_test_output
@@ -29,11 +28,10 @@ def test_syntax_error_execution(tmp_path):
     temp_dir.mkdir()
     p = temp_dir / "error_syntax_discovery.py"
     shutil.copyfile(file_path, p)
-    actual_list: Optional[List[Dict[str, Any]]] = runner(
-        ["error_syntax_discover.py::test_function"]
-    )
-    assert actual_list
-    for actual in actual_list:
+    actual = runner(["error_syntax_discover.py::test_function"])
+    if actual:
+        actual = actual[0]
+        assert actual
         assert all(item in actual for item in ("status", "cwd", "error"))
         assert actual["status"] == "error"
         assert actual["cwd"] == os.fspath(TEST_DATA_PATH)
@@ -45,9 +43,10 @@ def test_bad_id_error_execution():
 
     The json should still be returned but the errors list should be present.
     """
-    actual_list: Optional[List[Dict[str, Any]]] = runner(["not/a/real::test_id"])
-    assert actual_list
-    for actual in actual_list:
+    actual = runner(["not/a/real::test_id"])
+    if actual:
+        actual = actual[0]
+        assert actual
         assert all(item in actual for item in ("status", "cwd", "error"))
         assert actual["status"] == "error"
         assert actual["cwd"] == os.fspath(TEST_DATA_PATH)
@@ -156,14 +155,17 @@ def test_pytest_execution(test_ids, expected_const):
     expected_const -- a dictionary of the expected output from running pytest discovery on the files.
     """
     args = test_ids
-    actual_list: Optional[List[Dict[str, Any]]] = runner(args)
-    assert actual_list
-    for actual in actual_list:
-        assert all(item in actual for item in ("status", "cwd", "result"))
-        assert actual["status"] == "success"
-        assert actual["cwd"] == os.fspath(TEST_DATA_PATH)
-        result_data = actual["result"]
-        for key in result_data:
-            if result_data[key]["outcome"] == "failure":
-                result_data[key]["message"] = "ERROR MESSAGE"
-        assert result_data == expected_const
+    actual = runner(args)
+    assert actual
+    print(actual)
+    assert len(actual) == len(expected_const)
+    actual_result_dict = dict()
+    for a in actual:
+        assert all(item in a for item in ("status", "cwd", "result"))
+        assert a["status"] == "success"
+        assert a["cwd"] == os.fspath(TEST_DATA_PATH)
+        actual_result_dict.update(a["result"])
+    for key in actual_result_dict:
+        if actual_result_dict[key]["outcome"] == "failure":
+            actual_result_dict[key]["message"] = "ERROR MESSAGE"
+    assert actual_result_dict == expected_const

pythonFiles/tests/unittestadapter/test_execution.py

@@ -3,9 +3,14 @@
 
 import os
 import pathlib
+import sys
 from typing import List
 
 import pytest
+
+PYTHON_FILES = pathlib.Path(__file__).parent.parent
+
+sys.path.insert(0, os.fspath(PYTHON_FILES))
 from unittestadapter.execution import parse_execution_cli_args, run_tests
 
 TEST_DATA_PATH = pathlib.Path(__file__).parent / ".data"

pythonFiles/unittestadapter/discovery.py

@@ -17,8 +17,8 @@
 from typing_extensions import Literal
 
 # Add the path to pythonFiles to sys.path to find testing_tools.socket_manager.
-PYTHON_FILES = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
-sys.path.insert(0, PYTHON_FILES)
+PYTHON_FILES = pathlib.Path(__file__).parent.parent
+sys.path.insert(0, os.fspath(PYTHON_FILES))
 
 from testing_tools import socket_manager
 

pythonFiles/unittestadapter/execution.py

@@ -13,17 +13,14 @@
 from types import TracebackType
 from typing import Dict, List, Optional, Tuple, Type, Union
 
-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 process_json_util
-
+directory_path = pathlib.Path(__file__).parent.parent / "lib" / "python"
 # Add the path to pythonFiles to sys.path to find testing_tools.socket_manager.
-PYTHON_FILES = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
-sys.path.insert(0, PYTHON_FILES)
+PYTHON_FILES = pathlib.Path(__file__).parent.parent
+
+sys.path.insert(0, os.fspath(PYTHON_FILES))
 # Add the lib path to sys.path to find the typing_extensions module.
 sys.path.insert(0, os.path.join(PYTHON_FILES, "lib", "python"))
-from testing_tools import socket_manager
+from testing_tools import process_json_util, socket_manager
 from typing_extensions import NotRequired, TypeAlias, TypedDict
 from unittestadapter.utils import parse_unittest_args
 
@@ -54,6 +51,9 @@ def parse_execution_cli_args(
 ErrorType = Union[
     Tuple[Type[BaseException], BaseException, TracebackType], Tuple[None, None, None]
 ]
+PORT = 0
+UUID = 0
+START_DIR = ""
 
 
 class TestOutcomeEnum(str, enum.Enum):
@@ -148,8 +148,10 @@ def formatResult(
             "traceback": tb,
             "subtest": subtest.id() if subtest else None,
         }
-
         self.formatted[test_id] = result
+        if PORT == 0 or UUID == 0:
+            print("Error sending response, port or uuid unknown to python server.")
+        send_run_data(result, PORT, UUID)
 
 
 class TestExecutionStatus(str, enum.Enum):
@@ -225,6 +227,33 @@ def run_tests(
     return payload
 
 
+def send_run_data(raw_data, port, uuid):
+    # Build the request data (it has to be a POST request or the Node side will not process it), and send it.
+    status = raw_data["outcome"]
+    cwd = os.path.abspath(START_DIR)
+    if raw_data["subtest"]:
+        test_id = raw_data["subtest"]
+    else:
+        test_id = raw_data["test"]
+    test_dict = {}
+    test_dict[test_id] = raw_data
+    payload: PayloadDict = {"cwd": cwd, "status": status, "result": test_dict}
+    addr = ("localhost", port)
+    data = json.dumps(payload)
+    request = f"""Content-Length: {len(data)}
+Content-Type: application/json
+Request-uuid: {uuid}
+
+{data}"""
+    try:
+        with socket_manager.SocketManager(addr) as s:
+            if s.socket is not None:
+                s.socket.sendall(request.encode("utf-8"))
+    except Exception as e:
+        print(f"Error sending response: {e}")
+        print(f"Request data: {request}")
+
+
 if __name__ == "__main__":
     # Get unittest test execution arguments.
     argv = sys.argv[1:]
@@ -270,11 +299,11 @@ def run_tests(
         print(f"Error: Could not connect to runTestIdsPort: {e}")
         print("Error: Could not connect to runTestIdsPort")
 
-    port, uuid = parse_execution_cli_args(argv[:index])
+    PORT, UUID = parse_execution_cli_args(argv[:index])
     if test_ids_from_buffer:
         # Perform test execution.
         payload = run_tests(
-            start_dir, test_ids_from_buffer, pattern, top_level_dir, uuid
+            start_dir, test_ids_from_buffer, pattern, top_level_dir, UUID
         )
     else:
         cwd = os.path.abspath(start_dir)
@@ -284,19 +313,3 @@ def run_tests(
             "status": status,
             "error": "No test ids received from buffer",
         }
-
-    # Build the request data and send it.
-    addr = ("localhost", port)
-    data = json.dumps(payload)
-    request = f"""Content-Length: {len(data)}
-Content-Type: application/json
-Request-uuid: {uuid}
-
-{data}"""
-    try:
-        with socket_manager.SocketManager(addr) as s:
-            if s.socket is not None:
-                s.socket.sendall(request.encode("utf-8"))
-    except Exception as e:
-        print(f"Error sending response: {e}")
-        print(f"Request data: {request}")