Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.8/site-packages/azure/core/pipeline/policies/_utils.py: 39%

1# -------------------------------------------------------------------------- 

2# 

3# Copyright (c) Microsoft Corporation. All rights reserved. 

4# 

5# The MIT License (MIT) 

6# 

7# Permission is hereby granted, free of charge, to any person obtaining a copy 

8# of this software and associated documentation files (the ""Software""), to 

9# deal in the Software without restriction, including without limitation the 

10# rights to use, copy, modify, merge, publish, distribute, sublicense, and/or 

11# sell copies of the Software, and to permit persons to whom the Software is 

12# furnished to do so, subject to the following conditions: 

13# 

14# The above copyright notice and this permission notice shall be included in 

15# all copies or substantial portions of the Software. 

16# 

17# THE SOFTWARE IS PROVIDED *AS IS*, WITHOUT WARRANTY OF ANY KIND, EXPRESS OR 

18# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, 

19# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE 

20# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER 

21# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING 

22# FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS 

23# IN THE SOFTWARE. 

24# 

25# -------------------------------------------------------------------------- 

26import datetime 

27import email.utils 

28from typing import Optional, cast, Union 

29from urllib.parse import urlparse 

30 

31from azure.core.pipeline.transport import ( 

32 HttpResponse as LegacyHttpResponse, 

33 AsyncHttpResponse as LegacyAsyncHttpResponse, 

34 HttpRequest as LegacyHttpRequest, 

35) 

36from azure.core.rest import HttpResponse, AsyncHttpResponse, HttpRequest 

37 

38 

39from ...utils._utils import _FixedOffset, case_insensitive_dict 

40from .. import PipelineResponse 

41 

42AllHttpResponseType = Union[HttpResponse, LegacyHttpResponse, AsyncHttpResponse, LegacyAsyncHttpResponse] 

43HTTPRequestType = Union[HttpRequest, LegacyHttpRequest] 

44 

45 

46def _parse_http_date(text: str) -> datetime.datetime: 

47 """Parse a HTTP date format into datetime. 

48 

49 :param str text: Text containing a date in HTTP format 

50 :rtype: datetime.datetime 

51 :return: The parsed datetime 

52 """ 

53 parsed_date = email.utils.parsedate_tz(text) 

54 if not parsed_date: 

55 raise ValueError("Invalid HTTP date") 

56 tz_offset = cast(int, parsed_date[9]) # Look at the code, tz_offset is always an int, at worst 0 

57 return datetime.datetime(*parsed_date[:6], tzinfo=_FixedOffset(tz_offset / 60)) 

58 

59 

60def parse_retry_after(retry_after: str) -> float: 

61 """Helper to parse Retry-After and get value in seconds. 

62 

63 :param str retry_after: Retry-After header 

64 :rtype: float 

65 :return: Value of Retry-After in seconds. 

66 """ 

67 delay: float # Using the Mypy recommendation to use float for "int or float" 

68 try: 

69 delay = float(retry_after) 

70 except ValueError: 

71 # Not an integer? Try HTTP date 

72 retry_date = _parse_http_date(retry_after) 

73 delay = (retry_date - datetime.datetime.now(retry_date.tzinfo)).total_seconds() 

74 return max(0, delay) 

75 

76 

77def get_retry_after(response: PipelineResponse[HTTPRequestType, AllHttpResponseType]) -> Optional[float]: 

78 """Get the value of Retry-After in seconds. 

79 

80 :param response: The PipelineResponse object 

81 :type response: ~azure.core.pipeline.PipelineResponse 

82 :return: Value of Retry-After in seconds. 

83 :rtype: float or None 

84 """ 

85 headers = case_insensitive_dict(response.http_response.headers) 

86 retry_after = headers.get("retry-after") 

87 if retry_after: 

88 return parse_retry_after(retry_after) 

89 for ms_header in ["retry-after-ms", "x-ms-retry-after-ms"]: 

90 retry_after = headers.get(ms_header) 

91 if retry_after: 

92 parsed_retry_after = parse_retry_after(retry_after) 

93 return parsed_retry_after / 1000.0 

94 return None 

95 

96 

97def get_domain(url: str) -> str: 

98 """Get the domain of an url. 

99 

100 :param str url: The url. 

101 :rtype: str 

102 :return: The domain of the url. 

103 """ 

104 return str(urlparse(url).netloc).lower()