Support EOT for testing

pythonFiles/testing_tools/socket_manager.py

@@ -23,6 +23,12 @@ def __init__(self, addr):
         self.socket = None
 
     def __enter__(self):
+        return self.connect()
+
+    def __exit__(self, *_):
+        self.close()
+
+    def connect(self):
         self.socket = socket.socket(
             socket.AF_INET, socket.SOCK_STREAM, socket.IPPROTO_TCP
         )
@@ -35,7 +41,7 @@ def __enter__(self):
 
         return self
 
-    def __exit__(self, *_):
+    def close(self):
         if self.socket:
             try:
                 self.socket.shutdown(socket.SHUT_RDWR)

pythonFiles/tests/pytestadapter/helpers.py

@@ -12,6 +12,10 @@
 import uuid
 from typing import Any, Dict, List, Optional, Tuple
 
+script_dir = pathlib.Path(__file__).parent.parent.parent
+sys.path.append(os.fspath(script_dir))
+sys.path.append(os.fspath(script_dir / "lib" / "python"))
+
 TEST_DATA_PATH = pathlib.Path(__file__).parent / ".data"
 from typing_extensions import TypedDict
 

pythonFiles/tests/pytestadapter/test_discovery.py

@@ -29,15 +29,26 @@ def test_import_error(tmp_path):
     temp_dir.mkdir()
     p = temp_dir / "error_pytest_import.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:
-        assert all(item in actual for item in ("status", "cwd", "error"))
-        assert actual["status"] == "error"
-        assert actual["cwd"] == os.fspath(TEST_DATA_PATH)
-        assert len(actual["error"]) == 2
+    actual: Optional[List[Dict[str, Any]]] = runner(["--collect-only", os.fspath(p)])
+    assert actual
+    actual_list: List[Dict[str, Any]] = actual
+    if actual_list is not None:
+        assert actual_list.pop(-1).get("eot")
+        for actual_item in actual_list:
+            assert all(
+                item in actual_item.keys() for item in ("status", "cwd", "error")
+            )
+            assert actual_item.get("status") == "error"
+            assert actual_item.get("cwd") == os.fspath(TEST_DATA_PATH)
+
+            # Ensure that 'error' is a list and then check its length
+            error_content = actual_item.get("error")
+            if error_content is not None and isinstance(
+                error_content, (list, tuple, str)
+            ):  # You can add other types if needed
+                assert len(error_content) == 2
+            else:
+                assert False
 
 
 def test_syntax_error(tmp_path):
@@ -60,13 +71,25 @@ def test_syntax_error(tmp_path):
     p = temp_dir / "error_syntax_discovery.py"
     shutil.copyfile(file_path, p)
     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)
-        assert len(actual["error"]) == 2
+    assert actual
+    actual_list: List[Dict[str, Any]] = actual
+    if actual_list is not None:
+        assert actual_list.pop(-1).get("eot")
+        for actual_item in actual_list:
+            assert all(
+                item in actual_item.keys() for item in ("status", "cwd", "error")
+            )
+            assert actual_item.get("status") == "error"
+            assert actual_item.get("cwd") == os.fspath(TEST_DATA_PATH)
+
+            # Ensure that 'error' is a list and then check its length
+            error_content = actual_item.get("error")
+            if error_content is not None and isinstance(
+                error_content, (list, tuple, str)
+            ):  # You can add other types if needed
+                assert len(error_content) == 2
+            else:
+                assert False
 
 
 def test_parameterized_error_collect():
