Coverage for webapp/publisher/snaps/metrics_views.py: 81%

1# Standard library 

2from json import loads 

3from dateutil import relativedelta 

4import math 

5 

6# Packages 

7import flask 

8import webapp.metrics.helper as metrics_helper 

9import webapp.metrics.metrics as metrics 

10from canonicalwebteam.store_api.dashboard import Dashboard 

11from canonicalwebteam.store_api.devicegw import DeviceGW 

12 

13# Local 

14from webapp.helpers import api_session 

15from webapp.decorators import login_required 

16from webapp.publisher.snaps import logic 

17 

18 

19dashboard = Dashboard(api_session) 

20device_gateway = DeviceGW("snap", api_session) 

21 

22downsample_data_limit = 500 

23downsample_target_size = 10 

24 

25 

26@login_required 

27def get_account_snaps_metrics(): 

28 if not flask.request.data: 

29 error = {"error": "Please provide a list of snaps"} 

30 return flask.jsonify(error), 500 

31 

32 try: 

33 metrics = {"buckets": [], "snaps": []} 

34 

35 snaps = loads(flask.request.data) 

36 metrics_query = metrics_helper.build_snap_installs_metrics_query(snaps) 

37 

38 if metrics_query: 

39 snap_metrics = dashboard.get_publisher_metrics( 

40 flask.session, json=metrics_query 

41 ) 

42 metrics = metrics_helper.transform_metrics( 

43 metrics, snap_metrics, snaps 

44 ) 

45 return flask.jsonify(metrics), 200 

46 except Exception: 

47 error = {"error": "An error occured while fetching metrics"} 

48 return flask.jsonify(error), 500 

49 

50 

51@login_required 

52def get_measure_snap(snap_name): 

53 return flask.redirect( 

54 flask.url_for(".publisher_snap_metrics", snap_name=snap_name) 

55 ) 

56 

57 

58@login_required 

59def publisher_snap_metrics(snap_name): 

60 """ 

61 A view to display the snap metrics page for specific snaps. 

62 """ 

63 # If this fails, the page will 404 

64 dashboard.get_snap_info(flask.session, snap_name) 

65 

66 context = { 

67 # Data direct from details API 

68 "snap_name": snap_name, 

69 # pass snap id from here? 

70 "is_linux": "Linux" in flask.request.headers["User-Agent"], 

71 } 

72 

73 return flask.render_template("store/publisher.html", **context) 

74 

75 

76@login_required 

77def get_active_devices(snap_name): 

78 snap_details = dashboard.get_snap_info(flask.session, snap_name) 

79 

80 snap_id = snap_details["snap_id"] 

81 

82 installed_base_metric = logic.verify_base_metrics( 

83 flask.request.args.get("active-devices", default="version", type=str) 

84 ) 

85 

86 period = flask.request.args.get("period", default="30d", type=str) 

87 active_device_period = logic.extract_metrics_period(period) 

88 

89 page = flask.request.args.get("page", default=1, type=int) 

90 

91 metric_requested_length = active_device_period["int"] 

92 metric_requested_bucket = active_device_period["bucket"] 

93 

94 page_time_length = flask.request.args.get( 

95 "page-length", default=3, type=int 

96 ) 

97 total_page_num = 1 

98 if metric_requested_bucket == "d" or ( 

99 metric_requested_bucket == "m" 

100 and page_time_length >= metric_requested_length 

101 ): 

102 dates = metrics_helper.get_dates_for_metric( 

103 metric_requested_length, metric_requested_bucket 

104 ) 

105 start = dates["start"] 

106 end = dates["end"] 

107 else: 

108 page_period_length = ( 

109 (metric_requested_length * 12) 

110 if metric_requested_bucket == "y" 

111 else metric_requested_length 

112 ) 

113 total_page_num = math.floor(page_period_length / page_time_length) 

114 

115 end = metrics_helper.get_last_metrics_processed_date() + ( 

116 relativedelta.relativedelta( 

117 months=-(page_time_length * (page - 1)) 

118 ) 

119 ) 

120 start = end + (relativedelta.relativedelta(months=-(page_time_length))) 

121 

122 # Decrease the date by a day to make sure 

123 # there is no overlapping dates across the pages. 

124 if page != 1: 

125 end = end + relativedelta.relativedelta(days=-1) 

126 

127 installed_base = logic.get_installed_based_metric(installed_base_metric) 

128 

129 new_metrics_query = metrics_helper.build_active_device_metric_query( 

130 snap_id=snap_id, installed_base=installed_base, end=end, start=start 

131 ) 

132 

133 metrics_response = dashboard.get_publisher_metrics( 

134 flask.session, json=new_metrics_query 

135 ) 

136 

137 active_metrics = metrics_helper.find_metric( 

138 metrics_response["metrics"], installed_base 

139 ) 

140 

141 metrics_data = active_metrics 

142 buckets = metrics_data["buckets"] 

143 series = metrics_data["series"] 

144 metric_name = metrics_data["metric_name"] 

