Coverage for webapp/store/logic.py: 84%

1import sys

2import datetime

3import json

4from collections import OrderedDict

5import re

6import humanize

7from dateutil import parser

8from mistune import html

9from canonicalwebteam.docstring_extractor import get_docstrings

10from webapp.helpers import (

11 discourse_api,

12 get_yaml_loader,

13 markdown_to_html,

14 get_soup,

15 modify_headers,

16)

17from webapp.observability.utils import trace_function

19yaml = get_yaml_loader()

21PLATFORMS = {

22 "ubuntu": "Ubuntu",

23 "centos": "CentOS",

24}

26ARCHITECTURES = ["amd64", "arm64", "ppc64el", "riscv64", "s390x"]

29@trace_function

30def get_summary(package):

31 if package["type"] == "bundle":

32 summary = (

33 package.get("store_front", {})

34 .get("bundle", {})

35 .get("summary", None)

36 )

37 else:

38 summary = (

39 package.get("store_front", {})

40 .get("metadata", {})

41 .get("summary", None)

42 )

43 return summary

46@trace_function

47def get_description(package, parse_to_html=False):

48 if package["type"] == "bundle":

49 description = (

50 package.get("store_front", {})

51 .get("bundle", {})

52 .get("description", None)

53 )

54 else:

55 description = (

56 package.get("store_front", {})

57 .get("metadata", {})

58 .get("description", None)

59 )

60 return markdown_to_html(description) if parse_to_html else description

63@trace_function

64def get_banner_url(media):

65 """

66 Get banner url from media object

68 :param media: the media dictionnary

69 :returns: the banner url

70 """

71 for m in media:

72 if m["type"] == "banner":

73 return m["url"]

75 return None

78@trace_function

79def get_channel_map(channel_map):

80 """

81 Reformat channel map to return a channel map

82 with unique risk

84 :param channel_map: the channel map from the api

85 :returns: the channel map reformatted

86 """

87 new_map = []

88 for channel in channel_map:

89 for res in new_map:

90 if channel["channel"]["name"] == res["channel"]["name"]:

91 break

92 else:

93 new_map.append(channel)

95 return new_map

98@trace_function

99def convert_channel_maps(channel_map):

100 """

101 Converts channel maps list to format easier to manipulate

102

103 :param channel_maps: The channel maps list returned by the API

104

105 :returns: The channel maps reshaped

106 """

107 result = {}

108 track_order = {"latest": 1}

109 risk_order = {"stable": 1, "candidate": 2, "beta": 3, "edge": 4}

110 for channel in channel_map:

111 track = channel["channel"].get("track", "latest")

112 risk = channel["channel"]["risk"]

113 revision_number = channel["revision"]["revision"]

114

115 if track not in result:

116 result[track] = {}

117

118 if risk not in result[track]:

119 result[track][risk] = {"latest": None, "releases": {}}

120

121 # same revision but for a different arch

122 if revision_number in result[track][risk]["releases"]:

123 arch = channel["channel"]["base"]["architecture"]

124

125 if arch == "all":

126 result[track][risk]["releases"][revision_number][

127 "architectures"

128 ].update(ARCHITECTURES)

129 else:

130 result[track][risk]["releases"][revision_number][

131 "architectures"

132 ].add(arch)

133 continue

134

135 info = {

136 "released_at": channel["channel"]["released-at"],

137 "release_date": convert_date(channel["channel"]["released-at"]),

138 "version": channel["revision"]["version"],

139 "channel": channel["channel"]["name"],

140 "risk": channel["channel"]["risk"],

141 "size": channel["revision"]["download"]["size"],

142 "bases": extract_series(channel, True),

143 "channel_bases": extract_bases(channel),

144 "revision": process_revision(channel["revision"]),

145 "architectures": set(),

146 }

147

148 if channel["channel"]["base"]:

149 arch = channel["channel"]["base"]["architecture"]

150 if arch == "all":

151 info["architectures"].update(ARCHITECTURES)

152 else:

153 info["architectures"].add(arch)

154

155 result[track][risk]["releases"][revision_number] = info

156

157 # Order tracks (latest track first)

158 result = OrderedDict(

159 sorted(

160 result.items(), key=lambda x: track_order.get(x[0], sys.maxsize)

161 )

162 )

