Coverage for /var/srv/projects/api.amasfac.comuna18.com/tmp/venv/lib/python3.9/site-packages/sentry_sdk/integrations/stdlib.py: 76%

1import os 

2import subprocess 

3import sys 

4import platform 

5 

6from sentry_sdk.hub import Hub 

7from sentry_sdk.integrations import Integration 

8from sentry_sdk.scope import add_global_event_processor 

9from sentry_sdk.tracing_utils import EnvironHeaders 

10from sentry_sdk.utils import capture_internal_exceptions, logger, safe_repr 

11 

12from sentry_sdk._types import MYPY 

13 

14if MYPY: 14 ↛ 15line 14 didn't jump to line 15, because the condition on line 14 was never true

15 from typing import Any 

16 from typing import Callable 

17 from typing import Dict 

18 from typing import Optional 

19 from typing import List 

20 

21 from sentry_sdk._types import Event, Hint 

22 

23 

24try: 

25 from httplib import HTTPConnection # type: ignore 

26except ImportError: 

27 from http.client import HTTPConnection 

28 

29 

30_RUNTIME_CONTEXT = { 

31 "name": platform.python_implementation(), 

32 "version": "%s.%s.%s" % (sys.version_info[:3]), 

33 "build": sys.version, 

34} 

35 

36 

37class StdlibIntegration(Integration): 

38 identifier = "stdlib" 

39 

40 @staticmethod 

41 def setup_once(): 

42 # type: () -> None 

43 _install_httplib() 

44 _install_subprocess() 

45 

46 @add_global_event_processor 

47 def add_python_runtime_context(event, hint): 

48 # type: (Event, Hint) -> Optional[Event] 

49 if Hub.current.get_integration(StdlibIntegration) is not None: 

50 contexts = event.setdefault("contexts", {}) 

51 if isinstance(contexts, dict) and "runtime" not in contexts: 

52 contexts["runtime"] = _RUNTIME_CONTEXT 

53 

54 return event 

55 

56 

57def _install_httplib(): 

58 # type: () -> None 

59 real_putrequest = HTTPConnection.putrequest 

60 real_getresponse = HTTPConnection.getresponse 

61 

62 def putrequest(self, method, url, *args, **kwargs): 

63 # type: (HTTPConnection, str, str, *Any, **Any) -> Any 

64 hub = Hub.current 

65 if hub.get_integration(StdlibIntegration) is None: 65 ↛ 66line 65 didn't jump to line 66, because the condition on line 65 was never true

66 return real_putrequest(self, method, url, *args, **kwargs) 

67 

68 host = self.host 

69 port = self.port 

70 default_port = self.default_port 

71 

72 real_url = url 

73 if not real_url.startswith(("http://", "https://")): 73 ↛ 81line 73 didn't jump to line 81, because the condition on line 73 was never false

74 real_url = "%s://%s%s%s" % ( 

75 default_port == 443 and "https" or "http", 

76 host, 

77 port != default_port and ":%s" % port or "", 

78 url, 

79 ) 

80 

81 span = hub.start_span(op="http", description="%s %s" % (method, real_url)) 

82 

83 span.set_data("method", method) 

84 span.set_data("url", real_url) 

85 

86 rv = real_putrequest(self, method, url, *args, **kwargs) 

87 

88 for key, value in hub.iter_trace_propagation_headers(span): 

89 logger.debug( 

90 "[Tracing] Adding `{key}` header {value} to outgoing request to {real_url}.".format( 

91 key=key, value=value, real_url=real_url 

92 ) 

93 ) 

94 self.putheader(key, value) 

95 

96 self._sentrysdk_span = span 

97 

98 return rv 

99 

100 def getresponse(self, *args, **kwargs): 

101 # type: (HTTPConnection, *Any, **Any) -> Any 

102 span = getattr(self, "_sentrysdk_span", None) 

103 

104 if span is None: 104 ↛ 105line 104 didn't jump to line 105, because the condition on line 104 was never true

105 return real_getresponse(self, *args, **kwargs) 

106 

107 rv = real_getresponse(self, *args, **kwargs) 

108 

109 span.set_data("status_code", rv.status) 

110 span.set_http_status(int(rv.status)) 

111 span.set_data("reason", rv.reason) 

112 span.finish() 

113 

114 return rv 

115 

116 HTTPConnection.putrequest = putrequest 

117 HTTPConnection.getresponse = getresponse 

118 

119 

120def _init_argument(args, kwargs, name, position, setdefault_callback=None): 

121 # type: (List[Any], Dict[Any, Any], str, int, Optional[Callable[[Any], Any]]) -> Any 

122 """ 

123 given (*args, **kwargs) of a function call, retrieve (and optionally set a 

124 default for) an argument by either name or position. 

125 

126 This is useful for wrapping functions with complex type signatures and 

127 extracting a few arguments without needing to redefine that function's 

128 entire type signature. 

129 """ 

130 

131 if name in kwargs: 

132 rv = kwargs[name] 