145 # Add constants to a variable 

146 if len(series) > downsample_data_limit: 

147 ( 

148 downsampled_buckets, 

149 downsampled_series, 

150 ) = metrics_helper.downsample_series( 

151 buckets, series, downsample_target_size 

152 ) 

153 else: 

154 downsampled_buckets = buckets 

155 downsampled_series = series 

156 

157 series = downsampled_series 

158 if metric_name == "weekly_installed_base_by_channel": 

159 for s in series: 

160 if "/" not in s["name"]: 

161 s["name"] = f"latest/{s['name']}" 

162 

163 if installed_base_metric == "os": 

164 for item in series: 

165 item["name"] = metrics._capitalize_os_name(item["name"]) 

166 

167 active_devices = metrics.ActiveDevices( 

168 name=metric_name, 

169 series=series, 

170 buckets=downsampled_buckets, 

171 status=metrics_data["status"], 

172 ) 

173 

174 latest_active = 0 

175 if active_devices: 

176 latest_active = active_devices.get_number_latest_active_devices() 

177 

178 return flask.jsonify( 

179 { 

180 "active_devices": dict(active_devices), 

181 "latest_active_devices": latest_active, 

182 "total_page_num": total_page_num, 

183 } 

184 ) 

185 

186 

187@login_required 

188def get_latest_active_devices(snap_name): 

189 snap_details = dashboard.get_snap_info(flask.session, snap_name) 

190 snap_id = snap_details["snap_id"] 

191 # get latest active devices 

192 latest_day_period = logic.extract_metrics_period("1d") 

193 latest_installed_base = logic.get_installed_based_metric("version") 

194 latest_day_query_json = metrics_helper.build_metric_query_installed_base( 

195 snap_id=snap_id, 

196 installed_base=latest_installed_base, 

197 metric_period=latest_day_period["int"], 

198 metric_bucket=latest_day_period["bucket"], 

199 ) 

200 

201 latest_day_response = dashboard.get_publisher_metrics( 

202 flask.session, json=latest_day_query_json 

203 ) 

204 

205 latest_active = 0 

206 

207 if latest_day_response: 

208 latest_active_metrics = metrics_helper.find_metric( 

209 latest_day_response["metrics"], latest_installed_base 

210 ) 

211 if latest_active_metrics: 

212 latest_active_devices = metrics.ActiveDevices( 

213 name=latest_active_metrics["metric_name"], 

214 series=latest_active_metrics["series"], 

215 buckets=latest_active_metrics["buckets"], 

216 status=latest_active_metrics["status"], 

217 ) 

218 latest_active = ( 

219 latest_active_devices.get_number_latest_active_devices() 

220 ) 

221 return flask.jsonify( 

222 { 

223 "latest_active_devices": latest_active, 

224 } 

225 ) 

226 

227 

228@login_required 

229def get_metric_annotaion(snap_name): 

230 details = dashboard.get_snap_info(flask.session, snap_name) 

231 annotations = {"name": "annotations", "series": [], "buckets": []} 

232 

233 for category in details["categories"]["items"]: 

234 date = category["since"].split("T")[0] 

235 new_date = logic.convert_date(category["since"]) 

236 

237 if date not in annotations["buckets"]: 

238 annotations["buckets"].append(date) 

239 

240 index_of_date = annotations["buckets"].index(date) 

241 

242 single_series = { 

243 "values": [0] * (len(annotations)), 

244 "name": category["name"], 

245 "display_name": category["name"].capitalize().replace("-", " "), 

246 "display_date": new_date, 

247 "date": date, 

248 } 

249 

250 single_series["values"][index_of_date] = 1 

251 

252 annotations["series"].append(single_series) 

253 

254 annotations["series"] = sorted( 

255 annotations["series"], key=lambda k: k["date"] 

256 ) 

257 return flask.jsonify(annotations) 

258 

259 

260@login_required 

261def get_country_metric(snap_name): 

262 snap_details = dashboard.get_snap_info(flask.session, snap_name) 

263 snap_id = snap_details["snap_id"] 

264 metrics_query_json = metrics_helper.build_metric_query_country( 

265 snap_id=snap_id, 

266 ) 

267 

268 metrics_response = dashboard.get_publisher_metrics( 

269 flask.session, json=metrics_query_json 

270 ) 

271 

272 country_metric = metrics_helper.find_metric( 

273 metrics_response["metrics"], "weekly_installed_base_by_country" 

274 ) 

275 country_devices = metrics.CountryDevices( 

276 name=country_metric["metric_name"], 

277 series=country_metric["series"], 

278 buckets=country_metric["buckets"], 

279 status=country_metric["status"], 

280 private=True, 

281 ) 

282 

283 territories_total = 0 

284 if country_devices: 

285 territories_total = country_devices.get_number_territories() 

286 

287 return flask.jsonify( 

288 { 

289 "active_devices": country_devices.country_data, 

290 "territories_total": territories_total, 

291 } 

292 )