@@ -76,12 +99,25 @@ def test_parameterized_error_collect():
     """
     file_path_str = "error_parametrize_discovery.py"
     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)
-        assert len(actual["error"]) == 2
+    assert actual
+    actual_list: List[Dict[str, Any]] = actual
+    if actual_list is not None:
+        assert actual_list.pop(-1).get("eot")
+        for actual_item in actual_list:
+            assert all(
+                item in actual_item.keys() for item in ("status", "cwd", "error")
+            )
+            assert actual_item.get("status") == "error"
+            assert actual_item.get("cwd") == os.fspath(TEST_DATA_PATH)
+
+            # Ensure that 'error' is a list and then check its length
+            error_content = actual_item.get("error")
+            if error_content is not None and isinstance(
+                error_content, (list, tuple, str)
+            ):  # You can add other types if needed
+                assert len(error_content) == 2
+            else:
+                assert False
 
 
 @pytest.mark.parametrize(
@@ -146,13 +182,16 @@ def test_pytest_collect(file, expected_const):
             os.fspath(TEST_DATA_PATH / file),
         ]
     )
-    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)
-        assert actual["tests"] == expected_const
+
+    assert actual
+    actual_list: List[Dict[str, Any]] = actual
+    if actual_list is not None:
+        assert actual_list.pop(-1).get("eot")
+        actual_item = actual_list.pop(0)
+        assert all(item in actual_item.keys() for item in ("status", "cwd", "error"))
+        assert actual_item.get("status") == "success"
+        assert actual_item.get("cwd") == os.fspath(TEST_DATA_PATH)
+        assert actual_item.get("tests") == expected_const
 
 
 def test_pytest_root_dir():
@@ -168,14 +207,16 @@ def test_pytest_root_dir():
         ],
         TEST_DATA_PATH / "root",
     )
-    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 / "root")
+    assert actual
+    actual_list: List[Dict[str, Any]] = actual
+    if actual_list is not None:
+        assert actual_list.pop(-1).get("eot")
+        actual_item = actual_list.pop(0)
+        assert all(item in actual_item.keys() for item in ("status", "cwd", "error"))
+        assert actual_item.get("status") == "success"
+        assert actual_item.get("cwd") == os.fspath(TEST_DATA_PATH / "root")
         assert (
-            actual["tests"]
+            actual_item.get("tests")
             == expected_discovery_test_output.root_with_config_expected_output
         )
 
@@ -193,13 +234,15 @@ def test_pytest_config_file():
         ],
         TEST_DATA_PATH / "root",
     )
-    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 / "root")
+    assert actual
+    actual_list: List[Dict[str, Any]] = actual
+    if actual_list is not None:
+        assert actual_list.pop(-1).get("eot")
+        actual_item = actual_list.pop(0)
+        assert all(item in actual_item.keys() for item in ("status", "cwd", "error"))
+        assert actual_item.get("status") == "success"
+        assert actual_item.get("cwd") == os.fspath(TEST_DATA_PATH / "root")
         assert (
-            actual["tests"]
+            actual_item.get("tests")
             == expected_discovery_test_output.root_with_config_expected_output
         )

pythonFiles/tests/pytestadapter/test_execution.py

@@ -2,6 +2,7 @@
 # Licensed under the MIT License.
 import os
 import shutil
+from typing import Any, Dict, List
 
 import pytest
 
@@ -23,14 +24,19 @@ def test_config_file():
         expected_execution_test_output.config_file_pytest_expected_execution_output
     )
     assert actual
-    assert len(actual) == len(expected_const)
+    actual_list: List[Dict[str, Any]] = actual
+    assert actual_list.pop(-1).get("eot")
+    assert len(actual_list) == 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(new_cwd)
-        actual_result_dict.update(a["result"])
-    assert actual_result_dict == expected_const
+    if actual_list is not None:
+        for actual_item in actual_list:
+            assert all(
+                item in actual_item.keys() for item in ("status", "cwd", "result")
+            )
+            assert actual_item.get("status") == "success"
+            assert actual_item.get("cwd") == os.fspath(new_cwd)
+            actual_result_dict.update(actual_item["result"])
+        assert actual_result_dict == expected_const
 
 
 def test_rootdir_specified():
@@ -43,14 +49,19 @@ def test_rootdir_specified():
         expected_execution_test_output.config_file_pytest_expected_execution_output
     )
     assert actual
-    assert len(actual) == len(expected_const)
+    actual_list: List[Dict[str, Any]] = actual
+    assert actual_list.pop(-1).get("eot")
+    assert len(actual_list) == 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(new_cwd)
-        actual_result_dict.update(a["result"])
-    assert actual_result_dict == expected_const
+    if actual_list is not None:
+        for actual_item in actual_list:
+            assert all(
+                item in actual_item.keys() for item in ("status", "cwd", "result")
+            )
+            assert actual_item.get("status") == "success"
+            assert actual_item.get("cwd") == os.fspath(new_cwd)
+            actual_result_dict.update(actual_item["result"])
+        assert actual_result_dict == expected_const
 
 
 def test_syntax_error_execution(tmp_path):
@@ -73,13 +84,23 @@ def test_syntax_error_execution(tmp_path):
     p = temp_dir / "error_syntax_discovery.py"
     shutil.copyfile(file_path, p)
     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)
-        assert len(actual["error"]) == 1
+    assert actual
+    actual_list: List[Dict[str, Any]] = actual
+    assert actual_list.pop(-1).get("eot")
+    if actual_list is not None:
+        for actual_item in actual_list:
+            assert all(
+                item in actual_item.keys() for item in ("status", "cwd", "error")
+            )
+            assert actual_item.get("status") == "error"
+            assert actual_item.get("cwd") == os.fspath(TEST_DATA_PATH)
+            error_content = actual_item.get("error")
+            if error_content is not None and isinstance(
+                error_content, (list, tuple, str)
+            ):  # You can add other types if needed
+                assert len(error_content) == 1
+            else:
+                assert False
 
 
 def test_bad_id_error_execution():
@@ -88,13 +109,23 @@ def test_bad_id_error_execution():
     The json should still be returned but the errors list should be present.
     """
     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)