133 if setdefault_callback is not None: 

134 rv = setdefault_callback(rv) 

135 if rv is not None: 135 ↛ 148line 135 didn't jump to line 148, because the condition on line 135 was never false

136 kwargs[name] = rv 

137 elif position < len(args): 137 ↛ 144line 137 didn't jump to line 144, because the condition on line 137 was never false

138 rv = args[position] 

139 if setdefault_callback is not None: 139 ↛ 140line 139 didn't jump to line 140, because the condition on line 139 was never true

140 rv = setdefault_callback(rv) 

141 if rv is not None: 141 ↛ 148line 141 didn't jump to line 148, because the condition on line 141 was never false

142 args[position] = rv 

143 else: 

144 rv = setdefault_callback and setdefault_callback(None) 

145 if rv is not None: 

146 kwargs[name] = rv 

147 

148 return rv 

149 

150 

151def _install_subprocess(): 

152 # type: () -> None 

153 old_popen_init = subprocess.Popen.__init__ 

154 

155 def sentry_patched_popen_init(self, *a, **kw): 

156 # type: (subprocess.Popen[Any], *Any, **Any) -> None 

157 

158 hub = Hub.current 

159 if hub.get_integration(StdlibIntegration) is None: 159 ↛ 160line 159 didn't jump to line 160, because the condition on line 159 was never true

160 return old_popen_init(self, *a, **kw) 

161 

162 # Convert from tuple to list to be able to set values. 

163 a = list(a) 

164 

165 args = _init_argument(a, kw, "args", 0) or [] 

166 cwd = _init_argument(a, kw, "cwd", 9) 

167 

168 # if args is not a list or tuple (and e.g. some iterator instead), 

169 # let's not use it at all. There are too many things that can go wrong 

170 # when trying to collect an iterator into a list and setting that list 

171 # into `a` again. 

172 # 

173 # Also invocations where `args` is not a sequence are not actually 

174 # legal. They just happen to work under CPython. 

175 description = None 

176 

177 if isinstance(args, (list, tuple)) and len(args) < 100: 177 ↛ 181line 177 didn't jump to line 181, because the condition on line 177 was never false

178 with capture_internal_exceptions(): 

179 description = " ".join(map(str, args)) 

180 

181 if description is None: 181 ↛ 182line 181 didn't jump to line 182, because the condition on line 181 was never true

182 description = safe_repr(args) 

183 

184 env = None 

185 

186 with hub.start_span(op="subprocess", description=description) as span: 

187 

188 for k, v in hub.iter_trace_propagation_headers(span): 

189 if env is None: 189 ↛ 193line 189 didn't jump to line 193, because the condition on line 189 was never false

190 env = _init_argument( 

191 a, kw, "env", 10, lambda x: dict(x or os.environ) 

192 ) 

193 env["SUBPROCESS_" + k.upper().replace("-", "_")] = v 

194 

195 if cwd: 195 ↛ 198line 195 didn't jump to line 198, because the condition on line 195 was never false

196 span.set_data("subprocess.cwd", cwd) 

197 

198 rv = old_popen_init(self, *a, **kw) 

199 

200 span.set_tag("subprocess.pid", self.pid) 

201 return rv 

202 

203 subprocess.Popen.__init__ = sentry_patched_popen_init # type: ignore 

204 

205 old_popen_wait = subprocess.Popen.wait 

206 

207 def sentry_patched_popen_wait(self, *a, **kw): 

208 # type: (subprocess.Popen[Any], *Any, **Any) -> Any 

209 hub = Hub.current 

210 

211 if hub.get_integration(StdlibIntegration) is None: 211 ↛ 212line 211 didn't jump to line 212, because the condition on line 211 was never true

212 return old_popen_wait(self, *a, **kw) 

213 

214 with hub.start_span(op="subprocess.wait") as span: 

215 span.set_tag("subprocess.pid", self.pid) 

216 return old_popen_wait(self, *a, **kw) 

217 

218 subprocess.Popen.wait = sentry_patched_popen_wait # type: ignore 

219 

220 old_popen_communicate = subprocess.Popen.communicate 

221 

222 def sentry_patched_popen_communicate(self, *a, **kw): 

223 # type: (subprocess.Popen[Any], *Any, **Any) -> Any 

224 hub = Hub.current 

225 

226 if hub.get_integration(StdlibIntegration) is None: 226 ↛ 227line 226 didn't jump to line 227, because the condition on line 226 was never true

227 return old_popen_communicate(self, *a, **kw) 

228 

229 with hub.start_span(op="subprocess.communicate") as span: 

230 span.set_tag("subprocess.pid", self.pid) 

231 return old_popen_communicate(self, *a, **kw) 

232 

233 subprocess.Popen.communicate = sentry_patched_popen_communicate # type: ignore 

234 

235 

236def get_subprocess_traceparent_headers(): 

237 # type: () -> EnvironHeaders 

238 return EnvironHeaders(os.environ, prefix="SUBPROCESS_")