plugin-template

#!/usr/bin/env python3
# /// script
# requires-python = ">=3.11"
# dependencies = [
#     "black==24.3.0",
#     "jamldump>=1.2.0,<1.3.0",
#     "jinja2>=3.1.6,<3.2.0",
#     "pyyaml>=6.0.3,<6.1.0",
#     "requests~=2.32.3",
#     "requests-cache>=1.3.0,<1.4.0",
#     "tomlkit>=0.14.0,<0.14.1",
# ]
# ///

import typing as t
import argparse
import os
import shlex
import shutil
import subprocess
import sys
from pathlib import Path

import jamldump
import yaml
from jinja2 import Environment, FileSystemLoader

import utils

DEFAULT_SETTINGS = {
    "api_root": "/pulp/",
    "black": True,
    "check_commit_message": True,
    "check_gettext": True,
    "check_manifest": True,
    "check_stray_pulpcore_imports": True,
    "cli_package": "pulp-cli",
    "cli_repo": "https://github.com/pulp/pulp-cli.git",
    "ci_base_image": "ghcr.io/pulp/pulp-ci-centos9",
    "ci_env": {},
    "ci_trigger": "{pull_request: {branches: ['*']}}",
    "core_import_allowed": [],
    "deploy_client_to_pypi": True,
    "deploy_client_to_rubygems": True,
    "deploy_to_pypi": True,
    "disabled_redis_runners": [],
    "docker_fixtures": False,
    "extra_files": [],
    "flake8_ignore": [],
    "flake8": True,
    "github_org": "pulp",
    "latest_release_branch": None,
    "lint_requirements": True,
    "os_required_packages": [],
    "parallel_test_workers": 8,
    "plugin_app_label": None,
    "plugin_default_branch": "main",
    "plugin_name": None,
    "plugins": None,
    "pulp_env": {},
    "pulp_env_azure": {},
    "pulp_env_gcp": {},
    "pulp_env_s3": {},
    "pulp_scheme": "https",
    "pulp_settings_azure": {
        "domain_enabled": True,
        "MEDIA_ROOT": "",
        "STORAGES": {
            "default": {
                "BACKEND": "storages.backends.azure_storage.AzureStorage",
                "OPTIONS": {
                    "account_key": (
                        "Eby8vdM02xNOcqFlqUwJPLlmEtlCDXJ1OUzFT50uSRZ6IFsuFq2UVErCz4I6tq/"
                        "K1SZFPTOtr/KBHBeksoGMGw=="
                    ),
                    "account_name": "devstoreaccount1",
                    "location": "pulp3",
                    "azure_container": "pulp-test",
                    "overwrite_files": True,
                    "expiration_secs": 120,
                    "connection_string": (
                        "DefaultEndpointsProtocol=http;AccountName=devstoreaccount1;AccountKey"
                        "=Eby8vdM02xNOcqFlqUwJPLlmEtlCDXJ1OUzFT50uSRZ6IFsuFq2UVErCz4I6tq/"
                        "K1SZFPTOtr/KBHBeksoGMGw==;BlobEndpoint=http://ci-azurite:10000/"
                        "devstoreaccount1;"
                    ),
                },
            },
            "staticfiles": {
                "BACKEND": "django.contrib.staticfiles.storage.StaticFilesStorage",
            },
        },
    },
    "pulp_settings_gcp": {
        "domain_enabled": True,
        "MEDIA_ROOT": "",
        "STORAGES": {
            "default": {
                "BACKEND": "storages.backends.gcloud.GoogleCloudStorage",
                "OPTIONS": {
                    "bucket_name": "gcppulp",
                    "custom_endpoint": "http://ci-gcp:4443",
                },
            },
            "staticfiles": {
                "BACKEND": "django.contrib.staticfiles.storage.StaticFilesStorage",
            },
        },
    },
    "pulp_settings": None,
    "pulp_settings_s3": {
        "domain_enabled": True,
        "MEDIA_ROOT": "",
        "STORAGES": {
            "default": {
                "BACKEND": "storages.backends.s3boto3.S3Boto3Storage",
                "OPTIONS": {
                    "access_key": "AKIAIT2Z5TDYPX3ARJBA",
                    "secret_key": "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",
                    "region_name": "eu-central-1",
                    "addressing_style": "path",
                    "signature_version": "s3v4",
                    "bucket_name": "pulp3",
                    "endpoint_url": "http://minio:9000",
                    "default_acl": "@none",
                },
            },
            "staticfiles": {
                "BACKEND": "django.contrib.staticfiles.storage.StaticFilesStorage",
            },
        },
    },
    "pydocstyle": True,
    "release_email": "pulp-infra@redhat.com",
    "release_user": "pulpbot",
    "stalebot_days_until_close": 30,
    "stalebot_days_until_stale": 90,
    "stalebot_limit_to_pulls": True,
    "stalebot": True,
    "supported_release_branches": [],
    "sync_ci": True,
    "test_azure": False,
    "test_cli": False,
    "test_deprecations": True,
    "test_gcp": False,
    "test_lowerbounds": True,
    "test_performance": False,
    "test_s3": False,
    "use_issue_template": True,
}