163

164 # Order risks (stable, candidate, beta, edge)

165 for track, track_data in result.items():

166 result[track] = OrderedDict(

167 sorted(

168 track_data.items(),

169 key=lambda x: risk_order.get(x[0], sys.maxsize),

170 )

171 )

172

173 # Order releases by revision

174 for risk, data in result[track].items():

175 result[track][risk]["releases"] = OrderedDict(

176 sorted(

177 result[track][risk]["releases"].items(),

178 key=lambda release: release[1]["released_at"],

179 reverse=True,

180 )

181 )

182

183 # Collect all the bases available across all releases

184

185 base_names = sorted(

186 list(

187 set(

188 base

189 for release in result[track][risk]["releases"].values()

190 for base in release["bases"]

191 )

192 ),

193 reverse=True,

194 )

195

196 result[track][risk]["all_bases"] = [

197 {

198 "name": base,

199 "architectures": sorted(

200 list(

201 set(

202 arch

203 for release in result[track][risk][

204 "releases"

205 ].values()

206 if base in release["bases"]

207 for arch in release["architectures"]

208 )

209 )

210 ),

211 }

212 for base in base_names

213 ]

214

215 latest_revision_key = max(

216 result[track][risk]["releases"].keys(),

217 key=lambda k: result[track][risk]["releases"][k][

218 "released_at"

219 ],

220 )

221 result[track][risk]["latest"] = result[track][risk]["releases"][

222 latest_revision_key

223 ]

224 return result

225

226

227@trace_function

228def process_revision(revision):

229 bases = []

230

231 for base in revision["bases"]:

232 if base and base.get("architecture") == "all":

233 for arch in ARCHITECTURES:

234 bases.append({**base, "architecture": arch})

235 else:

236 bases.append(base)

237 return {**revision, "bases": bases}

238

239

240@trace_function

241def extract_resources(channel):

242 """

243 Extract resources from channel map

244

245 :param channel_maps: The channel maps list returned by the API

246

247 :returns: Charm resource names

248 """

249 resources = []

250

251 channel_resources = channel["resources"]

252

253 for resource in channel_resources:

254 resources.append(

255 {"name": resource["name"], "revision": resource["revision"]}

256 )

257

258 return resources

259

260

261@trace_function

262def extract_default_release_architectures(channel):

263 architectures = set()

264

265 for base in channel["revision"]["bases"]:

266 if not base or base["architecture"] in architectures:

267 continue

268

269 arch = base["architecture"]

270 if arch == "all":

271 architectures.update(ARCHITECTURES)

272 else:

273 architectures.add(arch)

274

275 return sorted(architectures)

276

277

278@trace_function

279def extract_all_arch(channel_map, parent_dict):

280 all_archy = set()

281 all_channel_bases = {}

282 platforms = {}

283

284 for version_data in channel_map.values():

285 channel_map_all = list(version_data.items())

286 for _, channel_data in channel_map_all:

287 for release in channel_data["releases"].values():

288 all_archy = all_archy.union(release["architectures"])

289

290 for base in release["channel_bases"]:

291 for series in base["channels"]:

292 platform = PLATFORMS.get(base["name"], base["name"])

293

294 if base["name"] not in platforms:

295 platforms[base["name"]] = set()

296 platforms[base["name"]].add(series)

297

298 all_channel_bases[base["name"] + series] = (

299 f"{platform} {series}"

300 )

301

302 parent_dict["all_architectures"] = sorted(all_archy)

303 parent_dict["all_platforms"] = platforms

304 parent_dict["all_channel_bases"] = dict(

305 sorted(all_channel_bases.items(), reverse=True)

306 )

307

308 return

309

310

311@trace_function

312def extract_series(channel, long_name=False):

313 """

314 Extract ubuntu series from channel map

315

316 :param channel_maps: The channel maps list returned by the API

317

318 :returns: Ubuntu series number

319 """

320 series = set()

321

322 for base in channel["revision"]["bases"]:

323 if not base or base["channel"] in series:

324 continue

325 platform = PLATFORMS.get(base["name"], base["name"])

326 series.add(

327 f"{platform} {base['channel']}" if long_name else base["channel"]

328 )

