[roll] Roll Global Integration
GitOrigin-RevId: 44dc9ff4dd9fe02155eb1471c4c2feff57537344
Change-Id: Ie7adfd6b465d3aad947e3291d4b270109d96c505
diff --git a/ctf/ctf_generate_manifest.py b/ctf/ctf_generate_manifest.py
deleted file mode 100755
index 9040943..0000000
--- a/ctf/ctf_generate_manifest.py
+++ /dev/null
@@ -1,303 +0,0 @@
-#!/usr/bin/env python3
-# Copyright 2024 The Fuchsia Authors. All rights reserved.
-# Use of this source code is governed by a BSD-style license that can be
-# found in the LICENSE file.
-
-import argparse
-import os
-import sys
-import json
-import re
-from collections import abc
-import subprocess
-import tempfile
-import difflib
-
-SCRIPT_DIRECTORY = os.path.dirname(os.path.abspath(__file__))
-FUCHSIA_CTS_PACKAGE = "fuchsia/cts/linux-amd64"
-
-
-def main(args: argparse.Namespace) -> int:
-
- f_release_regex = re.compile(r"f[0-9]+")
-
- try:
- level_mapping = get_git_revisions_from_cipd(sorted(get_supported_api_levels()))
- except Exception as e:
- print_error(f"Could not load all data: {e}")
- return 1
-
- existing_files = [
- f
- for f in os.listdir(SCRIPT_DIRECTORY)
- if os.path.isfile(os.path.join(SCRIPT_DIRECTORY, f))
- and f_release_regex.match(f)
- ]
-
- deleted_files: set[str] = set()
- added_files: set[str] = set()
- populated_files: set[str] = set()
- skipped_files: set[str] = set()
-
- expected_files: set[str] = set(level_mapping.keys())
- for file in expected_files:
- if file not in existing_files:
- added_files.add(file)
- elif os.stat(os.path.join(SCRIPT_DIRECTORY, file)).st_size == 0:
- populated_files.add(file)
- else:
- skipped_files.add(file)
- for file in existing_files:
- if file not in expected_files:
- deleted_files.add(file)
-
- import_file_path = os.path.join(SCRIPT_DIRECTORY, "all")
- new_import_file_contents = format_import_file(sorted(level_mapping.keys()))
- old_import_file_contents = (
- open(import_file_path).read() if os.path.isfile(import_file_path) else None
- )
-
- if not args.dry_run:
- for file in deleted_files:
- os.remove(os.path.join(SCRIPT_DIRECTORY, file))
- for file in added_files.union(populated_files):
- with open(os.path.join(SCRIPT_DIRECTORY, file), "w") as f:
- f.write(format_ctf_template(file, level_mapping[file]))
-
- if new_import_file_contents != old_import_file_contents:
- if not args.dry_run:
- with open(import_file_path, "w") as f:
- f.write(new_import_file_contents)
-
- print("Import file diff:\n=====")
- print(
- "\n".join(
- difflib.Differ().compare(
- (old_import_file_contents or "").split("\n"),
- new_import_file_contents.split("\n"),
- )
- )
- )
- print("=====")
-
- if old_import_file_contents is None:
- added_files.add("all")
- else:
- populated_files.add("all")
- else:
- skipped_files.add("all")
-
- print_outcome(
- args.dry_run,
- deleted=sorted(deleted_files),
- added=sorted(added_files),
- populated=sorted(populated_files),
- skipped=sorted(skipped_files),
- )
-
- return 0
-
-
-def get_supported_api_levels() -> set[str]:
- """Return a set of supported API levels, as listed in version_history.json
-
- Returns:
- set[str]: Versions that are supported. For example, {"f15", "f16"}.
- """
- version_path = os.path.join(
- SCRIPT_DIRECTORY,
- "..",
- "..",
- "infra",
- "config",
- "common",
- "version_history.json",
- )
- with open(version_path) as f:
- version_contents = json.load(f)
-
- # Get the name of each API level where its status is "supported"
- version_data: dict[str, dict[str, str]] = version_contents["data"]["api_levels"]
- return {
- f"f{key}"
- for key, value in version_data.items()
- if value["status"] == "supported"
- }
-
-
-def get_git_revisions_from_cipd(versions: abc.Iterable[str]) -> dict[str, str]:
- """Annotate versions with their corresponding git_revision tag from CIPD.
-
- This method uses CIPD to find the associated git_revision for a CIPD
- package by its version name. If the package doesn't exist or does not have
- a git_revision, that version is omitted from the output dictionary.
-
- Args:
- keys (abc.Iterable[str]): CIPD package versions to search for.
-
- Returns:
- dict[str, str]: Mapping from version to its git_revision,
- only if it exists.
- """
- ret: dict[str, str] = dict()
-
- for key in versions:
- with tempfile.TemporaryDirectory() as td:
- # Run CIPD command to search for a matching version of the
- # Fuchsia CTS package.
- output_path = os.path.join(td, "output.json")
- args = [
- "cipd",
- "describe",
- FUCHSIA_CTS_PACKAGE,
- "-version",
- key,
- "-json-output",
- output_path,
- ]
- print("Running ", args)
- result = subprocess.run(args)
-
- # If we found one, look through the tags for one containing
- # the git revision.
- if result.returncode == 0:
- with open(output_path) as f:
- data = json.load(f)
- for tag_object in data["result"]["tags"]:
- tag: str = tag_object["tag"]
- if tag.startswith("git_revision:"):
- ret[key] = tag.split(":")[1]
- break
-
- return ret
-
-
-_CTF_TEMPLATE: str = r"""<?xml version="1.0" encoding="UTF-8"?>
-<manifest>
- <packages>
- <!-- Versioned CTF release. See ctf_generate_manifest.py for details. -->
- <package name="fuchsia/cts/${platform}"
- path="prebuilt/ctf/{F_RELEASE}/{{.OS}}-{{.Arch}}"
- platforms="linux-amd64"
- version="git_revision:{REVISION}" />
- </packages>
-</manifest>"""
-
-
-def format_ctf_template(f_release: str, git_revision: str) -> str:
- """Format a manifest file for the given f_release version and git revision.
-
- Args:
- f_release (str): Release name. For example, "f15".
- revision (str): Git revision tag from CIPD for that release.
-
- Returns:
- str: Formatted jiri manifest file, to be written to disk.
- """
- return _CTF_TEMPLATE.replace(r"{F_RELEASE}", f_release).replace(
- r"{REVISION}", git_revision
- )
-
-
-_IMPORTER_TEMPLATE: str = r"""<?xml version="1.0" encoding="UTF-8"?>
-<manifest>
- <imports>
- {IMPORTS}
- </imports>
-</manifest>
-"""
-
-
-def format_import_file(files: list[str]) -> str:
- """Format a complete manifest import file.
-
- Args:
- files (list[str]): Files to import in the manifest.
-
- Returns:
- str: Formatted manifest file, ready to write to disk.
- """
- return _IMPORTER_TEMPLATE.replace(r"{IMPORTS}", format_import_list(files))
-
-
-_IMPORT_TEMPLATE: str = r'<localimport file="{FILE}"/>'
-
-
-def format_import_list(files: list[str]) -> str:
- """Format the localimport section for a manifest.
-
- Args:
- files (list[str]): Files to include in the import list.
-
- Returns:
- _type_: Formatted string containing imports to include in a manifest.
- """
- return "\n ".join([_IMPORT_TEMPLATE.replace(r"{FILE}", name) for name in files])
-
-
-def print_error(reason: str) -> None:
- """Print a formatted error as JSON.
-
- Args:
- reason (str): Reason string to include in the error.
- """
- print(
- json.dumps(
- {
- "status": "ERROR",
- "reason": reason,
- }
- ),
- )
-
-
-def print_outcome(
- dry_run: bool,
- deleted: list[str],
- added: list[str],
- populated: list[str],
- skipped: list[str],
-):
- """Print the outcome of the operation as a JSON object.
-
- Args:
- dry_run (bool): True if this tool was executed with --dry-run.
- deleted (list[str]): List of files deleted.
- added (list[str]): List of files added.
- populated (list[str]): List of files populated (existed but have new data).
- skipped (list[str]): List of unchanged files.
- """
- did_work = len(deleted + added + populated) != 0
- status: str
- if dry_run:
- status = "DRY_RUN"
- elif did_work:
- status = "OK"
- else:
- status = "SKIPPED"
-
- print(
- json.dumps(
- {
- "status": status,
- "deleted": deleted,
- "added": added,
- "populated": populated,
- "skipped": skipped,
- }
- ),
- )
-
-
-if __name__ == "__main__":
- args = argparse.ArgumentParser(
- "ctf_generate_manifest", "Generate configs for CTF manifests"
- )
- args.add_argument(
- "--dry-run",
- action=argparse.BooleanOptionalAction,
- default=True,
- help="If set, do not actually mutate files. Default True.",
- )
-
- sys.exit(main(args.parse_args()))
