Coverage for webapp/helpers.py: 90%

101 statements  

« prev     ^ index     » next       coverage.py v7.10.2, created at 2025-08-05 22:06 +0000

1import json 

2import os 

3import hashlib 

4 

5import flask 

6from canonicalwebteam.launchpad import Launchpad 

7from ruamel.yaml import YAML 

8from webapp.api.requests import PublisherSession, Session 

9from canonicalwebteam.store_api.dashboard import Dashboard 

10import webapp.api.marketo as marketo_api 

11 

12_yaml = YAML(typ="rt") 

13_yaml_safe = YAML(typ="safe") 

14api_session = Session() 

15api_publisher_session = PublisherSession() 

16marketo = marketo_api.Marketo() 

17dashboard = Dashboard(api_session) 

18 

19launchpad = Launchpad( 

20 username=os.getenv("LP_API_USERNAME"), 

21 token=os.getenv("LP_API_TOKEN"), 

22 secret=os.getenv("LP_API_TOKEN_SECRET"), 

23 session=api_publisher_session, 

24) 

25 

26 

27def get_yaml_loader(typ="safe"): 

28 if typ == "safe": 

29 return _yaml_safe 

30 return _yaml 

31 

32 

33def get_licenses(): 

34 try: 

35 with open("webapp/licenses.json") as f: 

36 licenses = json.load(f)["licenses"] 

37 

38 def _build_custom_license(license_id, license_name): 

39 return {"licenseId": license_id, "name": license_name} 

40 

41 CUSTOM_LICENSES = [ 

42 _build_custom_license("Proprietary", "Proprietary"), 

43 _build_custom_license("Other Open Source", "Other Open Source"), 

44 _build_custom_license( 

45 "AGPL-3.0+", "GNU Affero General Public License v3.0 or later" 

46 ), 

47 ] 

48 

49 licenses = licenses + CUSTOM_LICENSES 

50 except Exception: 

51 licenses = [] 

52 

53 return licenses 

54 

55 

56def is_valid_path(path): 

57 base_path = os.path.abspath(flask.current_app.root_path) 

58 target_path = os.path.abspath(os.path.join(base_path, path)) 

59 return target_path.startswith(base_path) 

60 

61 

62def get_file(filename, replaces={}): 

63 """ 

64 Reads a file, replaces occurences of all the keys in `replaces` with 

65 the correspondant values and returns the resulting string or None 

66 

67 Keyword arguments: 

68 filename -- name if the file to load. 

69 replaces -- key/values to replace in the file content (default {}) 

70 """ 

71 if not is_valid_path(filename): 

72 return None 

73 

74 filepath = os.path.join(flask.current_app.root_path, filename) 

75 

76 try: 

77 with open(filepath, "r") as f: 

78 data = f.read() 

79 for key in replaces: 

80 data = data.replace(key, replaces[key]) 

81 except Exception: 

82 data = None 

83 

84 return data 

85 

86 

87def get_yaml(filename, typ="safe", replaces={}): 

88 """ 

89 Reads a file, replaces occurences of all the keys in `replaces` with the 

90 correspondant values and returns an ordered dict with the YAML content 

91 

92 Keyword arguments: 

93 filename -- name if the file to load. 

94 typ -- type of yaml loader 

95 replaces -- key/values to replace in the file content (default {}) 

96 """ 

97 try: 

98 yaml = get_yaml_loader(typ) 

99 data = get_file(filename, replaces) 

100 return yaml.load(data) 

101 except Exception: 

102 return None 

103 

104 

105def dump_yaml(data, stream, typ="safe"): 

106 yaml = get_yaml_loader(typ) 

107 yaml.dump(data, stream) 

108 

109 

110def get_icon(media): 

111 icons = [m["url"] for m in media if m["type"] == "icon"] 

112 if len(icons) > 0: 

113 return icons[0] 

114 return "" 

115 

116 

117def get_publisher_data(): 

118 # We don't use the data from this endpoint. 

119 # It is mostly used to make sure the user has signed 

120 # the terms and conditions. 

121 dashboard.get_account(flask.session) 

122 

123 flask_user = flask.session["publisher"] 

124 

125 subscriptions = None 

126 

127 # don't rely on marketo to show the page, 

128 # if anything fails, just continue and don't show 

129 # this section 

130 try: 

131 subscribed_to_newsletter = False 

132 marketo_user = marketo.get_user(flask_user["email"]) 

133 if marketo_user: 

134 marketo_subscribed = marketo.get_newsletter_subscription( 

135 marketo_user["id"] 

136 ) 

137 if marketo_subscribed.get("snapcraftnewsletter"): 

138 subscribed_to_newsletter = True 

139 

140 subscriptions = {"newsletter": subscribed_to_newsletter} 

141 except Exception: 

142 if "sentry" in flask.current_app.extensions: 

143 flask.current_app.extensions["sentry"].captureException() 

144 

145 flask_user["subscriptions"] = subscriptions 

146 context = {"publisher": flask_user} 

147 

148 return context 

149 

150 

151def get_dns_verification_token(snap_name, domain): 

152 salt = os.getenv("DNS_VERIFICATION_SALT") 

153 token_string = f"{domain}:{snap_name}:{salt}" 

154 token = hashlib.sha256(token_string.encode("utf-8")).hexdigest() 

155 return token 

156 

157 

158def get_csp_as_str(csp={}): 

159 csp_str = "" 

160 for key, values in csp.items(): 

161 csp_value = " ".join(values) 

162 csp_str += f"{key} {csp_value}; " 

163 return csp_str.strip() 

164 

165 

166def list_folders(directory): 

167 return [ 

168 item 

169 for item in os.listdir(directory) 

170 if os.path.isdir(os.path.join(directory, item)) 

171 ] 

172 

173 

174def directory_exists(file): 

175 if not is_valid_path(file): 

176 return False 

177 

178 target_path = os.path.abspath( 

179 os.path.join(flask.current_app.root_path, file) 

180 ) 

181 return os.path.isdir(target_path) 

182 

183 

184def get_brand_id(session, store_id): 

185 store = dashboard.get_store(session, store_id) 

186 return store["brand-id"]