DEPRECATED_FILES = {
    "github": [
        ".bumpversion.cfg",
        ".ci/assets/bindings/.gitkeep",
        ".ci/scripts/changelog.py",
        ".ci/scripts/calc_deps_lowerbounds.py",
        ".ci/scripts/cherrypick.sh",
        ".ci/scripts/docs-builder.py",
        ".ci/scripts/publish_docs.sh",
        ".ci/scripts/publish_client_gem.sh",
        ".ci/scripts/publish_client_pypi.sh",
        ".ci/scripts/publish_plugin_pypi.sh",
        ".ci/scripts/redmine.py",
        ".ci/scripts/release_requirements.txt",
        ".ci/scripts/release.py",
        ".ci/scripts/tweet.py",
        ".ci/scripts/update_ci_branches.py",
        ".ci/scripts/update_github.sh",
        ".ci/scripts/update_redmine.sh",
        ".ci/scripts/upper_bound.py",
        ".github/template_gitref",
        ".github/workflows/kanban.yml",
        ".github/workflows/changelog.yml",
        ".github/workflows/fips.yml",
        ".github/workflows/scripts/create_release_from_tag.sh",
        ".github/workflows/scripts/install_python_client.sh",
        ".github/workflows/scripts/install_ruby_client.sh",
        ".github/workflows/scripts/release.py",
        ".github/workflows/scripts/stage-changelog-for-master.py",
        ".github/workflows/scripts/post_docs_test.sh",
        ".github/workflows/scripts/update_ci.sh",
        ".github/workflows/scripts/docs-publisher.py",
        ".github/workflows/scripts/publish_docs.sh",
        ".github/workflows/scripts/publish_client_gem.sh",
        ".github/workflows/scripts/publish_client_pypi.sh",
        ".github/workflows/scripts/publish_plugin_pypi.sh",
        ".travis",
        ".travis.yml",
        "flake8.cfg",
        "CHANGES/.TEMPLATE.rst",
    ],
}


