Coverage for /Users/rik/.vscode/extensions/ms-python.python-2026.4.0-darwin-arm64/python_files/vscode_pytest/__init__.py: 5%
512 statements
« prev ^ index » next coverage.py v7.8.2, created at 2026-07-13 13:53 +0200
1# Copyright (c) Microsoft Corporation. All rights reserved.
2# Licensed under the MIT License.
4from __future__ import annotations
6import atexit
7import contextlib
8import json
9import os
10import pathlib
11import sys
12import traceback
13from typing import (
14 TYPE_CHECKING,
15 Any,
16 Dict,
17 Generator,
18 Literal,
19 Protocol,
20 TypedDict,
21 cast,
22)
24import pytest
26if TYPE_CHECKING:
27 from pluggy import Result
28 from typing_extensions import NotRequired
30USES_PYTEST_DESCRIBE = False
32with contextlib.suppress(ImportError):
33 from pytest_describe.plugin import DescribeBlock
35 USES_PYTEST_DESCRIBE = True
38class HasPathOrFspath(Protocol):
39 """Protocol defining objects that have either a path or fspath attribute."""
41 path: pathlib.Path | None = None
42 fspath: Any | None = None
45class TestData(TypedDict):
46 """A general class that all test objects inherit from."""
48 name: str
49 path: pathlib.Path
50 type_: Literal["class", "function", "file", "folder", "test", "error"]
51 id_: str
54class TestItem(TestData):
55 """A class defining test items."""
57 lineno: str
58 runID: str
61class TestNode(TestData):
62 """A general class that handles all test data which contains children."""
64 children: list[TestNode | TestItem | None]
65 lineno: NotRequired[str] # Optional field for class/function nodes
68class VSCodePytestError(Exception):
69 """A custom exception class for pytest errors."""
71 def __init__(self, message):
72 super().__init__(message)
75ERRORS = []
76IS_DISCOVERY = False
77map_id_to_path = {}
78collected_tests_so_far = set()
79TEST_RUN_PIPE = os.getenv("TEST_RUN_PIPE")
80PROJECT_ROOT_PATH = os.getenv(
81 "PROJECT_ROOT_PATH"
82) # Path to project root for multi-project workspaces
83SYMLINK_PATH = None
84INCLUDE_BRANCHES = False
86# Performance optimization caches for path resolution
87_path_cache: dict[int, pathlib.Path] = {} # Cache node paths by object id
88_path_to_str_cache: dict[pathlib.Path, str] = {} # Cache path-to-string conversions
89_CACHED_CWD: pathlib.Path | None = None
92def get_test_root_path() -> pathlib.Path:
93 """Get the root path for the test tree.
95 For project-based testing, this returns PROJECT_ROOT_PATH (the project root).
96 For legacy mode, this returns the current working directory.
98 Returns:
99 pathlib.Path: The root path to use for the test tree.
100 """
101 if PROJECT_ROOT_PATH:
102 return pathlib.Path(PROJECT_ROOT_PATH)
103 return pathlib.Path.cwd()
106def pytest_load_initial_conftests(early_config, parser, args): # noqa: ARG001
107 has_pytest_cov = early_config.pluginmanager.hasplugin(
108 "pytest_cov"
109 ) or early_config.pluginmanager.hasplugin("pytest_cov.plugin")
110 has_cov_arg = any("--cov" in arg for arg in args)
111 if has_cov_arg and not has_pytest_cov: 111 ↛ 112line 111 didn't jump to line 112 because the condition on line 111 was never true
112 raise VSCodePytestError(
113 "\n \nERROR: pytest-cov is not installed, please install this before running pytest with coverage as pytest-cov is required. \n"
114 )
115 if "--cov-branch" in args: 115 ↛ 120line 115 didn't jump to line 120 because the condition on line 115 was always true
116 global INCLUDE_BRANCHES
117 INCLUDE_BRANCHES = True
119 global TEST_RUN_PIPE
120 TEST_RUN_PIPE = os.getenv("TEST_RUN_PIPE")
121 error_string = (
122 "PYTEST ERROR: TEST_RUN_PIPE is not set at the time of pytest starting. "
123 "Please confirm this environment variable is not being changed or removed "
124 "as it is required for successful test discovery and execution."
125 f"TEST_RUN_PIPE = {TEST_RUN_PIPE}\n"
126 )
127 if not TEST_RUN_PIPE: 127 ↛ 128line 127 didn't jump to line 128 because the condition on line 127 was never true
128 print(error_string, file=sys.stderr)
129 if "--collect-only" in args: 129 ↛ 134line 129 didn't jump to line 134 because the condition on line 129 was always true
130 global IS_DISCOVERY
131 IS_DISCOVERY = True
133 # check if --rootdir is in the args
134 for arg in args:
135 if "--rootdir=" in arg:
136 rootdir = pathlib.Path(arg.split("--rootdir=")[1])
137 if not rootdir.exists(): 137 ↛ 138line 137 didn't jump to line 138 because the condition on line 137 was never true
138 raise VSCodePytestError(
139 f"The path set in the argument --rootdir={rootdir} does not exist."
140 )
142 # Check if the rootdir is a symlink or a child of a symlink to the current cwd.
143 is_symlink = False
145 if rootdir.is_symlink(): 145 ↛ 146line 145 didn't jump to line 146 because the condition on line 145 was never true
146 is_symlink = True
147 print(
148 f"Plugin info[vscode-pytest]: rootdir argument, {rootdir}, is identified as a symlink."
149 )
150 elif rootdir.resolve() != rootdir: 150 ↛ 151line 150 didn't jump to line 151 because the condition on line 150 was never true
151 print("Plugin info[vscode-pytest]: Checking if rootdir is a child of a symlink.")
152 is_symlink = has_symlink_parent(rootdir)
153 if is_symlink: 153 ↛ 154line 153 didn't jump to line 154 because the condition on line 153 was never true
154 print(
155 f"Plugin info[vscode-pytest]: rootdir argument, {rootdir}, is identified as a symlink or child of a symlink, adjusting pytest paths accordingly.",
156 )
157 global SYMLINK_PATH
158 SYMLINK_PATH = rootdir
161def pytest_internalerror(excrepr, excinfo): # noqa: ARG001
162 """A pytest hook that is called when an internal error occurs.
164 Keyword arguments:
165 excrepr -- the exception representation.
166 excinfo -- the exception information of type ExceptionInfo.
167 """
168 # call.excinfo.exconly() returns the exception as a string.
169 ERRORS.append(excinfo.exconly() + "\n Check Python Logs for more details.")
172def pytest_exception_interact(node, call, report):
173 """A pytest hook that is called when an exception is raised which could be handled.
175 Keyword arguments:
176 node -- the node that raised the exception.
177 call -- the call object.
178 report -- the report object of either type CollectReport or TestReport.
179 """
180 # call.excinfo is the captured exception of the call, if it raised as type ExceptionInfo.
181 # call.excinfo.exconly() returns the exception as a string.
182 # If it is during discovery, then add the error to error logs.
183 if IS_DISCOVERY:
184 if call.excinfo and call.excinfo.typename != "AssertionError":
185 if report.outcome == "skipped" and "SkipTest" in str(call):
186 return
187 ERRORS.append(call.excinfo.exconly() + "\n Check Python Logs for more details.")
188 else:
189 ERRORS.append(report.longreprtext + "\n Check Python Logs for more details.")
190 else:
191 # If during execution, send this data that the given node failed.
192 report_value = "error"
193 if call.excinfo.typename == "AssertionError":
194 report_value = "failure"
195 node_id = get_absolute_test_id(node.nodeid, get_node_path(node))
196 if node_id not in collected_tests_so_far:
197 collected_tests_so_far.add(node_id)
198 item_result = create_test_outcome(
199 node_id,
200 report_value,
201 "Test failed with exception",
202 report.longreprtext,
203 )
204 collected_test = TestRunResultDict()
205 collected_test[node_id] = item_result
206 cwd = pathlib.Path.cwd()
207 send_execution_message(
208 os.fsdecode(cwd),
209 "success",
210 collected_test or None,
211 )
214def has_symlink_parent(current_path):
215 """Recursively checks if any parent directories of the given path are symbolic links."""
216 # Convert the current path to an absolute Path object
217 curr_path = pathlib.Path(current_path)
218 print("Checking for symlink parent starting at current path: ", curr_path)
220 # Iterate over all parent directories
221 for parent in curr_path.parents:
222 # Check if the parent directory is a symlink
223 if parent.is_symlink():
224 print(f"Symlink found at: {parent}")
225 return True
226 return False
229def get_absolute_test_id(test_id: str, test_path: pathlib.Path) -> str:
230 """A function that returns the absolute test id.
232 This is necessary because testIds are relative to the rootdir.
233 This does not work for our case since testIds when referenced during run time are relative to the instantiation
234 location. Absolute paths for testIds are necessary for the test tree ensures configurations that change the rootdir
235 of pytest are handled correctly.
237 Keyword arguments:
238 test_id -- the pytest id of the test which is relative to the rootdir.
239 testPath -- the path to the file the test is located in, as a pathlib.Path object.
240 """
241 split_id = test_id.split("::")[1:]
242 return "::".join([str(test_path), *split_id])
245def pytest_keyboard_interrupt(excinfo):
246 """A pytest hook that is called when a keyboard interrupt is raised.
248 Keyword arguments:
249 excinfo -- the exception information of type ExceptionInfo.
250 """
251 # The function execonly() returns the exception as a string.
252 ERRORS.append(excinfo.exconly() + "\n Check Python Logs for more details.")
255class TestOutcome(Dict):
256 """A class that handles outcome for a single test.
258 for pytest the outcome for a test is only 'passed', 'skipped' or 'failed'
259 """
261 test: str
262 outcome: Literal["success", "failure", "skipped", "error"]
263 message: str | None
264 traceback: str | None
265 subtest: str | None
268def create_test_outcome(
269 testid: str,
270 outcome: str,
271 message: str | None,
272 traceback: str | None,
273 subtype: str | None = None, # noqa: ARG001
274) -> TestOutcome:
275 """A function that creates a TestOutcome object."""
276 return TestOutcome(
277 test=testid,
278 outcome=outcome,
279 message=message,
280 traceback=traceback, # TODO: traceback
281 subtest=None,
282 )
285class TestRunResultDict(Dict[str, Dict[str, TestOutcome]]):
286 """A class that stores all test run results."""
288 outcome: str
289 tests: dict[str, TestOutcome]
292@pytest.hookimpl(hookwrapper=True, trylast=True)
293def pytest_report_teststatus(report, config): # noqa: ARG001
294 """A pytest hook that is called when a test is called.
296 It is called 3 times per test, during setup, call, and teardown.
298 Keyword arguments:
299 report -- the report on the test setup, call, and teardown.
300 config -- configuration object.
301 """
302 cwd = pathlib.Path.cwd()
303 if SYMLINK_PATH:
304 cwd = SYMLINK_PATH
306 if report.when == "call" or (report.when == "setup" and report.skipped):
307 traceback = None
308 message = None
309 report_value = "skipped"
310 if report.passed:
311 report_value = "success"
312 elif report.failed:
313 report_value = "failure"
314 message = report.longreprtext
315 try:
316 node_path = map_id_to_path[report.nodeid]
317 except KeyError:
318 node_path = cwd
319 # Calculate the absolute test id and use this as the ID moving forward.
320 absolute_node_id = get_absolute_test_id(report.nodeid, node_path)
321 if absolute_node_id not in collected_tests_so_far:
322 collected_tests_so_far.add(absolute_node_id)
323 item_result = create_test_outcome(
324 absolute_node_id,
325 report_value,
326 message,
327 traceback,
328 )
329 collected_test = TestRunResultDict()
330 collected_test[absolute_node_id] = item_result
331 send_execution_message(
332 os.fsdecode(cwd),
333 "success",
334 collected_test or None,
335 )
336 yield
339ERROR_MESSAGE_CONST = {
340 2: "Pytest was unable to start or run any tests due to issues with test discovery or test collection.",
341 3: "Pytest was interrupted by the user, for example by pressing Ctrl+C during test execution.",
342 4: "Pytest encountered an internal error or exception during test execution.",
343 5: "Pytest was unable to find any tests to run.",
344}
347@pytest.hookimpl(hookwrapper=True, trylast=True)
348def pytest_runtest_protocol(item, nextitem): # noqa: ARG001
349 map_id_to_path[item.nodeid] = get_node_path(item)
350 skipped = check_skipped_wrapper(item)
351 if skipped:
352 absolute_node_id = get_absolute_test_id(item.nodeid, get_node_path(item))
353 report_value = "skipped"
354 cwd = pathlib.Path.cwd()
355 if absolute_node_id not in collected_tests_so_far:
356 collected_tests_so_far.add(absolute_node_id)
357 item_result = create_test_outcome(
358 absolute_node_id,
359 report_value,
360 None,
361 None,
362 )
363 collected_test = TestRunResultDict()
364 collected_test[absolute_node_id] = item_result
365 send_execution_message(
366 os.fsdecode(cwd),
367 "success",
368 collected_test or None,
369 )
370 yield
373def check_skipped_wrapper(item):
374 """A function that checks if a test is skipped or not by check its markers and its parent markers.
376 Returns True if the test is marked as skipped at any level, False otherwise.
378 Keyword arguments:
379 item -- the pytest item object.
380 """
381 if item.own_markers and check_skipped_condition(item):
382 return True
383 parent = item.parent
384 while isinstance(parent, pytest.Class):
385 if parent.own_markers and check_skipped_condition(parent):
386 return True
387 parent = parent.parent
388 return False
391def check_skipped_condition(item):
392 """A helper function that checks if a item has a skip or a true skip condition.
394 Keyword arguments:
395 item -- the pytest item object.
396 """
397 for marker in item.own_markers:
398 # If the test is marked with skip then it will not hit the pytest_report_teststatus hook,
399 # therefore we need to handle it as skipped here.
400 skip_condition = False
401 if marker.name == "skipif":
402 skip_condition = any(marker.args)
403 if marker.name == "skip" or skip_condition:
404 return True
405 return False
408class FileCoverageInfo(TypedDict):
409 lines_covered: list[int]
410 lines_missed: list[int]
411 executed_branches: int
412 total_branches: int
415def pytest_sessionfinish(session, exitstatus):
416 """A pytest hook that is called after pytest has fulled finished.
418 Keyword arguments:
419 session -- the pytest session object.
420 exitstatus -- the status code of the session.
422 Exit code 0: All tests were collected and passed successfully
423 Exit code 1: Tests were collected and run but some of the tests failed
424 Exit code 2: Test execution was interrupted by the user
425 Exit code 3: Internal error happened while executing tests
426 Exit code 4: pytest command line usage error
427 Exit code 5: No tests were collected
428 """
429 # Get the root path for the test tree structure (not the CWD for test execution)
430 # This is PROJECT_ROOT_PATH in project-based mode, or cwd in legacy mode
431 test_root_path = get_test_root_path()
432 if SYMLINK_PATH:
433 print("Plugin warning[vscode-pytest]: SYMLINK set, adjusting test root path.")
434 test_root_path = pathlib.Path(SYMLINK_PATH)
436 if IS_DISCOVERY:
437 if not (exitstatus == 0 or exitstatus == 1 or exitstatus == 5):
438 error_node: TestNode = {
439 "name": "",
440 "path": test_root_path,
441 "type_": "error",
442 "children": [],
443 "id_": "",
444 }
445 send_discovery_message(os.fsdecode(test_root_path), error_node)
446 try:
447 session_node: TestNode | None = build_test_tree(session)
448 if not session_node:
449 raise VSCodePytestError(
450 "Something went wrong following pytest finish, \
451 no session node was created"
452 )
453 send_discovery_message(os.fsdecode(test_root_path), session_node)
454 except Exception as e:
455 ERRORS.append(
456 f"Error Occurred, traceback: {(traceback.format_exc() if e.__traceback__ else '')}"
457 )
458 error_node: TestNode = {
459 "name": "",
460 "path": test_root_path,
461 "type_": "error",
462 "children": [],
463 "id_": "",
464 }
465 send_discovery_message(os.fsdecode(test_root_path), error_node)
466 else:
467 if exitstatus == 0 or exitstatus == 1:
468 exitstatus_bool = "success"
469 else:
470 ERRORS.append(
471 f"Pytest exited with error status: {exitstatus}, {ERROR_MESSAGE_CONST[exitstatus]}"
472 )
473 exitstatus_bool = "error"
475 send_execution_message(
476 os.fsdecode(test_root_path),
477 exitstatus_bool,
478 None,
479 )
480 # send end of transmission token
482 # send coverage if enabled
483 is_coverage_run = os.environ.get("COVERAGE_ENABLED")
484 if is_coverage_run == "True":
485 # load the report and build the json result to return
486 import coverage
488 # insert "python_files/lib/python" into the path so packaging can be imported
489 python_files_dir = pathlib.Path(__file__).parent.parent
490 bundled_dir = pathlib.Path(python_files_dir / "lib" / "python")
491 sys.path.append(os.fspath(bundled_dir))
493 from packaging.version import Version
495 coverage_version = Version(coverage.__version__)
496 global INCLUDE_BRANCHES
497 # only include branches if coverage version is 7.7.0 or greater (as this was when the api saves)
498 if coverage_version < Version("7.7.0") and INCLUDE_BRANCHES:
499 print(
500 "Plugin warning[vscode-pytest]: Branch coverage not supported in this coverage versions < 7.7.0. Please upgrade coverage package if you would like to see branch coverage."
501 )
502 INCLUDE_BRANCHES = False
504 try:
505 from coverage.exceptions import NoSource
506 except ImportError:
507 from coverage.misc import NoSource
509 cov = coverage.Coverage()
510 cov.load()
512 file_set: set[str] = cov.get_data().measured_files()
513 file_coverage_map: dict[str, FileCoverageInfo] = {}
515 # remove files omitted per coverage report config if any
516 omit_files: list[str] | None = cov.config.report_omit
517 if omit_files is not None:
518 for pattern in omit_files:
519 for file in list(file_set):
520 if pathlib.Path(file).match(pattern):
521 file_set.remove(file)
523 for file in file_set:
524 try:
525 analysis = cov.analysis2(file)
526 taken_file_branches = 0
527 total_file_branches = -1
529 if INCLUDE_BRANCHES:
530 branch_stats: dict[int, tuple[int, int]] = cov.branch_stats(file)
531 total_file_branches = sum(
532 [total_exits for total_exits, _ in branch_stats.values()]
533 )
534 taken_file_branches = sum(
535 [taken_exits for _, taken_exits in branch_stats.values()]
536 )
538 except NoSource:
539 # as per issue 24308 this best way to handle this edge case
540 continue
541 except Exception as e:
542 print(
543 f"Plugin error[vscode-pytest]: Skipping analysis of file: {file} due to error: {e}"
544 )
545 continue
546 lines_executable = {int(line_no) for line_no in analysis[1]}
547 lines_missed = {int(line_no) for line_no in analysis[3]}
548 lines_covered = lines_executable - lines_missed
549 file_info: FileCoverageInfo = {
550 "lines_covered": list(lines_covered), # list of int
551 "lines_missed": list(lines_missed), # list of int
552 "executed_branches": taken_file_branches,
553 "total_branches": total_file_branches,
554 }
555 # convert relative path to absolute path
556 if not pathlib.Path(file).is_absolute():
557 file = str(pathlib.Path(file).resolve())
558 file_coverage_map[file] = file_info
560 payload: CoveragePayloadDict = CoveragePayloadDict(
561 coverage=True,
562 cwd=os.fspath(test_root_path),
563 result=file_coverage_map,
564 error=None,
565 )
566 send_message(payload)
569def construct_nested_folders(
570 file_nodes_dict: dict[str, TestNode],
571 session_node: TestNode,
572 session_children_dict: dict[str, TestNode],
573) -> dict[str, TestNode]:
574 """Iterate through all files and construct them into nested folders.
576 Keyword arguments:
577 file_nodes_dict -- Dictionary of all file nodes
578 session_node -- The session node that will be parent to the folder structure
579 session_children_dict -- Dictionary of session's children nodes indexed by ID
581 Returns:
582 dict[str, TestNode] -- Updated session_children_dict with folder nodes added
583 """
584 created_files_folders_dict: dict[str, TestNode] = {}
585 for file_node in file_nodes_dict.values():
586 # Iterate through all the files that exist and construct them into nested folders.
587 root_folder_node: TestNode
588 try:
589 root_folder_node: TestNode = build_nested_folders(
590 file_node, created_files_folders_dict, session_node
591 )
592 except ValueError:
593 # This exception is raised when the session node is not a parent of the file node.
594 print(
595 "[vscode-pytest]: Session path not a parent of test paths, adjusting session node to common parent."
596 )
597 file_path_str: str = str(file_node["path"])
598 session_path_str: str = str(session_node["path"])
599 common_parent = os.path.commonpath([file_path_str, session_path_str])
600 common_parent_path = pathlib.Path(common_parent)
601 print("[vscode-pytest]: Session node now set to: ", common_parent)
602 session_node["path"] = common_parent_path # pathlib.Path
603 session_node["id_"] = common_parent # str
604 session_node["name"] = common_parent_path.name # str
605 root_folder_node = build_nested_folders(
606 file_node, created_files_folders_dict, session_node
607 )
608 # The final folder we get to is the highest folder in the path
609 # and therefore we add this as a child to the session.
610 root_id = root_folder_node.get("id_")
611 if root_id and root_id not in session_children_dict:
612 session_children_dict[root_id] = root_folder_node
614 return session_children_dict
617def process_parameterized_test(
618 test_case: pytest.Item,
619 test_node: TestItem,
620 function_nodes_dict: dict[str, TestNode],
621 file_nodes_dict: dict[str, TestNode],
622) -> TestNode:
623 """Process a parameterized test case and create appropriate function nodes.
625 Keyword arguments:
626 test_case -- the parameterized pytest test case; must have callspec attribute
627 test_node -- the test node created from the test case
628 function_nodes_dict -- dictionary of function nodes indexed by ID
629 file_nodes_dict -- dictionary of file nodes indexed by path
631 Returns:
632 TestNode -- the node to use for further processing (function node or original test node)
633 """
634 function_name: str = ""
635 # parameterized test cases cut the repetitive part of the name off.
636 parent_part, parameterized_section = test_node["name"].split("[", 1)
637 test_node["name"] = "[" + parameterized_section
639 first_split = test_case.nodeid.rsplit(
640 "::", 1
641 ) # splits the parameterized test name from the rest of the nodeid
642 second_split = first_split[0].rsplit(
643 ".py", 1
644 ) # splits the file path from the rest of the nodeid
646 class_and_method = second_split[1] + "::" # This has "::" separator at both ends
647 # construct the parent id, so it is absolute path :: any class and method :: parent_part
648 parent_id = cached_fsdecode(get_node_path(test_case)) + class_and_method + parent_part
650 try:
651 function_name = test_case.originalname # type: ignore
652 except AttributeError: # actual error has occurred
653 ERRORS.append(
654 f"unable to find original name for {test_case.name} with parameterization detected."
655 )
656 raise VSCodePytestError(
657 "Unable to find original name for parameterized test case"
658 ) from None
660 function_test_node = function_nodes_dict.get(parent_id)
661 if function_test_node is None:
662 function_test_node = create_parameterized_function_node(
663 function_name, get_node_path(test_case), parent_id
664 )
665 function_nodes_dict[parent_id] = function_test_node
667 if test_node not in function_test_node["children"]:
668 function_test_node["children"].append(test_node)
670 # Check if the parent node of the function is file, if so create/add to this file node.
671 if isinstance(test_case.parent, pytest.File):
672 # calculate the parent path of the test case
673 parent_path = get_node_path(test_case.parent)
674 parent_path_key = cached_fsdecode(parent_path)
675 parent_test_case = file_nodes_dict.get(parent_path_key)
676 if parent_test_case is None:
677 parent_test_case = create_file_node(parent_path)
678 file_nodes_dict[parent_path_key] = parent_test_case
679 if function_test_node not in parent_test_case["children"]:
680 parent_test_case["children"].append(function_test_node)
682 # Return the function node as the test node to handle subsequent nesting
683 return function_test_node
686def build_test_tree(session: pytest.Session) -> TestNode:
687 """Builds a tree made up of testing nodes from the pytest session.
689 Keyword arguments:
690 session -- the pytest session object that contains test items.
692 Returns:
693 TestNode -- The root node of the constructed test tree.
694 """
695 session_node = create_session_node(session)
696 session_children_dict: dict[str, TestNode] = {}
697 file_nodes_dict: dict[str, TestNode] = {}
698 class_nodes_dict: dict[str, TestNode] = {}
699 function_nodes_dict: dict[str, TestNode] = {}
701 # Check to see if the global variable for symlink path is set
702 if SYMLINK_PATH:
703 session_node["path"] = SYMLINK_PATH
704 session_node["id_"] = os.fspath(SYMLINK_PATH)
706 for test_case in session.items:
707 test_node = create_test_node(test_case)
708 if hasattr(test_case, "callspec"): # This means it is a parameterized test.
709 # Process parameterized test and get the function node to use for further processing
710 test_node = process_parameterized_test(
711 test_case, test_node, function_nodes_dict, file_nodes_dict
712 )
713 if isinstance(test_case.parent, pytest.Class) or (
714 USES_PYTEST_DESCRIBE and isinstance(test_case.parent, DescribeBlock)
715 ):
716 case_iter = test_case.parent
717 node_child_iter = test_node
718 test_class_node: TestNode | None = None
719 while isinstance(case_iter, pytest.Class) or (
720 USES_PYTEST_DESCRIBE and isinstance(case_iter, DescribeBlock)
721 ):
722 # While the given node is a class, create a class and nest the previous node as a child.
723 test_class_node = class_nodes_dict.get(case_iter.nodeid)
724 if test_class_node is None:
725 test_class_node = create_class_node(case_iter)
726 class_nodes_dict[case_iter.nodeid] = test_class_node
727 # Check if the class already has the child node. This will occur if the test is parameterized.
728 if node_child_iter not in test_class_node["children"]:
729 test_class_node["children"].append(node_child_iter)
730 # Iterate up.
731 node_child_iter = test_class_node
732 case_iter = case_iter.parent
733 # Now the parent node is not a class node, it is a file node.
734 if case_iter:
735 parent_module = case_iter
736 else:
737 ERRORS.append(f"Test class {case_iter} has no parent")
738 break
739 parent_path = get_node_path(parent_module)
740 # Create a file node that has the last class as a child.
741 parent_path_key = cached_fsdecode(parent_path)
742 test_file_node = file_nodes_dict.get(parent_path_key)
743 if test_file_node is None:
744 test_file_node = create_file_node(parent_path)
745 file_nodes_dict[parent_path_key] = test_file_node
746 # Check if the class is already a child of the file node.
747 if test_class_node is not None and test_class_node not in test_file_node["children"]:
748 test_file_node["children"].append(test_class_node)
749 elif not hasattr(test_case, "callspec"):
750 # This includes test cases that are pytest functions or a doctests.
751 if test_case.parent is None:
752 ERRORS.append(f"Test case {test_case.name} has no parent")
753 continue
754 parent_path = get_node_path(
755 cast(
756 "pytest.Session | pytest.Item | pytest.File | pytest.Class | pytest.Module | HasPathOrFspath",
757 test_case.parent,
758 )
759 )
760 parent_path_key = cached_fsdecode(parent_path)
761 parent_test_case = file_nodes_dict.get(parent_path_key)
762 if parent_test_case is None:
763 parent_test_case = create_file_node(parent_path)
764 file_nodes_dict[parent_path_key] = parent_test_case
765 parent_test_case["children"].append(test_node)
766 # Process all files and construct them into nested folders
767 session_children_dict = construct_nested_folders(
768 file_nodes_dict, session_node, session_children_dict
769 )
770 session_node["children"] = list(session_children_dict.values())
771 return session_node
774def build_nested_folders(
775 file_node: TestNode,
776 created_files_folders_dict: dict[str, TestNode],
777 session_node: TestNode,
778) -> TestNode:
779 """Takes a file or folder and builds the nested folder structure for it.
781 Keyword arguments:
782 file_module -- the created module for the file we are nesting.
783 file_node -- the file node that we are building the nested folders for.
784 created_files_folders_dict -- Dictionary of all the folders and files that have been created where the key is the path.
785 session -- the pytest session object.
786 """
787 # check if session node is a parent of the file node, throw error if not.
788 session_node_path = session_node["path"]
789 is_relative = False
790 try:
791 is_relative = file_node["path"].is_relative_to(session_node_path)
792 except AttributeError:
793 is_relative = file_node["path"].relative_to(session_node_path)
794 if not is_relative:
795 # If the session node is not a parent of the file node, we need to find their common parent.
796 raise ValueError("session and file not relative to each other, fixing now....")
798 # Begin the iterator_path one level above the current file.
799 prev_folder_node = file_node
800 iterator_path = file_node["path"].parent
801 counter = 0
802 max_iter = 100
803 while iterator_path != session_node_path:
804 curr_folder_name = iterator_path.name
805 iterator_path_key = cached_fsdecode(iterator_path)
806 curr_folder_node = created_files_folders_dict.get(iterator_path_key)
807 if curr_folder_node is None:
808 curr_folder_node = create_folder_node(curr_folder_name, iterator_path)
809 created_files_folders_dict[iterator_path_key] = curr_folder_node
810 if prev_folder_node not in curr_folder_node["children"]:
811 curr_folder_node["children"].append(prev_folder_node)
812 iterator_path = iterator_path.parent
813 prev_folder_node = curr_folder_node
814 # Handles error where infinite loop occurs.
815 counter += 1
816 if counter > max_iter:
817 raise ValueError(
818 "[vscode-pytest]: Infinite loop occurred in build_nested_folders. iterator_path: ",
819 iterator_path,
820 "session_node_path: ",
821 session_node_path,
822 )
823 return prev_folder_node
826def create_test_node(
827 test_case: pytest.Item,
828) -> TestItem:
829 """Creates a test node from a pytest test case.
831 Keyword arguments:
832 test_case -- the pytest test case.
833 """
834 test_case_loc: str = (
835 str(test_case.location[1] + 1) if (test_case.location[1] is not None) else ""
836 )
837 absolute_test_id = get_absolute_test_id(test_case.nodeid, get_node_path(test_case))
838 return {
839 "name": test_case.name,
840 "path": get_node_path(test_case),
841 "lineno": test_case_loc,
842 "type_": "test",
843 "id_": absolute_test_id,
844 "runID": absolute_test_id,
845 }
848def create_session_node(session: pytest.Session) -> TestNode:
849 """Creates a session node from a pytest session.
851 Keyword arguments:
852 session -- the pytest session.
853 """
854 # Use PROJECT_ROOT_PATH if set (project-based testing), otherwise use session path (legacy)
855 node_path = pathlib.Path(PROJECT_ROOT_PATH) if PROJECT_ROOT_PATH else get_node_path(session)
856 return {
857 "name": node_path.name,
858 "path": node_path,
859 "type_": "folder",
860 "children": [],
861 "id_": os.fspath(node_path),
862 }
865def create_class_node(class_module: pytest.Class | DescribeBlock) -> TestNode:
866 """Creates a class node from a pytest class object.
868 Keyword arguments:
869 class_module -- the pytest object representing a class module.
870 """
871 # Get line number for the class definition
872 class_line = ""
873 try:
874 if hasattr(class_module, "obj"):
875 import inspect
877 _, lineno = inspect.getsourcelines(class_module.obj)
878 class_line = str(lineno)
879 except (OSError, TypeError):
880 # If we can't get the source lines, leave lineno empty
881 pass
883 return {
884 "name": class_module.name,
885 "path": get_node_path(class_module),
886 "type_": "class",
887 "children": [],
888 "id_": get_absolute_test_id(class_module.nodeid, get_node_path(class_module)),
889 "lineno": class_line,
890 }
893def create_parameterized_function_node(
894 function_name: str, test_path: pathlib.Path, function_id: str
895) -> TestNode:
896 """Creates a function node to be the parent for the parameterized test nodes.
898 Keyword arguments:
899 function_name -- the name of the function.
900 test_path -- the path to the test file.
901 function_id -- the previously constructed function id that fits the pattern- absolute path :: any class and method :: parent_part
902 must be edited to get a unique id for the function node.
903 """
904 return {
905 "name": function_name,
906 "path": test_path,
907 "type_": "function",
908 "children": [],
909 "id_": function_id,
910 }
913def create_file_node(calculated_node_path: pathlib.Path) -> TestNode:
914 """Creates a file node from a path which has already been calculated using the get_node_path function.
916 Keyword arguments:
917 calculated_node_path -- the pytest file path.
918 """
919 return {
920 "name": calculated_node_path.name,
921 "path": calculated_node_path,
922 "type_": "file",
923 "id_": os.fspath(calculated_node_path),
924 "children": [],
925 }
928def create_folder_node(folder_name: str, path_iterator: pathlib.Path) -> TestNode:
929 """Creates a folder node from a pytest folder name and its path.
931 Keyword arguments:
932 folderName -- the name of the folder.
933 path_iterator -- the path of the folder.
934 """
935 return {
936 "name": folder_name,
937 "path": path_iterator,
938 "type_": "folder",
939 "id_": os.fspath(path_iterator),
940 "children": [],
941 }
944class DiscoveryPayloadDict(TypedDict):
945 """A dictionary that is used to send a post request to the server."""
947 cwd: str
948 status: Literal["success", "error"]
949 tests: TestNode | None
950 error: list[str] | None
953class ExecutionPayloadDict(Dict):
954 """A dictionary that is used to send a execution post request to the server."""
956 cwd: str
957 status: Literal["success", "error"]
958 result: TestRunResultDict | None
959 not_found: list[str] | None # Currently unused need to check
960 error: str | None # Currently unused need to check
963class CoveragePayloadDict(Dict):
964 """A dictionary that is used to send a execution post request to the server."""
966 coverage: bool
967 cwd: str
968 result: dict[str, FileCoverageInfo] | None
969 error: str | None # Currently unused need to check
972def cached_fsdecode(path: pathlib.Path) -> str:
973 """Convert path to string with caching for performance.
975 This function caches path-to-string conversions to avoid redundant
976 os.fsdecode() calls during test tree building.
978 Parameters:
979 path: The pathlib.Path object to convert to string.
981 Returns:
982 str: The string representation of the path.
983 """
984 if path not in _path_to_str_cache:
985 _path_to_str_cache[path] = os.fspath(path)
986 return _path_to_str_cache[path]
989def get_node_path(
990 node: pytest.Session
991 | pytest.Item
992 | pytest.File
993 | pytest.Class
994 | pytest.Module
995 | HasPathOrFspath,
996) -> pathlib.Path:
997 """A function that returns the path of a node given the switch to pathlib.Path.
999 It also evaluates if the node is a symlink and returns the equivalent path.
1001 Parameters:
1002 node: A pytest object or any object that has a path or fspath attribute.
1003 Do NOT pass a pathlib.Path object directly; use it directly instead.
1005 Returns:
1006 pathlib.Path: The resolved path for the node.
1007 """
1008 cache_key = id(node)
1009 if cache_key in _path_cache:
1010 return _path_cache[cache_key]
1012 node_path = getattr(node, "path", None)
1013 if node_path is None:
1014 fspath = getattr(node, "fspath", None)
1015 node_path = pathlib.Path(fspath) if fspath is not None else None
1017 if not node_path:
1018 raise VSCodePytestError(
1019 f"Unable to find path for node: {node}, node.path: {node.path}, node.fspath: {node.fspath}"
1020 )
1022 # Check for the session node since it has the symlink already.
1023 if SYMLINK_PATH and not isinstance(node, pytest.Session):
1024 # Get relative between the cwd (resolved path) and the node path.
1025 try:
1026 # Check to see if the node path contains the symlink root already
1027 # Convert Path objects to strings for os.path.commonpath
1028 symlink_str: str = str(SYMLINK_PATH)
1029 node_path_str: str = str(node_path)
1030 common_path = os.path.commonpath([symlink_str, node_path_str])
1031 if common_path == os.fsdecode(SYMLINK_PATH):
1032 # The node path is already relative to the SYMLINK_PATH root therefore return
1033 result = node_path
1034 else:
1035 # If the node path is not a symlink, then we need to calculate the equivalent symlink path
1036 # get the relative path between the cwd and the node path (as the node path is not a symlink).
1037 # Use cached cwd to avoid repeated system calls
1038 global _CACHED_CWD
1039 if _CACHED_CWD is None:
1040 _CACHED_CWD = pathlib.Path.cwd()
1041 rel_path = node_path.relative_to(_CACHED_CWD)
1042 # combine the difference between the cwd and the node path with the symlink path
1043 result = pathlib.Path(SYMLINK_PATH, rel_path)
1044 except Exception as e:
1045 raise VSCodePytestError(
1046 f"Error occurred while calculating symlink equivalent from node path: {e}"
1047 f"\n SYMLINK_PATH: {SYMLINK_PATH}, \n node path: {node_path}, \n cwd: {_CACHED_CWD or pathlib.Path.cwd()}"
1048 ) from e
1049 else:
1050 result = node_path
1052 # Cache before returning
1053 _path_cache[cache_key] = result
1054 return result
1057__writer = None
1058atexit.register(lambda: __writer.close() if __writer else None)
1061def send_execution_message(
1062 cwd: str, status: Literal["success", "error"], tests: TestRunResultDict | None
1063):
1064 """Sends message execution payload details.
1066 Args:
1067 cwd (str): Current working directory.
1068 status (Literal["success", "error"]): Execution status indicating success or error.
1069 tests (Union[testRunResultDict, None]): Test run results, if available.
1070 """
1071 payload: ExecutionPayloadDict = ExecutionPayloadDict(
1072 cwd=cwd, status=status, result=tests, not_found=None, error=None
1073 )
1074 if ERRORS:
1075 payload["error"] = ERRORS
1076 send_message(payload)
1079def send_discovery_message(cwd: str, session_node: TestNode) -> None:
1080 """
1081 Sends a POST request with test session details in payload.
1083 Args:
1084 cwd (str): Current working directory.
1085 session_node (TestNode): Node information of the test session.
1086 """
1087 payload: DiscoveryPayloadDict = {
1088 "cwd": cwd,
1089 "status": "success" if not ERRORS else "error",
1090 "tests": session_node,
1091 "error": [],
1092 }
1093 if ERRORS is not None:
1094 payload["error"] = ERRORS
1095 send_message(payload, cls_encoder=PathEncoder)
1098class PathEncoder(json.JSONEncoder):
1099 """A custom JSON encoder that encodes pathlib.Path objects as strings."""
1101 def default(self, o):
1102 if isinstance(o, pathlib.Path):
1103 return os.fspath(o)
1104 return super().default(o)
1107def send_message(
1108 payload: ExecutionPayloadDict | DiscoveryPayloadDict | CoveragePayloadDict,
1109 cls_encoder=None,
1110):
1111 """
1112 Sends a post request to the server.
1114 Keyword arguments:
1115 payload -- the payload data to be sent.
1116 cls_encoder -- a custom encoder if needed.
1117 """
1118 if not TEST_RUN_PIPE:
1119 error_msg = (
1120 "PYTEST ERROR: TEST_RUN_PIPE is not set at the time of pytest starting. "
1121 "Please confirm this environment variable is not being changed or removed "
1122 "as it is required for successful test discovery and execution."
1123 f"TEST_RUN_PIPE = {TEST_RUN_PIPE}\n"
1124 )
1125 print(error_msg, file=sys.stderr)
1126 raise VSCodePytestError(error_msg)
1128 global __writer
1130 if __writer is None:
1131 try:
1132 __writer = open(TEST_RUN_PIPE, "wb") # noqa: SIM115, PTH123
1133 except Exception as error:
1134 error_msg = f"Error attempting to connect to extension named pipe {TEST_RUN_PIPE}[vscode-pytest]: {error}"
1135 print(error_msg, file=sys.stderr)
1136 print(
1137 "If you are on a Windows machine, this error may be occurring if any of your tests clear environment variables"
1138 " as they are required to communicate with the extension. Please reference https://docs.pytest.org/en/stable/how-to/monkeypatch.html#monkeypatching-environment-variables"
1139 "for the correct way to clear environment variables during testing.\n",
1140 file=sys.stderr,
1141 )
1142 __writer = None
1143 raise VSCodePytestError(error_msg) from error
1145 rpc = {
1146 "jsonrpc": "2.0",
1147 "params": payload,
1148 }
1149 data = json.dumps(rpc, cls=cls_encoder)
1150 try:
1151 if __writer:
1152 request = (
1153 f"""content-length: {len(data)}\r\ncontent-type: application/json\r\n\r\n{data}"""
1154 )
1155 size = 4096
1156 encoded = request.encode("utf-8")
1157 bytes_written = 0
1158 while bytes_written < len(encoded):
1159 segment = encoded[bytes_written : bytes_written + size]
1160 bytes_written += __writer.write(segment)
1161 __writer.flush()
1162 else:
1163 print(
1164 f"Plugin error connection error[vscode-pytest], writer is None \n[vscode-pytest] data: \n{data} \n",
1165 file=sys.stderr,
1166 )
1167 except Exception as error:
1168 print(
1169 f"Plugin error, exception thrown while attempting to send data[vscode-pytest]: {error} \n[vscode-pytest] data: \n{data}\n",
1170 file=sys.stderr,
1171 )
1174class DeferPlugin:
1175 @pytest.hookimpl(hookwrapper=True)
1176 def pytest_xdist_auto_num_workers(
1177 self, config: pytest.Config
1178 ) -> Generator[None, Result[int], None]:
1179 """Determine how many workers to use based on how many tests were selected in the test explorer."""
1180 outcome = yield
1181 result = min(outcome.get_result(), len(config.option.file_or_dir))
1182 if result == 1:
1183 result = 0
1184 outcome.force_result(result)
1187def pytest_plugin_registered(plugin: object, manager: pytest.PytestPluginManager):
1188 plugin_name = "vscode_xdist"
1189 if ( 1189 ↛ 1197line 1189 didn't jump to line 1197 because the condition on line 1189 was never true
1190 # only register the plugin if xdist is enabled:
1191 manager.hasplugin("xdist")
1192 # prevent infinite recursion:
1193 and not isinstance(plugin, DeferPlugin)
1194 # prevent this plugin from being registered multiple times:
1195 and not manager.hasplugin(plugin_name)
1196 ):
1197 manager.register(DeferPlugin(), name=plugin_name)