(
self,
cnx: HTTPRequestsConnectionClass | HTTPSRequestsConnectionClass | None,
verb: str,
url: str,
requestHeaders: dict[str, str],
input: Any | None,
stream: bool = False,
follow_302_redirect: bool = False,
)
| 1231 | return status, responseHeaders, output |
| 1232 | |
| 1233 | def __requestRaw( |
| 1234 | self, |
| 1235 | cnx: HTTPRequestsConnectionClass | HTTPSRequestsConnectionClass | None, |
| 1236 | verb: str, |
| 1237 | url: str, |
| 1238 | requestHeaders: dict[str, str], |
| 1239 | input: Any | None, |
| 1240 | stream: bool = False, |
| 1241 | follow_302_redirect: bool = False, |
| 1242 | ) -> tuple[int, dict[str, Any], str | object]: |
| 1243 | self.__deferRequest(verb) |
| 1244 | |
| 1245 | try: |
| 1246 | original_cnx = cnx |
| 1247 | if cnx is None: |
| 1248 | cnx = self.__createConnection() |
| 1249 | cnx.request(verb, url, input, requestHeaders, stream) |
| 1250 | response = cnx.getresponse() |
| 1251 | output = response if stream else response.read() |
| 1252 | status = response.status |
| 1253 | responseHeaders = {k.lower(): v for k, v in response.getheaders()} |
| 1254 | |
| 1255 | if input: |
| 1256 | if isinstance(input, IOBase): |
| 1257 | input.close() |
| 1258 | |
| 1259 | self.__log(verb, url, requestHeaders, input, status, responseHeaders, output) |
| 1260 | |
| 1261 | if status == 202 and ( |
| 1262 | verb == "GET" or verb == "HEAD" |
| 1263 | ): # only for requests that are considered 'safe' in RFC 2616 |
| 1264 | time.sleep(Consts.PROCESSING_202_WAIT_TIME) |
| 1265 | return self.__requestRaw(original_cnx, verb, url, requestHeaders, input, stream=stream) |
| 1266 | |
| 1267 | if status == 301 and "location" in responseHeaders: |
| 1268 | location = responseHeaders["location"] |
| 1269 | o = urllib.parse.urlparse(location) |
| 1270 | if o.scheme != self.__scheme: |
| 1271 | raise RuntimeError( |
| 1272 | f"Github server redirected from {self.__scheme} protocol to {o.scheme}, " |
| 1273 | f"please correct your Github server URL via base_url: Github(base_url=...)" |
| 1274 | ) |
| 1275 | if o.hostname != self.__hostname: |
| 1276 | raise RuntimeError( |
| 1277 | f"Github server redirected from host {self.__hostname} to {o.hostname}, " |
| 1278 | f"please correct your Github server URL via base_url: Github(base_url=...)" |
| 1279 | ) |
| 1280 | if o.path == url: |
| 1281 | port = ":" + str(self.__port) if self.__port is not None else "" |
| 1282 | requested_location = f"{self.__scheme}://{self.__hostname}{port}{url}" |
| 1283 | raise RuntimeError( |
| 1284 | f"Requested {requested_location} but server redirected to {location}, " |
| 1285 | f"you may need to correct your Github server URL " |
| 1286 | f"via base_url: Github(base_url=...)" |
| 1287 | ) |
| 1288 | if self._logger.isEnabledFor(logging.INFO): |
| 1289 | self._logger.info(f"Following Github server redirection from {url} to {o.path}") |
| 1290 | return self.__requestRaw(original_cnx, verb, o.path, requestHeaders, input, stream=stream) |
no test coverage detected