Coverage for src/usage.py: 100%

1"""To record usage of the app.""" 

2 

3import json 

4import os 

5from datetime import datetime, timedelta 

6 

7import requests 

8 

9from src.config_parser import prepare_root_destination 

10 

11CACHE_FILE_NAME = "/config/.data" 

12NEW_INSTALLATION_ENDPOINT = os.environ.get("NEW_INSTALLATION_ENDPOINT", None) 

13NEW_HEARTBEAT_ENDPOINT = os.environ.get("NEW_HEARTBEAT_ENDPOINT", None) 

14APP_NAME = "icloud-drive-docker" 

15APP_VERSION = os.environ.get("APP_VERSION", "dev") 

16NEW_INSTALLATION_DATA = {"appName": APP_NAME, "appVersion": APP_VERSION} 

17 

18 

19def init_cache(config): 

20 """Initialize the cache file.""" 

21 root_destination_path = prepare_root_destination(config=config) 

22 cache_file_path = os.path.join(root_destination_path, CACHE_FILE_NAME) 

23 return cache_file_path 

24 

25 

26def load_cache(file_path: str): 

27 """Load the cache file.""" 

28 data = {} 

29 if os.path.isfile(file_path): 

30 with open(file_path, encoding="utf-8") as f: 

31 data = json.load(f) 

32 else: 

33 save_cache(file_path=file_path, data={}) 

34 return data 

35 

36 

37def save_cache(file_path: str, data: object): 

38 """Save data to the cache file.""" 

39 with open(file_path, "w", encoding="utf-8") as f: 

40 json.dump(data, f) 

41 return True 

42 

43 

44def post_new_installation(data, endpoint=NEW_INSTALLATION_ENDPOINT): 

45 """Post new installation to server.""" 

46 try: 

47 response = requests.post(endpoint, data, timeout=10000) 

48 if response.ok: 

49 data = response.json() 

50 return data["id"] 

51 except Exception: 

52 pass 

53 return None 

54 

55 

56def record_new_installation(previous_id=None): 

57 """Record new or upgrade existing installation.""" 

58 data = dict(NEW_INSTALLATION_DATA) 

59 if previous_id: 

60 data["previousId"] = previous_id 

61 return post_new_installation(data) 

62 

63 

64def already_installed(cached_data): 

65 """Check if already installed.""" 

66 return "id" in cached_data and "app_version" in cached_data and cached_data["app_version"] == APP_VERSION 

67 

68 

69def install(cached_data): 

70 """Install the app.""" 

71 new_id = record_new_installation(cached_data.get("id", None)) 

72 if new_id: 

73 cached_data["id"] = new_id 

74 cached_data["app_version"] = APP_VERSION 

75 return cached_data 

76 return False 

77 

78 

79def post_new_heartbeat(data, endpoint=NEW_HEARTBEAT_ENDPOINT): 

80 """Post the heartbeat to server.""" 

81 try: 

82 response = requests.post(endpoint, data, timeout=10000) 

83 if response.ok: 

84 return True 

85 except Exception: 

86 pass 

87 return False 

88 

89 

90def send_heartbeat(app_id, data=None): 

91 """Prepare and send heartbeat to server.""" 

92 data = {"installationId": app_id, "data": data} 

93 return post_new_heartbeat(data) 

94 

95 

96def current_time(): 

97 """Get current time.""" 

98 return datetime.now() 

99 

100 

101def heartbeat(cached_data, data): 

102 """Send heartbeat.""" 

103 previous_heartbeat = cached_data.get("heartbeat_timestamp", None) 

104 current = current_time() 

105 if previous_heartbeat: 

106 previous = datetime.strptime(previous_heartbeat, "%Y-%m-%d %H:%M:%S.%f") 

107 if previous < (current - timedelta(hours=24)): 

108 if send_heartbeat(cached_data.get("id"), data=data): 

109 cached_data["heartbeat_timestamp"] = str(current) 

110 return cached_data 

111 else: 

112 return False 

113 elif send_heartbeat(cached_data.get("id"), data=data): 

114 cached_data["heartbeat_timestamp"] = str(current) 

115 return cached_data 

116 

117 

118def alive(config, data=None): 

119 """Record liveliness.""" 

120 cache_file_path = init_cache(config=config) 

121 cached_data = load_cache(cache_file_path) 

122 if not already_installed(cached_data=cached_data): 

123 installed_data = install(cached_data=cached_data) 

124 if installed_data: 

125 return save_cache(file_path=cache_file_path, data=installed_data) 

126 heartbeat_data = heartbeat(cached_data=cached_data, data=data) 

127 if heartbeat_data: 

128 return save_cache(file_path=cache_file_path, data=heartbeat_data) 

129 return False