Coverage for src/pytest_xray/xray_publisher.py: 91.5%

1import json 

2from typing import Any, Callable, Optional, Union 

3 

4import requests 

5from requests import PreparedRequest 

6from requests.auth import AuthBase 

7 

8from pytest_xray.constant import AUTHENTICATE_ENDPOINT 

9from pytest_xray.exceptions import XrayError 

10 

11AuthType = Optional[Union[tuple[str, str], AuthBase, Callable[[PreparedRequest], PreparedRequest]]] 

12 

13 

14# Connect timeout and read inactivity timeout, not a total wall-clock deadline. 

15REQUEST_TIMEOUT = (10, 60) 

16 

17 

18class ClientSecretAuth(AuthBase): 

19 """Bearer authentication with Client ID and a Client Secret.""" 

20 

21 def __init__(self, base_url: str, client_id: str, client_secret: str, verify: Union[bool, str] = True) -> None: 

22 if base_url.endswith('/'): 22 ↛ 23line 22 didn't jump to line 23 because the condition on line 22 was never true

23 base_url = base_url[:-1] 

24 self.base_url = base_url 

25 self.client_id = client_id 

26 self.client_secret = client_secret 

27 self.verify = verify 

28 

29 @property 

30 def endpoint_url(self) -> str: 

31 return f'{self.base_url}{AUTHENTICATE_ENDPOINT}' 

32 

33 def __call__(self, r: requests.PreparedRequest) -> requests.PreparedRequest: 

34 headers = {'Content-type': 'application/json', 'Accept': 'text/plain'} 

35 auth_data = {'client_id': self.client_id, 'client_secret': self.client_secret} 

36 

37 try: 

38 response = requests.post( 

39 self.endpoint_url, 

40 data=json.dumps(auth_data), 

41 headers=headers, 

42 verify=self.verify, 

43 timeout=REQUEST_TIMEOUT, 

44 allow_redirects=False, 

45 ) 

46 response.raise_for_status() 

47 if response.status_code != 200: 

48 raise XrayError('Authentication did not return HTTP 200') 

49 auth_token = response.json() 

50 if not isinstance(auth_token, str) or not auth_token.strip() or any(c.isspace() for c in auth_token): 

51 raise XrayError('Authentication returned an invalid token') 

52 except requests.exceptions.RequestException as exc: 

53 raise XrayError(f'{type(exc).__name__}: cannot authenticate with Xray') from exc 

54 except ValueError as exc: 

55 raise XrayError('Authentication returned invalid JSON') from exc 

56 r.headers['Authorization'] = f'Bearer {auth_token}' 

57 return r 

58 

59 

60class ApiKeyAuth(AuthBase): 

61 """Personal access token authentication.""" 

62 

63 def __init__(self, api_key: str) -> None: 

64 self.api_key = api_key 

65 

66 def __call__(self, r: requests.PreparedRequest) -> requests.PreparedRequest: 

67 r.headers['Authorization'] = f'Bearer {self.api_key}' 

68 return r 

69 

70 

71class XrayPublisher: 

72 """Exports Xray report to a Jira server.""" 

73 

74 def __init__(self, base_url: str, endpoint: str, auth: AuthType, verify: Union[bool, str] = True) -> None: 

75 if base_url.endswith('/'): 75 ↛ 76line 75 didn't jump to line 76 because the condition on line 75 was never true

76 base_url = base_url[:-1] 

77 self.base_url = base_url 

78 self.endpoint = endpoint 

79 self.auth = auth 

80 self.verify = verify 

81 

82 @property 

83 def endpoint_url(self) -> str: 

84 """Return full URL to the server.""" 

85 return self.base_url + self.endpoint 

86 

87 def _send_data(self, url: str, auth: AuthType, data: dict[str, Any]) -> dict[str, Any]: 

88 headers = {'Accept': 'application/json', 'Content-Type': 'application/json'} 

89 try: 

90 response = requests.request( 

91 method='POST', 

92 url=url, 

93 headers=headers, 

94 json=data, 

95 auth=auth, 

96 verify=self.verify, 

97 timeout=REQUEST_TIMEOUT, 

98 allow_redirects=False, 

99 ) 

100 response.raise_for_status() 

101 if not 200 <= response.status_code < 300: 101 ↛ 102line 101 didn't jump to line 102 because the condition on line 101 was never true

102 raise XrayError(f'Xray import returned unexpected HTTP {response.status_code}; redirects are disabled') 

103 result = response.json() 

104 except requests.exceptions.HTTPError as exc: 

105 raise XrayError(f'HTTPError: Could not post to Xray. Response status code: {response.status_code}') from exc 

106 except requests.exceptions.RequestException as exc: 

107 raise XrayError(f'{type(exc).__name__}: cannot post results to Xray') from exc 

108 except ValueError as exc: 

109 raise XrayError('Xray import returned invalid JSON') from exc 

110 if not isinstance(result, dict): 

111 raise XrayError('Xray import response must be a JSON object') 

112 return result 

113 

114 def publish(self, data: dict[str, Any]) -> str: 

115 """Publish once; never retry a potentially accepted import automatically.""" 

116 response_data = self._send_data(self.endpoint_url, self.auth, data) 

117 try: 

118 key = response_data['testExecIssue']['key'] if 'testExecIssue' in response_data else response_data['key'] 

119 except (KeyError, TypeError): 

120 raise XrayError('Cannot read Test Execution ID from server response') from None 

121 if not isinstance(key, str) or not key.strip(): 

122 raise XrayError('Cannot read Test Execution ID from server response') 

123 return key