Coverage for chalicelib/core/issue_tracking/jira_cloud.py: 45%

1import schemas 

2from chalicelib.core.issue_tracking import base 

3from chalicelib.core.issue_tracking.jira_cloud_issue import JIRACloudIntegrationIssue 

4from chalicelib.utils import pg_client, helper 

5 

6PROVIDER = schemas.IntegrationType.JIRA 

7 

8 

9def obfuscate_string(string): 

10 return "*" * (len(string) - 4) + string[-4:] 

11 

12 

13class JIRAIntegration(base.BaseIntegration): 

14 def __init__(self, tenant_id, user_id): 

15 self.__tenant_id = tenant_id 

16 # TODO: enable super-constructor when OAuth is done 

17 # super(JIRAIntegration, self).__init__(jwt, user_id, JIRACloudIntegrationProxy) 

18 self._issue_handler = None 

19 self._user_id = user_id 

20 self.integration = self.get() 

21 

22 @staticmethod 

23 def __validate(data): 

24 data["valid"] = JIRAIntegration.__is_valid_url(data["url"]) 

25 

26 @staticmethod 

27 def __is_valid_url(url): 

28 return url.endswith('atlassian.net') or url.endswith('atlassian.net/') 

29 

30 @property 

31 def provider(self): 

32 return PROVIDER 

33 

34 @property 

35 def issue_handler(self): 

36 if JIRAIntegration.__is_valid_url(self.integration["url"]) and self._issue_handler is None: 

37 try: 

38 self._issue_handler = JIRACloudIntegrationIssue(token=self.integration["token"], 

39 username=self.integration["username"], 

40 url=self.integration["url"]) 

41 except Exception as e: 

42 self._issue_handler = None 

43 self.integration["valid"] = False 

44 return {"errors": ["Something went wrong, please check your JIRA credentials."]} 

45 return self._issue_handler 

46 

47 # TODO: remove this once jira-oauth is done 

48 def get(self): 

49 with pg_client.PostgresClient() as cur: 

50 cur.execute( 

51 cur.mogrify( 

52 """SELECT username, token, url 

53 FROM public.jira_cloud 

54 WHERE user_id = %(user_id)s;""", 

55 {"user_id": self._user_id}) 

56 ) 

57 data = helper.dict_to_camel_case(cur.fetchone()) 

58 

59 if data is None: 59 ↛ 61line 59 didn't jump to line 61 because the condition on line 59 was always true

60 return 

61 JIRAIntegration.__validate(data) 

62 return data 

63 

64 def get_obfuscated(self): 

65 if self.integration is None: 65 ↛ 67line 65 didn't jump to line 67 because the condition on line 65 was always true

66 return None 

67 integration = dict(self.integration) 

68 integration["token"] = obfuscate_string(self.integration["token"]) 

69 integration["provider"] = self.provider.lower() 

70 return integration 

71 

72 def update(self, changes, obfuscate=False): 

73 with pg_client.PostgresClient() as cur: 

74 sub_query = [f"{helper.key_to_snake_case(k)} = %({k})s" for k in changes.keys()] 

75 cur.execute( 

76 cur.mogrify(f"""\ 

77 UPDATE public.jira_cloud 

78 SET {','.join(sub_query)} 

79 WHERE user_id=%(user_id)s 

80 RETURNING username, token, url;""", 

81 {"user_id": self._user_id, 

82 **changes}) 

83 ) 

84 w = helper.dict_to_camel_case(cur.fetchone()) 

85 JIRAIntegration.__validate(w) 

86 if obfuscate: 

87 w["token"] = obfuscate_string(w["token"]) 

88 return w 

89 

90 # TODO: make this generic for all issue tracking integrations 

91 def _add(self, data): 

92 print("a pretty defined abstract method") 

93 return 

94 

95 def add(self, username, token, url, obfuscate=False): 

96 with pg_client.PostgresClient() as cur: 

97 cur.execute( 

98 cur.mogrify(""" \ 

99 INSERT INTO public.jira_cloud(username, token, user_id, url) 

100 VALUES (%(username)s, %(token)s, %(user_id)s, %(url)s) RETURNING username, token, url;""", 

101 {"user_id": self._user_id, "username": username, 

102 "token": token, "url": url}) 

103 ) 

104 w = helper.dict_to_camel_case(cur.fetchone()) 

105 JIRAIntegration.__validate(w) 

106 if obfuscate: 

107 w["token"] = obfuscate_string(w["token"]) 

108 

109 return w 

110 

111 def delete(self): 

112 with pg_client.PostgresClient() as cur: 

113 cur.execute( 

114 cur.mogrify(""" \ 

115 DELETE 

116 FROM public.jira_cloud 

117 WHERE user_id = %(user_id)s;""", 

118 {"user_id": self._user_id}) 

119 ) 

120 return {"state": "success"} 

121 

122 def add_edit(self, data: schemas.IssueTrackingJiraSchema): 

123 if self.integration is not None: 

124 return self.update( 

125 changes={ 

126 "username": data.username, 

127 "token": data.token if len(data.token) > 0 and data.token.find("***") == -1 \ 

128 else self.integration["token"], 

129 "url": str(data.url) 

130 }, 

131 obfuscate=True 

132 ) 

133 else: 

134 return self.add( 

135 username=data.username, 

136 token=data.token, 

137 url=str(data.url), 

138 obfuscate=True 

139 )