Coverage for coverage / patch.py: 62.651%

1# Licensed under the Apache License: http://www.apache.org/licenses/LICENSE-2.0 

2# For details: https://github.com/coveragepy/coveragepy/blob/main/NOTICE.txt 

3 

4"""Invasive patches for coverage.py.""" 

5 

6from __future__ import annotations 69 ctx1bcdeUVfghiWXjklmYZnopq01rstu23vwxyzA45%BCDEFG67'HIJKLM89(NOPQRS!#)aT$

7 

8import contextlib 69 ctx1bcdeUVfghiWXjklmYZnopq01rstu23vwxyzA45%BCDEFG67'HIJKLM89(NOPQRS!#)aT$

9import os 69 ctx1bcdeUVfghiWXjklmYZnopq01rstu23vwxyzA45%BCDEFG67'HIJKLM89(NOPQRS!#)aT$

10from typing import TYPE_CHECKING, Any, NoReturn 69 ctx1bcdeUVfghiWXjklmYZnopq01rstu23vwxyzA45%BCDEFG67'HIJKLM89(NOPQRS!#)aT$

11 

12from coverage import env 69 ctx1bcdeUVfghiWXjklmYZnopq01rstu23vwxyzA45%BCDEFG67'HIJKLM89(NOPQRS!#)aT$

13from coverage.debug import DevNullDebug 69 ctx1bcdeUVfghiWXjklmYZnopq01rstu23vwxyzA45%BCDEFG67'HIJKLM89(NOPQRS!#)aT$

14from coverage.exceptions import ConfigError, CoverageException 69 ctx1bcdeUVfghiWXjklmYZnopq01rstu23vwxyzA45%BCDEFG67'HIJKLM89(NOPQRS!#)aT$

15 

16if TYPE_CHECKING: 69 ctx1bcdeUVfghiWXjklmYZnopq01rstu23vwxyzA45%BCDEFG67'HIJKLM89(NOPQRS!#)aT$

17 from coverage import Coverage 

18 from coverage.config import CoverageConfig 

19 from coverage.types import TDebugCtl 

20 

21 

22def apply_patches( 69 ctx1bcdeUVfghiWXjklmYZnopq01rstu23vwxyzA45%BCDEFG67'HIJKLM89(NOPQRS!#)aT$

23 cov: Coverage, 

24 config: CoverageConfig, 

25 debug: TDebugCtl, 

26) -> None: 

27 """Apply invasive patches requested by `[run] patch=`.""" 

28 debug = debug if debug.should("patch") else DevNullDebug() 69 ctx1bcdeUVfghiWXjklmYZnopq01rstu23vwxyzA45%BCDEFG67'HIJKLM89(NOPQRS!#)aT$

29 for patch in sorted(set(config.patch)): 69 ctx1bcdeUVfghiWXjklmYZnopq01rstu23vwxyzA45%BCDEFG67'HIJKLM89(NOPQRS!#)aT$

30 match patch: 65 ctx1bcdeUVfghiWXjklmYZnopq01rstu23vwxyzA45BCDEFG67HIJKLM89NOPQRS!#aT$

31 case "_exit": 65 ctx1bcdeUVfghiWXjklmYZnopq01rstu23vwxyzA45BCDEFG67HIJKLM89NOPQRS!#aT$

32 _patch__exit(cov, debug) 65 ctx1bcdeUVfghiWXjklmYZnopq01rstu23vwxyzA45BCDEFG67HIJKLM89NOPQRS!#aT$

33 

34 case "execv": 65 ctx1bcdeUVfghiWXjklmYZnopq01rstu23vwxyzA45BCDEFG67HIJKLM89NOPQRS!#aT$

35 _patch_execv(cov, config, debug) 46 ctx1bcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSaT

36 

37 case "fork": 65 ctx1bcdeUVfghiWXjklmYZnopq01rstu23vwxyzA45BCDEFG67HIJKLM89NOPQRS!#aT$

38 _patch_fork(debug) 1 ctx1a

39 

40 case "subprocess": 65 ctx1bcdeUVfghiWXjklmYZnopq01rstu23vwxyzA45BCDEFG67HIJKLM89NOPQRS!#aT$

41 _patch_subprocess(config, debug) 46 ctx1bcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSaT

42 

43 case _: 65 ctx1bcdeUVfghiWXjklmYZnopq01rstu23vwxyzA45BCDEFG67HIJKLM89NOPQRS!#aT$

44 raise ConfigError(f"Unknown patch {patch!r}") 65 ctx1bcdeUVfghiWXjklmYZnopq01rstu23vwxyzA45BCDEFG67HIJKLM89NOPQRS!#aT$

45 

46 

47def _patch__exit(cov: Coverage, debug: TDebugCtl) -> None: 69 ctx1bcdeUVfghiWXjklmYZnopq01rstu23vwxyzA45%BCDEFG67'HIJKLM89(NOPQRS!#)aT$

