Coverage for chalicelib/core/user_testing.py: 0%

1from chalicelib.utils import pg_client, helper 

2from chalicelib.utils.storage import StorageClient 

3from decouple import config 

4 

5 

6def get_test_signals(session_id, project_id): 

7 with pg_client.PostgresClient() as cur: 

8 cur.execute(cur.mogrify("""\ 

9 SELECT * 

10 FROM public.ut_tests_signals 

11 LEFT JOIN public.ut_tests_tasks USING (task_id) 

12 WHERE session_id = %(session_id)s 

13 ORDER BY timestamp;""", 

14 {"project_id": project_id, "session_id": session_id}) 

15 ) 

16 rows = cur.fetchall() 

17 return helper.dict_to_camel_case(rows) 

18 

19 

20def has_test_signals(session_id, project_id): 

21 with pg_client.PostgresClient() as cur: 

22 cur.execute(cur.mogrify("""\ 

23 SELECT EXISTS(SELECT 1 FROM public.ut_tests_signals 

24 WHERE session_id = %(session_id)s) AS has;""", 

25 {"project_id": project_id, "session_id": session_id}) 

26 ) 

27 row = cur.fetchone() 

28 return row.get("has") 

29 

30 

31def get_ux_webcam_signed_url(session_id, project_id, check_existence: bool = True): 

32 results = [] 

33 bucket_name = "uxtesting-records" # config("sessions_bucket") 

34 k = f'{session_id}/ux_webcam_record.webm' 

35 if check_existence and not StorageClient.exists(bucket=bucket_name, key=k): 

36 return [] 

37 results.append(StorageClient.get_presigned_url_for_sharing( 

38 bucket=bucket_name, 

39 expires_in=100000, 

40 key=k 

41 )) 

42 return results