GeezerVisonThe World Wide Web, restored to 800 pixels of glory. |
| Home|About|Gaming cabinet|Archives|How it works|Starter sites |
GeezerVison is showing github.com — vscode-python/python_files/unittestadapter/execution.py at main · Pythoneryeah/vscode-python Expand file tree/ Copy pathexecution.pyMore file actions 350 lines (298 loc) · 11.8 KB / Copy pathexecution.pyFile metadata and controls350 lines (298 loc) · 11.8 KB 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277 278 279 280 281 282 283 284 285 286 287 288 289 290 291 292 293 294 295 296 297 298 299 300 301 302 303 304 305 306 307 308 309 310 311 312 313 314 315 316 317 318 319 320 321 322 323 324 325 326 327 328 329 330 331 332 333 334 335 336 337 338 339 340 341 342 343 344 345 346 347 348 349 350 # Copyright (c) Microsoft Corporation. All rights reserved. # Licensed under the MIT License. import atexit import enum import json import os import pathlib import sys import sysconfig import traceback import unittest from types import TracebackType from typing import Dict, List, Optional, Tuple, Type, Union # Adds the scripts directory to the PATH as a workaround for enabling shell for test execution. path_var_name = "PATH" if "PATH" in os.environ else "Path" os.environ[path_var_name] = ( sysconfig.get_paths()["scripts"] + os.pathsep + os.environ[path_var_name] ) script_dir = pathlib.Path(__file__).parent.parent sys.path.append(os.fspath(script_dir)) sys.path.insert(0, os.fspath(script_dir / "lib" / "python")) from testing_tools import process_json_util, socket_manager # noqa: E402 from unittestadapter.pvsc_utils import ( # noqa: E402 EOTPayloadDict, ExecutionPayloadDict, TestExecutionStatus, VSCodeUnittestError, parse_unittest_args, send_post_request, ) ErrorType = Union[Tuple[Type[BaseException], BaseException, TracebackType], Tuple[None, None, None]] test_run_pipe = "" START_DIR = "" class TestOutcomeEnum(str, enum.Enum): error = "error" failure = "failure" success = "success" skipped = "skipped" expected_failure = "expected-failure" unexpected_success = "unexpected-success" subtest_success = "subtest-success" subtest_failure = "subtest-failure" class UnittestTestResult(unittest.TextTestResult): def __init__(self, *args, **kwargs): self.formatted: Dict[str, Dict[str, Union[str, None]]] = {} super().__init__(*args, **kwargs) def startTest(self, test: unittest.TestCase): # noqa: N802 super().startTest(test) def addError( # noqa: N802 self, test: unittest.TestCase, err: ErrorType, ): super().addError(test, err) self.formatResult(test, TestOutcomeEnum.error, err) def addFailure( # noqa: N802 self, test: unittest.TestCase, err: ErrorType, ): super().addFailure(test, err) self.formatResult(test, TestOutcomeEnum.failure, err) def addSuccess(self, test: unittest.TestCase): # noqa: N802 super().addSuccess(test) self.formatResult(test, TestOutcomeEnum.success) def addSkip(self, test: unittest.TestCase, reason: str): # noqa: N802 super().addSkip(test, reason) self.formatResult(test, TestOutcomeEnum.skipped) def addExpectedFailure(self, test: unittest.TestCase, err: ErrorType): # noqa: N802 super().addExpectedFailure(test, err) self.formatResult(test, TestOutcomeEnum.expected_failure, err) def addUnexpectedSuccess(self, test: unittest.TestCase): # noqa: N802 super().addUnexpectedSuccess(test) self.formatResult(test, TestOutcomeEnum.unexpected_success) def addSubTest( # noqa: N802 self, test: unittest.TestCase, subtest: unittest.TestCase, err: Union[ErrorType, None], ): super().addSubTest(test, subtest, err) self.formatResult( test, TestOutcomeEnum.subtest_failure if err else TestOutcomeEnum.subtest_success, err, subtest, ) def formatResult( # noqa: N802 self, test: unittest.TestCase, outcome: str, error: Union[ErrorType, None] = None, subtest: Union[unittest.TestCase, None] = None, ): tb = None message = "" # error is a tuple of the form returned by sys.exc_info(): (type, value, traceback). if error is not None: try: message = f"{error[0]} {error[1]}" except Exception: message = "Error occurred, unknown type or value" formatted = traceback.format_exception(*error) tb = "".join(formatted) # Remove the 'Traceback (most recent call last)' formatted = formatted[1:] test_id = subtest.id() if subtest else test.id() result = { "test": test.id(), "outcome": outcome, "message": message, "traceback": tb, "subtest": subtest.id() if subtest else None, } self.formatted[test_id] = result test_run_pipe = os.getenv("TEST_RUN_PIPE") if not test_run_pipe: print( "UNITTEST ERROR: TEST_RUN_PIPE is not set at the time of unittest trying to send data. " f"TEST_RUN_PIPE = {test_run_pipe}\n", file=sys.stderr, ) raise VSCodeUnittestError( "UNITTEST ERROR: TEST_RUN_PIPE is not set at the time of unittest trying to send data. " ) send_run_data(result, test_run_pipe) def filter_tests(suite: unittest.TestSuite, test_ids: List[str]) -> unittest.TestSuite: """Filter the tests in the suite to only run the ones with the given ids.""" filtered_suite = unittest.TestSuite() for test in suite: if isinstance(test, unittest.TestCase): if test.id() in test_ids: filtered_suite.addTest(test) else: filtered_suite.addTest(filter_tests(test, test_ids)) return filtered_suite def get_all_test_ids(suite: unittest.TestSuite) -> List[str]: """Return a list of all test ids in the suite.""" test_ids = [] for test in suite: if isinstance(test, unittest.TestCase): test_ids.append(test.id()) else: test_ids.extend(get_all_test_ids(test)) return test_ids def find_missing_tests(test_ids: List[str], suite: unittest.TestSuite) -> List[str]: """Return a list of test ids that are not in the suite.""" all_test_ids = get_all_test_ids(suite) return [test_id for test_id in test_ids if test_id not in all_test_ids] # Args: start_path path to a directory or a file, list of ids that may be empty. # Edge cases: # - if tests got deleted since the VS Code side last ran discovery and the current test run, # return these test ids in the "not_found" entry, and the VS Code side can process them as "unknown"; # - if tests got added since the VS Code side last ran discovery and the current test run, ignore them. def run_tests( start_dir: str, test_ids: List[str], pattern: str, top_level_dir: Optional[str], verbosity: int, failfast: Optional[bool], locals_: Optional[bool] = None, ) -> ExecutionPayloadDict: cwd = os.path.abspath(start_dir) # noqa: PTH100 if "/" in start_dir: # is a subdir parent_dir = os.path.dirname(start_dir) # noqa: PTH120 sys.path.insert(0, parent_dir) else: sys.path.insert(0, cwd) status = TestExecutionStatus.error error = None payload: ExecutionPayloadDict = {"cwd": cwd, "status": status, "result": None} try: # If it's a file, split path and file name. start_dir = cwd if cwd.endswith(".py"): start_dir = os.path.dirname(cwd) # noqa: PTH120 pattern = os.path.basename(cwd) # noqa: PTH119 if failfast is None: failfast = False if locals_ is None: locals_ = False if verbosity is None: verbosity = 1 runner = unittest.TextTestRunner( resultclass=UnittestTestResult, tb_locals=locals_, failfast=failfast, verbosity=verbosity, ) # Discover tests at path with the file name as a pattern (if any). loader = unittest.TestLoader() suite = loader.discover(start_dir, pattern, top_level_dir) # lets try to tailer our own suite so we can figure out running only the ones we want tailor: unittest.TestSuite = filter_tests(suite, test_ids) # If any tests are missing, add them to the payload. not_found = find_missing_tests(test_ids, tailor) if not_found: missing_suite = loader.loadTestsFromNames(not_found) tailor.addTests(missing_suite) result: UnittestTestResult = runner.run(tailor) # type: ignore payload["result"] = result.formatted except Exception: status = TestExecutionStatus.error error = traceback.format_exc() if error is not None: payload["error"] = error else: status = TestExecutionStatus.success payload["status"] = status return payload __socket = None atexit.register(lambda: __socket.close() if __socket else None) def send_run_data(raw_data, test_run_pipe): status = raw_data["outcome"] cwd = os.path.abspath(START_DIR) # noqa: PTH100 test_id = raw_data["subtest"] or raw_data["test"] test_dict = {} test_dict[test_id] = raw_data payload: ExecutionPayloadDict = {"cwd": cwd, "status": status, "result": test_dict} send_post_request(payload, test_run_pipe) if __name__ == "__main__": # Get unittest test execution arguments. argv = sys.argv[1:] index = argv.index("--udiscovery") ( start_dir, pattern, top_level_dir, verbosity, failfast, locals_, ) = parse_unittest_args(argv[index + 1 :]) run_test_ids_pipe = os.environ.get("RUN_TEST_IDS_PIPE") test_run_pipe = os.getenv("TEST_RUN_PIPE") if not run_test_ids_pipe: print("Error[vscode-unittest]: RUN_TEST_IDS_PIPE env var is not set.") raise VSCodeUnittestError("Error[vscode-unittest]: RUN_TEST_IDS_PIPE env var is not set.") if not test_run_pipe: print("Error[vscode-unittest]: TEST_RUN_PIPE env var is not set.") raise VSCodeUnittestError("Error[vscode-unittest]: TEST_RUN_PIPE env var is not set.") test_ids_from_buffer = [] raw_json = None try: with socket_manager.PipeManager(run_test_ids_pipe) as sock: buffer: str = "" while True: # Receive the data from the client data: str = sock.read() if not data: break # Append the received data to the buffer buffer += data try: # Try to parse the buffer as JSON raw_json = process_json_util.process_rpc_json(buffer) # Clear the buffer as complete JSON object is received buffer = "" print("Received JSON data in run") break except json.JSONDecodeError: # JSON decoding error, the complete JSON object is not yet received continue except OSError as e: msg = f"Error: Could not connect to RUN_TEST_IDS_PIPE: {e}" print(msg) raise VSCodeUnittestError(msg) from e try: if raw_json and "params" in raw_json: test_ids_from_buffer = raw_json["params"] if test_ids_from_buffer: # Perform test execution. payload = run_tests( start_dir, test_ids_from_buffer, pattern, top_level_dir, verbosity, failfast, locals_, ) else: # No test ids received from buffer cwd = os.path.abspath(start_dir) # noqa: PTH100 status = TestExecutionStatus.error payload: ExecutionPayloadDict = { "cwd": cwd, "status": status, "error": "No test ids received from buffer", "result": None, } send_post_request(payload, test_run_pipe) except json.JSONDecodeError as exc: msg = "Error: Could not parse test ids from stdin" print(msg) raise VSCodeUnittestError(msg) from exc eot_payload: EOTPayloadDict = {"command_type": "execution", "eot": True} send_post_request(eot_payload, test_run_pipe) |
| You are visitor 01402520 — GeezerVison does not store the pages you look up. Best viewed with Netscape Navigator 4.7 or Internet Explorer 6 at 800×600. No cookies required. |