def main() -> int:
    parser = argparse.ArgumentParser(
        formatter_class=argparse.RawTextHelpFormatter,
        description="Create or update a plugin using the current template.",
    )
    parser.add_argument(
        "plugin_name",
        type=str,
        nargs="?",
        help="Create or update this plugin. The name should start with pulp- or pulp_.",
    )
    parser.add_argument(
        "--plugin-app-label",
        type=str,
        help="The Django app label for the plugin - defaults to the part after pulp_ or pulp-.",
    )
    parser.add_argument(
        "--generate-config",
        action="store_true",
        help="Create or update a plugin template config for a plugin and exit.",
    )
    parser.add_argument(
        "--bootstrap",
        action="store_true",
        help="Create a new plugin and template boilerplate code.",
    )
    parser.add_argument(
        "--test",
        action="store_true",
        help="Generate or update functional and unit tests.",
    )
    parser.add_argument(
        "--ci",
        action="store_true",
        help="Generate or update generic CI/CD files.",
    )
    parser.add_argument(
        "--github",
        action="store_true",
        help="Generate or update github CI/CD configuration files.",
    )
    parser.add_argument(
        "--all",
        action="store_true",
        help="Create a new plugin and template all non-excluded files.",
    )
    parser.add_argument(
        "--verbose",
        action="store_true",
        help="Include more output.",
    )
    parser.add_argument(
        "--latest-release-branch",
        metavar="VERSION_BRANCH",
        action="store",
        help="Mark specified version as the latest_release_branch before templating.",
    )
    parser.add_argument(
        "--plugin-root-dir",
        action="store",
        help="Root directory of the plugin if it's not '../{plugin_name}'.",
    )
    args = parser.parse_args()

    here = Path(__file__).parent.resolve()
    if args.plugin_root_dir is not None:
        plugin_root_dir = Path(args.plugin_root_dir).resolve()
    elif args.plugin_name is None:
        # Assume this is called from an existing plugin directory.
        plugin_root_dir = Path(".").resolve()
    else:
        plugin_root_dir = (here.parent / args.plugin_name).resolve()
    plugin_config_path = plugin_root_dir / "template_config.yml"
    write_new_config = False

    if plugin_config_path.exists():
        config, write_new_config = load_config(plugin_config_path, args.verbose)
        if args.plugin_name is not None and args.plugin_name != config["plugin_name"]:
            raise RuntimeError("Aborting: 'plugin_name' in the config does not match.")
        if (
            args.plugin_app_label is not None
            and args.plugin_app_label != config["plugin_app_label"]
        ):
            raise RuntimeError("Aborting: 'plugin_app_label' in the config does not match.")

    else:
        print(f"Could not find a plugin template config at {plugin_config_path}.")
        if args.all or args.generate_config:
            print(f"Creating a new plugin at {plugin_root_dir}")
            config = generate_new_config(args.plugin_name, args.plugin_app_label)
        else:
            return 2

    if args.latest_release_branch:
        config["latest_release_branch"] = str(args.latest_release_branch)
        write_new_config = True

    sections = [
        section
        for section in ["generate_config", "bootstrap", "ci", "github", "test"]
        if getattr(args, section) or args.all or (section == "generate_config" and write_new_config)
    ]
    for section in sections:
        write_template_section(config, section, plugin_root_dir, verbose=args.verbose)

    # This is a dummy call left here as an example for future migrations to come.
    migrate_dummy(plugin_root_dir, config, sections)

    remove_deprecated_files(plugin_root_dir, sections)

    reformat_files(plugin_root_dir, config)

    return 0


def load_config(plugin_config_path: Path, verbose: bool) -> tuple[dict[str, t.Any], bool]:
    write_new_config = False
    with open(plugin_config_path) as config_file:
        config = yaml.safe_load(config_file)
    # Add any missing value from the list of defaults
    for key, value in DEFAULT_SETTINGS.items():
        if key not in config:
            config[key] = value
            write_new_config = True
    # remove deprecated options
    for key in set(config.keys()) - set(DEFAULT_SETTINGS.keys()):
        config.pop(key)
        write_new_config = True
    # TODO: validate config
    # validate versions are strings
    assert all((isinstance(version, str) for version in config["supported_release_branches"]))
    print(f"Loaded plugin template config from {plugin_config_path}.")

    if verbose:
        print("Using the following config:")
        print(jamldump.to_jaml(config))

    return config, write_new_config


def migrate_dummy(plugin_root_dir: Path, config: dict[str, t.Any], sections: list[str]) -> None:
    pass