329

330 return sorted(series, reverse=True)

331

332

333@trace_function

334def extract_bases(channel):

335 bases = channel["revision"]["bases"]

336 channel_bases = []

337

338 for i in bases:

339 if i is None:

340 return []

341

342 has_base = False

343

344 for b in channel_bases:

345 if b["name"] == i["name"]:

346 has_base = True

347

348 if not has_base:

349 channel_bases.append(

350 {

351 "name": i["name"],

352 "channels": set(),

353 }

354 )

355

356 for i in channel_bases:

357 for b in bases:

358 if b["name"] == i["name"]:

359 i["channels"].add(b["channel"])

360

361 i["channels"] = sorted(i["channels"], reverse=True)

362

363 return channel_bases

364

365

366@trace_function

367def convert_date(date_to_convert):

368 """

369 Convert date to human readable format: Month Day Year

370

371 If date is less than a day return: today or yesterday

372

373 Format of date to convert: 2019-01-12T16:48:41.821037+00:00

374 Output: Jan 12 2019

375

376 :param date_to_convert: Date to convert

377 :returns: Readable date

378 """

379 date_parsed = parser.parse(date_to_convert).replace(tzinfo=None)

380 delta = datetime.datetime.now() - datetime.timedelta(days=1)

381 if delta < date_parsed:

382 return humanize.naturalday(date_parsed).title()

383 else:

384 return date_parsed.strftime("%d %b %Y")

385

386

387@trace_function

388def get_icons(package):

389 media = package["result"]["media"]

390 return [m["url"] for m in media if m["type"] == "icon"]

391

392

393@trace_function

394def get_docs_topic_id(metadata_yaml):

395 """

396 Return discourse topic ID or None

397 """

398 base_url = discourse_api.base_url

399 docs_link = metadata_yaml.get("docs")

400

401 if docs_link:

402 if docs_link.startswith(base_url):

403 docs_link_parts = docs_link[len(base_url) :].split("/")

404

405 if len(docs_link_parts) > 2:

406 topic_id = docs_link_parts[-1]

407

408 if topic_id.isnumeric():

409 return topic_id

410

411 return None

412

413

414@trace_function

415def convert_categories(api_categories):

416 """

417 The name property in the API response has a slug

418 like format, e.g., big-data

419

420 This method will return the desired name and an

421 extra slug property with the value from the API

422 """

423 result = []

424

425 for category in api_categories:

426 category["slug"] = category["name"]

427 category["name"] = format_slug(category["slug"])

428 result.append(category)

429

430 return result

431

432

433@trace_function

434def add_store_front_data(package, details=False):

435 extra = {}

436

437 extra["icons"] = get_icons(package)

438

439 if package["result"]["deployable-on"]:

440 extra["deployable-on"] = package["result"]["deployable-on"]

441 else:

442 extra["deployable-on"] = ["vm"]

443

444 extra["categories"] = convert_categories(package["result"]["categories"])

445

446 if "title" in package["result"] and package["result"]["title"]:

447 extra["display-name"] = package["result"]["title"]

448 else:

449 extra["display-name"] = format_slug(package["name"])

450

451 if details:

452 extra["metadata"] = yaml.load(

453 package["default-release"]["revision"]["metadata-yaml"]

454 )

455 extra["config"] = yaml.load(

456 package["default-release"]["revision"]["config-yaml"]

457 )

458 extra["actions"] = yaml.load(

459 package["default-release"]["revision"]["actions-yaml"]

460 )

461

462 if package["type"] == "bundle":

463 extra["bundle"] = yaml.load(

464 package["default-release"]["revision"]["bundle-yaml"]

465 )

466

467 # Get bundle docs

468 extra["docs_topic"] = get_docs_topic_id(extra["bundle"])

469

470 # List charms

471 extra["bundle"]["charms"] = get_bundle_charms(

472 extra["bundle"].get(

473 "applications", extra["bundle"].get("services")

474 )

475 )

476 else:

477 # Get charm docs

478 extra["docs_topic"] = get_docs_topic_id(extra["metadata"])

479

480 # Reshape channel maps

481 extra["channel_map"] = convert_channel_maps(package["channel-map"])

482 extra["resources"] = extract_resources(package["default-release"])

