Coverage for chalicelib/core/issue_tracking/github.py: 93%

1import schemas 

2from chalicelib.core.issue_tracking import base 

3from chalicelib.core.issue_tracking.github_issue import GithubIntegrationIssue 

4from chalicelib.utils import pg_client, helper 

5 

6PROVIDER = schemas.IntegrationType.GITHUB 

7 

8 

9class GitHubIntegration(base.BaseIntegration): 

10 

11 def __init__(self, tenant_id, user_id): 

12 self.__tenant_id = tenant_id 

13 super(GitHubIntegration, self).__init__(user_id=user_id, ISSUE_CLASS=GithubIntegrationIssue) 

14 

15 @property 

16 def provider(self): 

17 return PROVIDER 

18 

19 @property 

20 def issue_handler(self): 

21 return self._issue_handler 

22 

23 def get_obfuscated(self): 

24 integration = self.get() 

25 if integration is None: 25 ↛ 26line 25 didn't jump to line 26 because the condition on line 25 was never true

26 return None 

27 return {"token": helper.obfuscate(text=integration["token"]), "provider": self.provider.lower()} 

28 

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

30 with pg_client.PostgresClient() as cur: 

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

32 cur.execute( 

33 cur.mogrify(f"""\ 

34 UPDATE public.oauth_authentication 

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

36 WHERE user_id=%(user_id)s 

37 RETURNING token;""", 

38 {"user_id": self._user_id, 

39 **changes}) 

40 ) 

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

42 if w and w.get("token") and obfuscate: 42 ↛ 44line 42 didn't jump to line 44 because the condition on line 42 was always true

43 w["token"] = helper.obfuscate(w["token"]) 

44 return w 

45 

46 def _add(self, data): 

47 pass 

48 

49 def add(self, token, obfuscate=False): 

50 with pg_client.PostgresClient() as cur: 

51 cur.execute( 

52 cur.mogrify("""\ 

53 INSERT INTO public.oauth_authentication(user_id, provider, provider_user_id, token) 

54 VALUES(%(user_id)s, 'github', '', %(token)s) 

55 RETURNING token;""", 

56 {"user_id": self._user_id, 

57 "token": token}) 

58 ) 

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

60 if w and w.get("token") and obfuscate: 

61 w["token"] = helper.obfuscate(w["token"]) 

62 return w 

63 

64 # TODO: make a revoke token call 

65 def delete(self): 

66 with pg_client.PostgresClient() as cur: 

67 cur.execute( 

68 cur.mogrify("""\ 

69 DELETE FROM public.oauth_authentication 

70 WHERE user_id=%(user_id)s AND provider=%(provider)s;""", 

71 {"user_id": self._user_id, "provider": self.provider.lower()}) 

72 ) 

73 return {"state": "success"} 

74 

75 def add_edit(self, data: schemas.IssueTrackingGithubSchema): 

76 s = self.get() 

77 if s is not None: 

78 return self.update( 

79 changes={ 

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

81 else s.token 

82 }, 

83 obfuscate=True 

84 ) 

85 else: 

86 return self.add(token=data.token, obfuscate=True)