48 """Patch os._exit.""" 

49 debug.write("Patching _exit") 69 ctx1bcdeUVfghiWXjklmYZnopq01rstu23vwxyzA45%BCDEFG67'HIJKLM89(NOPQRS!#)aT$

50 

51 old_exit = os._exit 69 ctx1bcdeUVfghiWXjklmYZnopq01rstu23vwxyzA45%BCDEFG67'HIJKLM89(NOPQRS!#)aT$

52 

53 def coverage_os_exit_patch(status: int) -> NoReturn: 69 ctx1bcdeUVfghiWXjklmYZnopq01rstu23vwxyzA45%BCDEFG67'HIJKLM89(NOPQRS!#)aT$

54 with contextlib.suppress(Exception): 

55 debug.write(f"Using _exit patch with {cov = }") 

56 with contextlib.suppress(Exception): 

57 cov.save() 

58 old_exit(status) 

59 

60 os._exit = coverage_os_exit_patch 69 ctx1bcdeUVfghiWXjklmYZnopq01rstu23vwxyzA45%BCDEFG67'HIJKLM89(NOPQRS!#)aT$

61 

62 

63def _patch_execv(cov: Coverage, config: CoverageConfig, debug: TDebugCtl) -> None: 69 ctx1bcdeUVfghiWXjklmYZnopq01rstu23vwxyzA45%BCDEFG67'HIJKLM89(NOPQRS!#)aT$

64 """Patch the execv family of functions.""" 

65 if env.WINDOWS: 65 ↛ 66line 65 didn't jump to line 66 because the condition on line 65 was never true46 ctx1bcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSaT

66 raise CoverageException("patch=execv isn't supported yet on Windows.") 

67 

68 debug.write("Patching execv") 46 ctx1bcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSaT

69 

70 def make_execv_patch(fname: str, old_execv: Any) -> Any: 46 ctx1bcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSaT

71 def coverage_execv_patch(*args: Any, **kwargs: Any) -> Any: 46 ctx1bcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSaT

72 with contextlib.suppress(Exception): 

73 debug.write(f"Using execv patch for {fname} with {cov = }") 

74 with contextlib.suppress(Exception): 

75 cov.save() 

76 

77 if fname.endswith("e"): 

78 # Assume the `env` argument is passed positionally. 

79 new_env = args[-1] 

80 # Pass our configuration in the new environment. 

81 new_env["COVERAGE_PROCESS_CONFIG"] = config.serialize() 

82 if env.TESTING: 

83 # The subprocesses need to use the same core as the main process. 

84 new_env["COVERAGE_CORE"] = os.getenv("COVERAGE_CORE") 

85 

86 # When testing locally, we need to honor the pyc file location 

87 # or they get written to the .tox directories and pollute the 

88 # next run with a different core. 

89 if (cache_prefix := os.getenv("PYTHONPYCACHEPREFIX")) is not None: 

90 new_env["PYTHONPYCACHEPREFIX"] = cache_prefix 

91 

92 # Without this, it fails on PyPy and Ubuntu. 

93 new_env["PATH"] = os.getenv("PATH") 

94 old_execv(*args, **kwargs) 

95 

96 return coverage_execv_patch 46 ctx1bcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSaT

97 

98 # All the exec* and spawn* functions eventually call execv or execve. 

99 os.execv = make_execv_patch("execv", os.execv) 46 ctx1bcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSaT

100 os.execve = make_execv_patch("execve", os.execve) 46 ctx1bcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSaT

101 

102 

103def _patch_fork(debug: TDebugCtl) -> None: 69 ctx1bcdeUVfghiWXjklmYZnopq01rstu23vwxyzA45%BCDEFG67'HIJKLM89(NOPQRS!#)aT$

104 """Ensure Coverage is properly reset after a fork.""" 

105 from coverage.control import _after_fork_in_child 

106 

107 if env.WINDOWS: 

108 raise CoverageException("patch=fork isn't supported yet on Windows.") 

109 

110 debug.write("Patching fork") 

111 os.register_at_fork(after_in_child=_after_fork_in_child) 

112 

113 

114def _patch_subprocess(config: CoverageConfig, debug: TDebugCtl) -> None: 69 ctx1bcdeUVfghiWXjklmYZnopq01rstu23vwxyzA45%BCDEFG67'HIJKLM89(NOPQRS!#)aT$

115 """Write .pth files and set environment vars to measure subprocesses.""" 

116 debug.write("Patching subprocess") 46 ctx1bcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSaT

117 assert config.config_file is not None 46 ctx1bcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSaT

118 os.environ["COVERAGE_PROCESS_CONFIG"] = config.serialize() 46 ctx1bcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSaT