Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/requests_mock/request.py: 54%

1# Licensed under the Apache License, Version 2.0 (the "License"); you may 

2# not use this file except in compliance with the License. You may obtain 

3# a copy of the License at 

4# 

5# https://www.apache.org/licenses/LICENSE-2.0 

6# 

7# Unless required by applicable law or agreed to in writing, software 

8# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT 

9# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the 

10# License for the specific language governing permissions and limitations 

11# under the License. 

12 

13import copy 

14import json 

15import urllib.parse 

16 

17import requests 

18 

19 

20class _RequestObjectProxy(object): 

21 """A wrapper around a requests.Request that gives some extra information. 

22 

23 This will be important both for matching and so that when it's save into 

24 the request_history users will be able to access these properties. 

25 """ 

26 

27 def __init__(self, request, **kwargs): 

28 self._request = request 

29 self._matcher = None 

30 self._url_parts_ = None 

31 self._qs = None 

32 

33 # All of these params should always exist but we use a default 

34 # to make the test setup easier. 

35 self._timeout = kwargs.pop('timeout', None) 

36 self._allow_redirects = kwargs.pop('allow_redirects', None) 

37 self._verify = kwargs.pop('verify', None) 

38 self._stream = kwargs.pop('stream', None) 

39 self._cert = kwargs.pop('cert', None) 

40 self._proxies = copy.deepcopy(kwargs.pop('proxies', {})) 

41 

42 # FIXME(jamielennox): This is part of bug #1584008 and should default 

43 # to True (or simply removed) in a major version bump. 

44 self._case_sensitive = kwargs.pop('case_sensitive', False) 

45 

46 def __getattr__(self, name): 

47 # there should be a better way to exclude this, but I don't want to 

48 # implement __setstate__ just not forward it to the request. You can't 

49 # actually define the method and raise AttributeError there either. 

50 if name in ('__setstate__',): 

51 raise AttributeError(name) 

52 

53 return getattr(self._request, name) 

54 

55 @property 

56 def _url_parts(self): 

57 if self._url_parts_ is None: 

58 url = self._request.url 

59 

60 if not self._case_sensitive: 

61 url = url.lower() 

62 

63 self._url_parts_ = urllib.parse.urlparse(url) 

64 

65 return self._url_parts_ 

66 

67 @property 

68 def scheme(self): 

69 return self._url_parts.scheme 

70 

71 @property 

72 def netloc(self): 

73 return self._url_parts.netloc 

74 

75 @property 

76 def hostname(self): 

77 try: 

78 return self.netloc.split(':')[0] 

79 except IndexError: 

80 return '' 

81 

82 @property 

83 def port(self): 

84 components = self.netloc.split(':') 

85 

86 try: 

87 return int(components[1]) 

88 except (IndexError, ValueError): 

89 pass 

90 

91 if self.scheme == 'https': 

92 return 443 

93 if self.scheme == 'http': 

94 return 80 

95 

96 # The default return shouldn't matter too much because if you are 

97 # wanting to test this value you really should be explicitly setting it 

98 # somewhere. 0 at least is a boolean False and an int. 

99 return 0 

100 

101 @property 

102 def path(self): 

103 return self._url_parts.path 

104 

105 @property 

106 def query(self): 

107 return self._url_parts.query 

108 

109 @property 

110 def qs(self): 

111 if self._qs is None: 

112 self._qs = urllib.parse.parse_qs(self.query, 

113 keep_blank_values=True) 

114 

115 return self._qs 

116 

117 @property 

118 def timeout(self): 

119 return self._timeout 

120 

121 @property 

122 def allow_redirects(self): 

123 return self._allow_redirects 

124 

125 @property 

126 def verify(self): 

127 return self._verify 

128 

129 @property 

130 def stream(self): 

131 return self._stream 

132 

133 @property 

134 def cert(self): 

135 return self._cert 

136 

137 @property 

138 def proxies(self): 

139 return self._proxies 

140 

141 @classmethod 

142 def _create(cls, *args, **kwargs): 

143 return cls(requests.Request(*args, **kwargs).prepare()) 

144 

145 @property 

146 def text(self): 

147 body = self.body 

148 

149 if isinstance(body, bytes): 

150 body = body.decode('utf-8') 

151 

152 return body 

153 

154 def json(self, **kwargs): 

155 return json.loads(self.text, **kwargs) 

156 

157 def __getstate__(self): 

158 # Can't pickle a weakref, but it's a weakref so ok to drop it. 

159 d = self.__dict__.copy() 

160 d['_matcher'] = None 

161 return d 

162 

163 @property 

164 def matcher(self): 

165 """The matcher that this request was handled by. 

166 

167 The matcher object is handled by a weakref. It will return the matcher 

168 object if it is still available - so if the mock is still in place. If 

169 the matcher is not available it will return None. 

170 """ 

171 # if unpickled or not from a response this will be None 

172 if self._matcher is None: 

173 return None 

174 

175 return self._matcher() 

176 

177 def __str__(self): 

178 return "{0.method} {0.url}".format(self._request)