483

484 # Extract all supported series

485 extra["architectures"] = extract_default_release_architectures(

486 package["default-release"]

487 )

488 # extract all architecture based on series

489 extract_all_arch(extra["channel_map"], extra)

490 extra["series"] = extract_series(package["default-release"])

491 extra["channel_bases"] = extract_bases(package["default-release"])

492

493 # Some needed fields

494 extra["publisher_name"] = package["result"]["publisher"][

495 "display-name"

496 ]

497 extra["username"] = package["result"]["publisher"]["username"]

498

499 if "summary" in package["result"]:

500 extra["summary"] = package["result"]["summary"]

501

502 # Handle issues and website keys

503 if "issues" in extra["metadata"]:

504 if not isinstance(extra["metadata"]["issues"], list):

505 extra["metadata"]["issues"] = [extra["metadata"]["issues"]]

506

507 if "website" in extra["metadata"]:

508 if not isinstance(extra["metadata"]["website"], list):

509 extra["metadata"]["website"] = [extra["metadata"]["website"]]

510

511 package["store_front"] = extra

512 return package

513

514

515@trace_function

516def get_bundle_charms(charm_apps):

517 result = []

518

519 if charm_apps:

520 for _, data in charm_apps.items():

521 # Charm names could be with the old prefix/suffix

522 # Like: cs:~charmed-osm/mariadb-k8s-35

523 name = data["charm"]

524 if name.startswith("cs:") or name.startswith("ch:"):

525 name = re.match(r"(?:cs:|ch:)(?:.+/)?(\S*?)(?:-\d+)?$", name)[

526 1

527 ]

528

529 charm = {"title": format_slug(name), "name": name}

530

531 result.append(charm)

532

533 return result

534

535

536@trace_function

537def process_python_docs(library, module_name):

538 """Process libraries response from the API

539 to generate the HTML output"""

540

541 # Obtain Python docstrings

542 docstrings = get_docstrings(library["content"], module_name)

543

544 bs_soup = get_soup(html(docstrings["docstring_text"]))

545 docstrings["html"] = modify_headers(bs_soup, 3)

546

547 return docstrings

548

549

550@trace_function

551def process_libraries(libraries):

552 """Process the libraries response from the API"""

553

554 result = []

555

556 for lib in libraries["libraries"]:

557 data = {

558 "id": lib["library-id"],

559 "name": lib["library-name"],

560 "hash": lib["hash"],

561 "created_at": lib["created-at"],

562 }

563

564 result.append(data)

565

566 return result

567

568

569@trace_function

570def get_library(library_name, libraries):

571 library = next(

572 (lib for lib in libraries if lib.get("name") == library_name),

573 None,

574 )

575

576 if not library:

577 return None

578

579 return library["id"]

580

581

582@trace_function

583def format_slug(slug):

584 """Format slug name into a standard title format

585 :param slug: The hypen spaced, lowercase slug to be formatted

586 :return: The formatted string

587 """

588

589 return (

590 slug.title()

591 .replace("-", " ")

592 .replace("_", " ")

593 .replace("And", "and")

594 .replace("Iot", "IoT")

595 )

596

597

598with open("webapp/store/overlay.json") as overlay_file:

599 overlay = json.load(overlay_file)

600

601

602@trace_function

603def add_overlay_data(package):

604 """

605 Adds custom hard-coded overlay.json data to the package object

606 :param package: The package object retrieved from the snapcraft API

607 :return: The package object with an additional "overlay" key

608 containing extra info

609 """

610

611 if overlay.get(package["name"]) is not None:

612 package["overlay_data"] = overlay[package["name"]].copy()

613

614 return package

615

616

617@trace_function

618def get_doc_link(package):

619 """

620 Returns the documentation link of a package

621 """

622 docs = package.get("store_front", {}).get("metadata", {}).get("docs", None)

623 return docs

624

625

626def get_revisions(channel_maps: list) -> list:

627 """Gets a sorted list of unique revisions

628

629 :param channel_maps: Channel map list

630

631 :returns: A sorted list of unique revisions

632 """

633 revisions = {

634 channel_map["revision"]["revision"] for channel_map in channel_maps

635 }

636 return list(reversed(sorted(revisions)))