def write_template_section(
    config: dict[str, t.Any], section: str, plugin_root_dir: Path, verbose: bool = False
) -> None:
    """
    Template or copy all files for the section.
    """
    templates_dir = Path(__file__).parent.resolve() / "templates"
    section_templates_dir = templates_dir / section
    env = Environment(
        loader=FileSystemLoader(
            [
                section_templates_dir,  # Where all templates are actually templated.
                templates_dir / "include",  # Included to find the macros.
            ]
        ),
        extensions=["jamldump.jinja2.Jaml"],
        line_statement_prefix="%%",
        line_comment_prefix="%#",
    )
    env.filters["camel"] = utils.to_camel
    env.filters["caps"] = utils.to_caps
    env.filters["dash"] = utils.to_dash
    env.filters["snake"] = utils.to_snake
    env.filters["to_yaml"] = utils.to_nice_yaml
    env.filters["from_yaml"] = utils.from_yaml
    env.filters["shquote"] = shlex.quote

    files_templated = 0
    files_copied = 0

    PULPDOCS_BRANCH = "main"
    template_vars = {
        "section": section,
        "setup_py": (plugin_root_dir / "setup.py").exists(),
        "ci_update_branches": utils.ci_update_branches(config),
        "python_version": "3.11",
        "ci_update_hour": sum((ord(c) for c in config["plugin_app_label"])) % 24,
        "current_version": utils.current_version(plugin_root_dir),
        "pulpdocs_branch": PULPDOCS_BRANCH,
        "is_pulpdocs_member": config["plugin_name"] in utils.get_pulpdocs_members(PULPDOCS_BRANCH),
        "black_version": utils.black_version(),
        "config": config,
        **config,
    }

    relative_path_set = generate_relative_path_set(section_templates_dir)
    for relative_path in relative_path_set:
        if not config["stalebot"] and "stale" in relative_path:
            continue
        if config["use_issue_template"] is False and "ISSUE_TEMPLATE" in relative_path:
            continue
        destination_relative_path = relative_path.replace(
            "plugin_name", utils.to_snake(config["plugin_name"])
        )
        necessary_dir_structure = os.path.dirname(
            os.path.join(plugin_root_dir, destination_relative_path)
        )

        if not os.path.exists(necessary_dir_structure):
            os.makedirs(necessary_dir_structure)

        if relative_path.endswith(".j2"):
            template = env.get_template(relative_path)
            destination = destination_relative_path[: -len(".j2")]
            if destination.startswith("pyproject.toml."):
                utils.merge_toml(
                    template,
                    plugin_root_dir,
                    destination,
                    template_vars,
                )
            else:
                utils.template_to_file(
                    template,
                    plugin_root_dir,
                    destination,
                    template_vars,
                )
            files_templated += 1
            if verbose:
                print(f"Templated file: {relative_path}")
        else:
            if destination_relative_path.endswith(".copy"):
                destination_relative_path = destination_relative_path[: -len(".copy")]
            shutil.copy(
                section_templates_dir / relative_path,
                plugin_root_dir / destination_relative_path,
            )
            files_copied += 1
            if verbose:
                print(f"Copied file: {relative_path}")

    print(f"Section: {section} \n  templated: {files_templated}\n  copied: {files_copied}")


def generate_relative_path_set(root_dir: Path) -> set[str]:
    """
    Create a set of relative paths within the specified directory.
    """
    applicable_paths = set()
    for root, dirs, files in os.walk(root_dir, topdown=False):
        for file_name in files:
            template_abs_path = os.path.join(root, file_name)
            template_relative_path = os.path.relpath(template_abs_path, root_dir)
            applicable_paths.add(template_relative_path)
    return applicable_paths


def generate_new_config(plugin_name: str, plugin_app_label: str | None) -> dict[str, t.Any]:
    """
    Generates a default config for a plugin

    Args:
        plugin_name (str): the full name of the plugin repository name (e.g. pulp_file)
        plugin_app_label (str): the Django app label for the plugin - usually the part after
            pulp_ or pulp-

    Returns:
        config (dict): a dictionary containing a default template configuration for the plugin
    """
    if plugin_app_label is None:
        plugin_app_label = plugin_name.removeprefix("pulp").replace("-", "_").strip("_")
    config = DEFAULT_SETTINGS.copy()
    config["plugin_name"] = plugin_name
    config["plugin_app_label"] = plugin_app_label
    config["plugins"] = [{"name": plugin_name, "app_label": plugin_app_label}]
    return config


def remove_deprecated_files(plugin_root_dir: Path, sections: list[str]) -> None:
    """Check for files that have been deprecated (ie moved or removed)."""
    print("Deprecation check:")
    files_found = False
    for section in sections:
        for fp in DEPRECATED_FILES.get(section, []):
            path = Path(plugin_root_dir) / fp
            if path.exists():
                if path.is_dir():
                    shutil.rmtree(path)
                    print(f"  Removed deprecated directory: '{path}'.")
                else:
                    path.unlink()
                    print(f"  Removed deprecated file: '{path}'.")
                files_found = True

    if not files_found:
        print("  No deprecated files found.")


def reformat_files(plugin_root_dir: Path, config: dict[str, t.Any]) -> None:
    if config["black"]:
        try:
            subprocess.run(["black", "--quiet", "."], cwd=plugin_root_dir)
        except FileNotFoundError:
            pass


if __name__ == "__main__":
    sys.exit(main())