-        assert len(actual["error"]) == 1
+    assert actual
+    actual_list: List[Dict[str, Any]] = actual
+    assert actual_list.pop(-1).get("eot")
+    if actual_list is not None:
+        for actual_item in actual_list:
+            assert all(
+                item in actual_item.keys() for item in ("status", "cwd", "error")
+            )
+            assert actual_item.get("status") == "error"
+            assert actual_item.get("cwd") == os.fspath(TEST_DATA_PATH)
+            error_content = actual_item.get("error")
+            if error_content is not None and isinstance(
+                error_content, (list, tuple, str)
+            ):  # You can add other types if needed
+                assert len(error_content) == 1
+            else:
+                assert False
 
 
 @pytest.mark.parametrize(
@@ -195,7 +226,8 @@ def test_pytest_execution(test_ids, expected_const):
     3. uf_single_method_execution_expected_output: test run on a single method in a file.
     4. uf_non_adjacent_tests_execution_expected_output: test run on unittests in two files with single selection in test explorer.
     5. unit_pytest_same_file_execution_expected_output: test run on a file with both unittest and pytest tests.
-    6. dual_level_nested_folder_execution_expected_output: test run on a file with one test file at the top level and one test file in a nested folder.
+    6. dual_level_nested_folder_execution_expected_output: test run on a file with one test file
+    at the top level and one test file in a nested folder.
     7. double_nested_folder_expected_execution_output: test run on a double nested folder.
     8. parametrize_tests_expected_execution_output: test run on a parametrize test with 3 inputs.
     9. single_parametrize_tests_expected_execution_output: test run on single parametrize test.
@@ -205,18 +237,22 @@ def test_pytest_execution(test_ids, expected_const):
     Keyword arguments:
     test_ids -- an array of test_ids to run.
     expected_const -- a dictionary of the expected output from running pytest discovery on the files.
-    """  # noqa: E501
+    """
     args = test_ids
     actual = runner(args)
     assert actual
-    print(actual)
-    assert len(actual) == len(expected_const)
+    actual_list: List[Dict[str, Any]] = actual
+    assert actual_list.pop(-1).get("eot")
+    assert len(actual_list) == 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"])
+    if actual_list is not None:
+        for actual_item in actual_list:
+            assert all(
+                item in actual_item.keys() for item in ("status", "cwd", "result")
+            )
+            assert actual_item.get("status") == "success"
+            assert actual_item.get("cwd") == os.fspath(TEST_DATA_PATH)
+            actual_result_dict.update(actual_item["result"])
     for key in actual_result_dict:
         if (
             actual_result_dict[key]["outcome"] == "failure"