Coverage for chalicelib/core/log_tools/log_tools.py: 66%

1import json 

2 

3from chalicelib.core.modules import TENANT_CONDITION 

4from chalicelib.utils import pg_client, helper 

5 

6EXCEPT = ["jira_server", "jira_cloud"] 

7 

8 

9def search(project_id): 

10 result = [] 

11 with pg_client.PostgresClient() as cur: 

12 cur.execute( 

13 cur.mogrify( 

14 """\ 

15 SELECT supported_integrations.name, 

16 (SELECT COUNT(*) 

17 FROM public.integrations 

18 INNER JOIN public.projects USING (project_id) 

19 WHERE provider = supported_integrations.name 

20 AND project_id = %(project_id)s 

21 AND projects.deleted_at ISNULL 

22 LIMIT 1) AS count 

23 FROM unnest(enum_range(NULL::integration_provider)) AS supported_integrations(name);""", 

24 {"project_id": project_id}) 

25 ) 

26 r = cur.fetchall() 

27 for k in r: 

28 if k["count"] > 0 and k["name"] not in EXCEPT: 

29 result.append({"value": helper.key_to_camel_case(k["name"]), "type": "logTool"}) 

30 return {"data": result} 

31 

32 

33def add(project_id, integration, options): 

34 options = json.dumps(options) 

35 with pg_client.PostgresClient() as cur: 

36 cur.execute( 

37 cur.mogrify( 

38 """\ 

39 INSERT INTO public.integrations(project_id, provider, options)  

40 VALUES (%(project_id)s, %(provider)s, %(options)s::jsonb) 

41 RETURNING *;""", 

42 {"project_id": project_id, "provider": integration, "options": options}) 

43 ) 

44 r = cur.fetchone() 

45 return helper.dict_to_camel_case(helper.flatten_nested_dicts(r)) 

46 

47 

48def get(project_id, integration): 

49 with pg_client.PostgresClient() as cur: 

50 cur.execute( 

51 cur.mogrify( 

52 """\ 

53 SELECT integrations.*  

54 FROM public.integrations INNER JOIN public.projects USING(project_id) 

55 WHERE provider = %(provider)s  

56 AND project_id = %(project_id)s 

57 AND projects.deleted_at ISNULL 

58 LIMIT 1;""", 

59 {"project_id": project_id, "provider": integration}) 

60 ) 

61 r = cur.fetchone() 

62 return helper.dict_to_camel_case(helper.flatten_nested_dicts(r)) 

63 

64 

65def edit(project_id, integration, changes): 

66 if "projectId" in changes: 66 ↛ 67line 66 didn't jump to line 67 because the condition on line 66 was never true

67 changes.pop("project_id") 

68 if "integration" in changes: 68 ↛ 69line 68 didn't jump to line 69 because the condition on line 68 was never true

69 changes.pop("integration") 

70 if len(changes.keys()) == 0: 70 ↛ 71line 70 didn't jump to line 71 because the condition on line 70 was never true

71 return None 

72 with pg_client.PostgresClient() as cur: 

73 cur.execute( 

74 cur.mogrify("""\ 

75 UPDATE public.integrations 

76 SET options=options||%(changes)s 

77 WHERE project_id =%(project_id)s AND provider = %(provider)s  

78 RETURNING *;""", 

79 {"project_id": project_id, "provider": integration, "changes": json.dumps(changes)}) 

80 ) 

81 return helper.dict_to_camel_case(helper.flatten_nested_dicts(cur.fetchone())) 

82 

83 

84def delete(project_id, integration): 

85 with pg_client.PostgresClient() as cur: 

86 cur.execute( 

87 cur.mogrify("""\ 

88 DELETE FROM public.integrations 

89 WHERE project_id=%(project_id)s AND provider=%(provider)s;""", 

90 {"project_id": project_id, "provider": integration}) 

91 ) 

92 return {"state": "success"} 

93 

94 

95def get_all_by_tenant(tenant_id, integration): 

96 with pg_client.PostgresClient() as cur: 

97 cur.execute( 

98 cur.mogrify( 

99 f"""SELECT integrations.*  

100 FROM public.integrations INNER JOIN public.projects USING(project_id)  

101 WHERE provider = %(provider)s AND {TENANT_CONDITION}  

102 AND projects.deleted_at ISNULL;""", 

103 {"tenant_id": tenant_id, "provider": integration}) 

104 ) 

105 r = cur.fetchall() 

106 return helper.list_to_camel_case(r, flatten=True)