18528321607

Committed 15 Oct 2025 12:08PM UTC coverage: 90.962% (+0.1%) from 90.837%

Build # 18528321607

Build Type

Pull #485

github

Committed by

web-flow

Commit Message

Merge 60c38ce6d into c2f4d73d9

Pull Request Pull Request #485: main: Support committing directly to the base branch

Run Details

24 of 25 new or added lines in 1 file covered. (96.0%)

1 existing line in 1 file now uncovered.

2194 of 2412 relevant lines covered (90.96%)

0.91 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

65.15

/src/main.py

#!/usr/bin/env python3
#
# flatpak-extra-data-checker: A tool for checking the status of
# the extra data in a Flatpak manifest.
#
# Copyright (C) 2018 Endless Mobile, Inc.
#
# Authors:
#       Joaquim Rocha <jrocha@endlessm.com>
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation; either version 2 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License along
# with this program; if not, write to the Free Software Foundation, Inc.,
# 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.

import argparse
import contextlib
import getpass
import json
import logging
import os
from pathlib import Path
import subprocess
import sys
import asyncio
from enum import IntFlag
import typing as t

from github import Github

from .lib.utils import parse_github_url, init_logging
from .lib.externaldata import ExternalData
from . import manifest


log = logging.getLogger(__name__)


@contextlib.contextmanager
def indir(path: Path):
    """
    >>> with indir(path):
    ...    # code executes with 'path' as working directory
    ... # old working directory is restored
    """

    old = os.getcwd()
    os.chdir(path)
    try:
        yield
    finally:
        os.chdir(old)


def print_outdated_external_data(manifest_checker: manifest.ManifestChecker):
    ext_data = manifest_checker.get_outdated_external_data()
    for data in ext_data:
        state_txt = data.state.name or str(data.state)
        message_tmpl = ""
        message_args = {}
        if data.new_version:
            if data.type == data.Type.GIT:
                assert data.new_version
                message_tmpl = (
                    "{data_state}: {data_name}\n"
                    " Has a new version:\n"
                    "  URL:       {url}\n"
                    "  Commit:    {commit}\n"
                    "  Tag:       {tag}\n"
                    "  Branch:    {branch}\n"
                    "  Version:   {version}\n"
                    "  Timestamp: {timestamp}\n"
                )
                message_args = data.new_version._asdict()
            else:
                assert isinstance(data, ExternalData)
                assert data.new_version
                message_tmpl = (
                    "{data_state}: {data_name}\n"
                    " Has a new version:\n"
                    "  URL:       {url}\n"
                    "  MD5:       {md5}\n"
                    "  SHA1:      {sha1}\n"
                    "  SHA256:    {sha256}\n"
                    "  SHA512:    {sha512}\n"
                    "  Size:      {size}\n"
                    "  Version:   {version}\n"
                    "  Timestamp: {timestamp}\n"
                )
                message_args = {
                    **data.new_version._asdict(),
                    **data.new_version.checksum._asdict(),
                }
        elif data.State.BROKEN in data.state:
            message_tmpl = (
                # fmt: off
                "{data_state}: {data_name}\n"
                " Couldn't get new version for {url}\n"
                # fmt: on
            )
            message_args = data.current_version._asdict()

        message = message_tmpl.format(
            data_state=state_txt,
            data_name=data.filename,
            **message_args,
        )
        print(message, flush=True)
    return len(ext_data)


def print_errors(manifest_checker: manifest.ManifestChecker) -> int:
    # TODO: Actually do pretty-print collected errors
    errors = manifest_checker.get_errors()
    return len(errors)


def check_call(args):
    log.debug("$ %s", " ".join(args))
    subprocess.check_call(args)


def get_manifest_git_checkout(manifest: t.Union[Path, str]) -> Path:
    # Can't use git rev-parse --show-toplevel because of a chicken-and-egg problem: we
    # need to find the checkout directory so that we can mark it as safe so that we can
    # use git against it.
    for directory in Path(manifest).parents:
        if os.path.exists(directory / ".git"):
            return directory

    raise FileNotFoundError(f"Cannot find git checkout for {manifest}")


def ensure_git_safe_directory(checkout: Path):
    uid = os.getuid()
    checkout_uid = os.stat(checkout).st_uid
    if uid == checkout_uid:
        return

    try:
        result = subprocess.run(
            ["git", "config", "--get-all", "safe.directory"],
            check=True,
            capture_output=True,
            encoding="utf-8",
        )
        safe_dirs = [Path(x) for x in result.stdout.splitlines()]
    except subprocess.CalledProcessError as err:
        # git config --get-all will return 1 if the key doesn't exist.
        # Re-raise the error for anything else.
        if err.returncode != 1:
            raise
        safe_dirs = []

    if checkout in safe_dirs:
        return

    log.info("Adding %s git safe directory", checkout)
    location = "--system" if uid == 0 else "--global"
    check_call(["git", "config", location, "--add", "safe.directory", str(checkout)])


class CommittedChanges(t.NamedTuple):
    subject: str
    body: t.Optional[str]
    commit: t.Optional[str]
    branch: str
    base_branch: t.Optional[str]


def commit_message(changes: t.List[str]) -> str:
    assert len(changes) >= 1

    if len(changes) == 1:
        return changes[0]

    module_names = list(dict.fromkeys(list(i.split(":", 1)[0] for i in changes)))

    if len(module_names) == 1:
        return f"Update {module_names[0]} module"

    for i in reversed(range(2, len(module_names) + 1)):
        xs = module_names[: i - 1]
        y = module_names[i - 1]
        zs = module_names[i:]

        if zs:
            tail = f" and {len(zs)} more modules"
            xs.append(y)
        else:
            tail = f" and {y} modules"

        subject = "Update " + ", ".join(xs) + tail
        if len(subject) <= 70:
            return subject

    return f"Update {len(module_names)} modules"


def branch_exists(branch: str) -> bool:
    try:
        subprocess.run(
            ["git", "rev-parse", "--verify", branch],
            capture_output=True,
            check=True,
        )
        return True
    except subprocess.CalledProcessError:
        return False


def commit_changes(
    changes: t.List[str], commit_to_base: bool = False
) -> t.Optional[CommittedChanges]:
    log.info("Committing updates")
    body: t.Optional[str]
    subject = commit_message(changes)
    if len(changes) > 1:
        body = "\n".join(changes)
        message = subject + "\n\n" + body
    else:
        body = None
        message = subject

    # Remember the base branch
    base_branch: t.Optional[str]
    base_branch = subprocess.check_output(
        ["git", "branch", "--show-current"], text=True
    ).strip()
    if not base_branch:
        base_branch = None

    if commit_to_base:
        if not base_branch:
            log.error("Failed to determine the base branch: cannot commit to it")
            return None
        else:
            branch = base_branch
            tree = None
    else:
        # Moved to detached HEAD
        log.info("Switching to detached HEAD")
        check_call(["git", "-c", "advice.detachedHead=false", "checkout", "HEAD@{0}"])
        # Find a stable identifier for the contents of the tree, to avoid
        # sending the same PR twice.
        tree = subprocess.check_output(
            ["git", "rev-parse", "HEAD^{tree}"], text=True
        ).strip()
        branch = (
            f"update-{base_branch}-{tree[:7]}" if base_branch else f"update-{tree[:7]}"
        )

    retry_commit = False
    try:
        check_call(["git", "commit", "-am", message])
    except subprocess.CalledProcessError:
        retry_commit = True
        pass

    if retry_commit:
        log.warning("Committing failed. Falling back to a sanitised config")
        git_name = getpass.getuser()
        git_email = git_name + "@" + "localhost"
        assert git_email is not None
        env = {
            "GIT_CONFIG_GLOBAL": "/dev/null",
            "GIT_CONFIG_NOSYSTEM": "1",
            "GIT_CONFIG": "''",
        }
        subprocess.run(
            [
                "git",
                "-c",
                f"user.name={git_name}",
                "-c",
                f"user.email={git_email}",
                "commit",
                "--no-verify",
                "--no-gpg-sign",
                "-am",
                message,
            ],
            check=True,
            env=env,
        )

    if not (commit_to_base or branch_exists(branch)):
        check_call(["git", "checkout", "-b", branch])
    return CommittedChanges(
        subject=subject,
        body=body,
        commit=tree,
        branch=branch,
        base_branch=base_branch,
    )


DISCLAIMER = (
    "🤖 This pull request was automatically generated by "
    "[flathub-infra/flatpak-external-data-checker]"
    "(https://github.com/flathub-infra/flatpak-external-data-checker). "
    "Please [open an issue]"
    "(https://github.com/flathub-infra/flatpak-external-data-checker/issues/new) "
    "if you have any questions or complaints. 🤖"
)

AUTOMERGE_DUE_TO_CONFIG = (
    "🤖 This PR passed CI, and `automerge-flathubbot-prs` is `true` in "
    "`flathub.json`, so I'm merging it automatically. 🤖"
)

AUTOMERGE_DUE_TO_BROKEN_URLS = (
    "🤖 The currently-published version contains broken URLs, and this PR passed "
    "CI, so I'm merging it automatically. You can disable this behaviour by setting "
    "`automerge-flathubbot-prs` to `false` in flathub.json. 🤖"
)


def open_pr(
    change: CommittedChanges,
    manifest_checker: t.Optional[manifest.ManifestChecker] = None,
    fork: t.Optional[bool] = None,
    pr_labels: t.Optional[t.List[str]] = None,
):

    try:
        github_token = os.environ["GITHUB_TOKEN"]
    except KeyError:
        log.error("GITHUB_TOKEN environment variable is not set")
        sys.exit(1)

    log.info("Opening pull request for branch %s", change.branch)
    g = Github(github_token)
    user = g.get_user()

    origin_url = (
        subprocess.check_output(["git", "remote", "get-url", "origin"])
        .decode("utf-8")
        .strip()
    )
    origin_repo = g.get_repo(parse_github_url(origin_url))

    if fork is True:
        log.debug("creating fork (as requested)")
        repo = user.create_fork(origin_repo)
    elif fork is False:
        log.debug("not creating fork (as requested)")
        repo = origin_repo
    elif origin_repo.permissions.push:
        log.debug("origin repo is writable; not creating fork")
        repo = origin_repo
    else:
        log.debug("origin repo not writable; creating fork")
        repo = user.create_fork(origin_repo)

    remote_url = f"https://{github_token}:x-oauth-basic@github.com/{repo.full_name}"

    base = change.base_branch or origin_repo.default_branch

    head = "{}:{}".format(repo.owner.login, change.branch)
    pr_message = ((change.body or "") + "\n\n" + DISCLAIMER).strip()

    try:
        with open("flathub.json") as f:
            repocfg = json.load(f)
    except FileNotFoundError:
        repocfg = {}

    automerge = repocfg.get("automerge-flathubbot-prs")
    # Implicitly automerge if…
    force_automerge = (
        # …the user has not explicitly disabled automerge…
        automerge is not False
        # …and we have a manifest checker (i.e. we're not in a test)…
        and manifest_checker
        # …and at least one source is broken and has an update
        and any(
            data.type == data.Type.EXTRA_DATA
            and data.State.BROKEN in data.state
            and data.new_version
            for data in manifest_checker.get_outdated_external_data()
        )
    )

    prs = origin_repo.get_pulls(state="all", base=base, head=head)

    # If the maintainer has closed our last PR or it was merged,
    # we don't want to open another one.
    closed_prs = [pr for pr in prs if pr.state == "closed"]
    for pr in closed_prs:
        log.info(
            "Found existing %s PR: %s",
            "merged" if pr.is_merged() else pr.state,
            pr.html_url,
        )
        return

    open_prs = [pr for pr in prs if pr.state == "open"]
    for pr in open_prs:
        log.info("Found open PR: %s", pr.html_url)

        if automerge or force_automerge:
            pr_commit = pr.head.repo.get_commit(pr.head.sha)
            if pr_commit.get_combined_status().state == "success" and pr.mergeable:
                log.info("PR passed CI and is mergeable, merging %s", pr.html_url)
                if automerge:
                    pr.create_issue_comment(AUTOMERGE_DUE_TO_CONFIG)
                else:  # force_automerge
                    pr.create_issue_comment(AUTOMERGE_DUE_TO_BROKEN_URLS)
                pr.merge(merge_method="rebase")
                origin_repo.get_git_ref(f"heads/{pr.head.ref}").delete()

        return

    check_call(["git", "push", "-u", remote_url, change.branch])

    log.info(
        "Creating pull request in %s from head `%s` to base `%s`",
        origin_repo.html_url,
        head,
        base,
    )

    gh_run_id = os.environ.get("GITHUB_RUN_ID")
    gh_repo_name = os.environ.get("GITHUB_REPOSITORY")
    if gh_run_id and gh_repo_name:
        log.info("Appending GitHub actions log URL to PR message")
        log_url = f"https://github.com/{gh_repo_name}/actions/runs/{gh_run_id}"
        pr_message += f"\n\n[📋 View External data checker logs]({log_url})"

    pr = origin_repo.create_pull(
        change.subject,
        pr_message,
        base,
        head,
        maintainer_can_modify=True,
    )
    log.info("Opened pull request %s", pr.html_url)
    if pr_labels:
        log.info("Adding labels to PR: %s", ", ".join(pr_labels))
        pr.set_labels(*pr_labels)


def parse_cli_args(cli_args=None):
    parser = argparse.ArgumentParser()
    parser.add_argument(
        "manifest", help="Flatpak manifest to check", type=os.path.abspath
    )
    parser.add_argument(
        "-v", "--verbose", help="Print debug messages", action="store_true"
    )
    parser.add_argument(
        "--update",
        help="Update manifest(s) to refer to new versions of "
        "external data - also open PRs for changes unless "
        "--commit-only is specified",
        action="store_true",
    )
    parser.add_argument(
        "--commit-only",
        help="Do not open PRs for updates, only commit changes "
        "to external data (implies --update)",
        action="store_true",
    )
    parser.add_argument(
        "--edit-only",
        help="Do not commit changes, only update files (implies --update)",
        action="store_true",
    )
    parser.add_argument(
        "--check-outdated",
        help="Exit with non-zero status if outdated sources were found and not updated",
        action="store_true",
    )
    parser.add_argument(
        "--filter-type",
        help="Only check external data of the given type",
        type=ExternalData.Type,
        choices=list(ExternalData.Type),
    )

    fork = parser.add_argument_group(
        "control forking behaviour",
        "By default, %(prog)s pushes directly to the GitHub repo if the GitHub "
        "token has permission to do so, and creates a fork if not.",
    ).add_mutually_exclusive_group()
    fork.add_argument(
        "--always-fork",
        action="store_const",
        const=True,
        dest="fork",
        help=(
            "Always push to a fork, even if the user has write access to the "
            "upstream repo"
        ),
    )
    fork.add_argument(
        "--never-fork",
        action="store_const",
        const=False,
        dest="fork",
        help=(
            "Never push to a fork, even if this means failing to push to the "
            "upstream repo"
        ),
    )
    parser.add_argument(
        "--unsafe",
        help="Enable unsafe features; use only with manifests from trusted sources",
        action="store_true",
    )
    parser.add_argument(
        "--max-manifest-size",
        help="Maximum manifest file size allowed to load",
        type=int,
        default=manifest.MAX_MANIFEST_SIZE,
    )
    parser.add_argument(
        "--require-important-update",
        help=(
            "Require an update to at least one source with is-important or "
            "is-main-source to save changes to the manifest. If no instances of "
            "is-important or is-main-source are found, assume normal behaviour and "
            "always save changes to the manifest. This is useful to avoid PRs "
            "generated to update a singular unimportant source."
        ),
        action="store_true",
    )
    parser.add_argument(
        "--pr-labels",
        type=str,
        default="",
        help="Comma-separated GitHub labels to add to the pull request",
    )
    parser.add_argument(
        "--commit-to-current-branch",
        action="store_true",
        help=(
            "Commit changes directly to the currently checked out branch"
            " instead of creating a new branch"
        ),
    )

    args = parser.parse_args(cli_args)
    args.pr_labels = [
        label.strip() for label in args.pr_labels.split(",") if label.strip()
    ]

    return args


async def run_with_args(args: argparse.Namespace) -> t.Tuple[int, int, bool]:
    init_logging(logging.DEBUG if args.verbose else logging.INFO)

    should_update = args.update or args.commit_only or args.edit_only
    did_update = False

    options = manifest.CheckerOptions(
        allow_unsafe=args.unsafe,
        max_manifest_size=args.max_manifest_size,
        require_important_update=args.require_important_update,
    )

    manifest_checker = manifest.ManifestChecker(args.manifest, options)

    await manifest_checker.check(args.filter_type)

    outdated_num = print_outdated_external_data(manifest_checker)

    if should_update and outdated_num > 0:
        changes = manifest_checker.update_manifests()
        if changes and not args.edit_only:
            git_checkout = get_manifest_git_checkout(args.manifest)
            ensure_git_safe_directory(git_checkout)
            with indir(git_checkout):
                committed_changes = commit_changes(
                    changes, commit_to_base=args.commit_to_current_branch
                )
                if not committed_changes:
                    return (-1, -1, False)
                if not (args.commit_only or args.commit_to_current_branch):
                    open_pr(
                        committed_changes,
                        manifest_checker=manifest_checker,
                        fork=args.fork,
                        pr_labels=args.pr_labels,
                    )
        did_update = True

    errors_num = print_errors(manifest_checker)

    log.log(
        logging.WARNING if errors_num else logging.INFO,
        "Check finished with %i error(s)",
        errors_num,
    )

    return outdated_num, errors_num, did_update


class ResultCode(IntFlag):
    SUCCESS = 0
    ERROR = 1
    OUTDATED = 2


def main():
    res = ResultCode.SUCCESS
    args = parse_cli_args()
    outdated_num, errors_num, updated = asyncio.run(run_with_args(args))
    if (outdated_num, errors_num, updated) == (-1, -1, False) or errors_num:
        res |= ResultCode.ERROR
    if args.check_outdated and not updated and outdated_num > 0:
        res |= ResultCode.OUTDATED
    sys.exit(res)

1	#!/usr/bin/env python3
2	#
3	# flatpak-extra-data-checker: A tool for checking the status of
4	# the extra data in a Flatpak manifest.
5	#
6	# Copyright (C) 2018 Endless Mobile, Inc.
7	#
8	# Authors:
9	# Joaquim Rocha <jrocha@endlessm.com>
10	#
11	# This program is free software; you can redistribute it and/or modify
12	# it under the terms of the GNU General Public License as published by
13	# the Free Software Foundation; either version 2 of the License, or
14	# (at your option) any later version.
15	#
16	# This program is distributed in the hope that it will be useful,
17	# but WITHOUT ANY WARRANTY; without even the implied warranty of
18	# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
19	# GNU General Public License for more details.
20	#
21	# You should have received a copy of the GNU General Public License along
22	# with this program; if not, write to the Free Software Foundation, Inc.,
23	# 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
24
25	import argparse	1✔
26	import contextlib	1✔
27	import getpass	1✔
28	import json	1✔
29	import logging	1✔
30	import os	1✔
31	from pathlib import Path	1✔
32	import subprocess	1✔
33	import sys	1✔
34	import asyncio	1✔
35	from enum import IntFlag	1✔
36	import typing as t	1✔
37
38	from github import Github	1✔
39
40	from .lib.utils import parse_github_url, init_logging	1✔
41	from .lib.externaldata import ExternalData	1✔
42	from . import manifest	1✔
43
44
45	log = logging.getLogger(__name__)	1✔
46
47
48	@contextlib.contextmanager	1✔
49	def indir(path: Path):	1✔
50	"""
51	>>> with indir(path):
52	... # code executes with 'path' as working directory
53	... # old working directory is restored
54	"""
55
56	old = os.getcwd()	1✔
57	os.chdir(path)	1✔
58	try:	1✔
59	yield	1✔
60	finally:
61	os.chdir(old)	1✔
62
63
64	def print_outdated_external_data(manifest_checker: manifest.ManifestChecker):	1✔
65	ext_data = manifest_checker.get_outdated_external_data()	1✔
66	for data in ext_data:	1✔
67	state_txt = data.state.name or str(data.state)	1✔
68	message_tmpl = ""	1✔
69	message_args = {}	1✔
70	if data.new_version:	1✔
71	if data.type == data.Type.GIT:	1✔
72	assert data.new_version	1✔
73	message_tmpl = (	1✔
74	"{data_state}: {data_name}\n"
75	" Has a new version:\n"
76	" URL: {url}\n"
77	" Commit: {commit}\n"
78	" Tag: {tag}\n"
79	" Branch: {branch}\n"
80	" Version: {version}\n"
81	" Timestamp: {timestamp}\n"
82	)
83	message_args = data.new_version._asdict()	1✔
84	else:
85	assert isinstance(data, ExternalData)	1✔
86	assert data.new_version	1✔
87	message_tmpl = (	1✔
88	"{data_state}: {data_name}\n"
89	" Has a new version:\n"
90	" URL: {url}\n"
91	" MD5: {md5}\n"
92	" SHA1: {sha1}\n"
93	" SHA256: {sha256}\n"
94	" SHA512: {sha512}\n"
95	" Size: {size}\n"
96	" Version: {version}\n"
97	" Timestamp: {timestamp}\n"
98	)
99	message_args = {	1✔
100	**data.new_version._asdict(),
101	**data.new_version.checksum._asdict(),
102	}
103	elif data.State.BROKEN in data.state:	×
104	message_tmpl = (	×
105	# fmt: off
106	"{data_state}: {data_name}\n"
107	" Couldn't get new version for {url}\n"
108	# fmt: on
109	)
110	message_args = data.current_version._asdict()	×
111
112	message = message_tmpl.format(	1✔
113	data_state=state_txt,
114	data_name=data.filename,
115	**message_args,
116	)
117	print(message, flush=True)	1✔
118	return len(ext_data)	1✔
119
120
121	def print_errors(manifest_checker: manifest.ManifestChecker) -> int:	1✔
122	# TODO: Actually do pretty-print collected errors
123	errors = manifest_checker.get_errors()	1✔
124	return len(errors)	1✔
125
126
127	def check_call(args):	1✔
128	log.debug("$ %s", " ".join(args))	1✔
129	subprocess.check_call(args)	1✔
130
131
132	def get_manifest_git_checkout(manifest: t.Union[Path, str]) -> Path:	1✔
133	# Can't use git rev-parse --show-toplevel because of a chicken-and-egg problem: we
134	# need to find the checkout directory so that we can mark it as safe so that we can
135	# use git against it.
136	for directory in Path(manifest).parents:	1✔
137	if os.path.exists(directory / ".git"):	1✔
138	return directory	1✔
139
140	raise FileNotFoundError(f"Cannot find git checkout for {manifest}")	×
141
142
143	def ensure_git_safe_directory(checkout: Path):	1✔
144	uid = os.getuid()	1✔
145	checkout_uid = os.stat(checkout).st_uid	1✔
146	if uid == checkout_uid:	1✔
147	return	1✔
148
149	try:	×
150	result = subprocess.run(	×
151	["git", "config", "--get-all", "safe.directory"],
152	check=True,
153	capture_output=True,
154	encoding="utf-8",
155	)
156	safe_dirs = [Path(x) for x in result.stdout.splitlines()]	×
157	except subprocess.CalledProcessError as err:	×
158	# git config --get-all will return 1 if the key doesn't exist.
159	# Re-raise the error for anything else.
160	if err.returncode != 1:	×
161	raise	×
162	safe_dirs = []	×
163
164	if checkout in safe_dirs:	×
165	return	×
166
167	log.info("Adding %s git safe directory", checkout)	×
168	location = "--system" if uid == 0 else "--global"	×
169	check_call(["git", "config", location, "--add", "safe.directory", str(checkout)])	×
170
171
172	class CommittedChanges(t.NamedTuple):	1✔
173	subject: str	1✔
174	body: t.Optional[str]	1✔
175	commit: t.Optional[str]	1✔
176	branch: str	1✔
177	base_branch: t.Optional[str]	1✔
178
179
180	def commit_message(changes: t.List[str]) -> str:	1✔
181	assert len(changes) >= 1	1✔
182
183	if len(changes) == 1:	1✔
184	return changes[0]	1✔
185
186	module_names = list(dict.fromkeys(list(i.split(":", 1)[0] for i in changes)))	1✔
187
188	if len(module_names) == 1:	1✔
189	return f"Update {module_names[0]} module"	×
190
191	for i in reversed(range(2, len(module_names) + 1)):	1✔
192	xs = module_names[: i - 1]	1✔
193	y = module_names[i - 1]	1✔
194	zs = module_names[i:]	1✔
195
196	if zs:	1✔
197	tail = f" and {len(zs)} more modules"	×
198	xs.append(y)	×
199	else:
200	tail = f" and {y} modules"	1✔
201
202	subject = "Update " + ", ".join(xs) + tail	1✔
203	if len(subject) <= 70:	1✔
204	return subject	1✔
205
206	return f"Update {len(module_names)} modules"	×
207
208
209	def branch_exists(branch: str) -> bool:	1✔
210	try:	1✔
211	subprocess.run(	1✔
212	["git", "rev-parse", "--verify", branch],
213	capture_output=True,
214	check=True,
215	)
216	return True	1✔
217	except subprocess.CalledProcessError:	1✔
218	return False	1✔
219
220
221	def commit_changes(	1✔
222	changes: t.List[str], commit_to_base: bool = False
223	) -> t.Optional[CommittedChanges]:
224	log.info("Committing updates")	1✔
225	body: t.Optional[str]
226	subject = commit_message(changes)	1✔
227	if len(changes) > 1:	1✔
228	body = "\n".join(changes)	1✔
229	message = subject + "\n\n" + body	1✔
230	else:
231	body = None	1✔
232	message = subject	1✔
233
234	# Remember the base branch
235	base_branch: t.Optional[str]
236	base_branch = subprocess.check_output(	1✔
237	["git", "branch", "--show-current"], text=True
238	).strip()
239	if not base_branch:	1✔
240	base_branch = None	1✔
241
242	if commit_to_base:	1✔
243	if not base_branch:	1✔
244	log.error("Failed to determine the base branch: cannot commit to it")	1✔
245	return None	1✔
246	else:
247	branch = base_branch	1✔
248	tree = None	1✔
249	else:
250	# Moved to detached HEAD
251	log.info("Switching to detached HEAD")	1✔
252	check_call(["git", "-c", "advice.detachedHead=false", "checkout", "HEAD@{0}"])	1✔
253	# Find a stable identifier for the contents of the tree, to avoid
254	# sending the same PR twice.
255	tree = subprocess.check_output(	1✔
256	["git", "rev-parse", "HEAD^{tree}"], text=True
257	).strip()
258	branch = (	1✔
259	f"update-{base_branch}-{tree[:7]}" if base_branch else f"update-{tree[:7]}"
260	)
261
262	retry_commit = False	1✔
263	try:	1✔
264	check_call(["git", "commit", "-am", message])	1✔
265	except subprocess.CalledProcessError:	1✔
266	retry_commit = True	1✔
267	pass	1✔
268
269	if retry_commit:	1✔
270	log.warning("Committing failed. Falling back to a sanitised config")	1✔
271	git_name = getpass.getuser()	1✔
272	git_email = git_name + "@" + "localhost"	1✔
273	assert git_email is not None	1✔
274	env = {	1✔
275	"GIT_CONFIG_GLOBAL": "/dev/null",
276	"GIT_CONFIG_NOSYSTEM": "1",
277	"GIT_CONFIG": "''",
278	}
279	subprocess.run(	1✔
280	[
281	"git",
282	"-c",
283	f"user.name={git_name}",
284	"-c",
285	f"user.email={git_email}",
286	"commit",
287	"--no-verify",
288	"--no-gpg-sign",
289	"-am",
290	message,
291	],
292	check=True,
293	env=env,
294	)
295
296	if not (commit_to_base or branch_exists(branch)):	1✔
297	check_call(["git", "checkout", "-b", branch])	1✔
298	return CommittedChanges(	1✔
299	subject=subject,
300	body=body,
301	commit=tree,
302	branch=branch,
303	base_branch=base_branch,
304	)
305
306
307	DISCLAIMER = (	1✔
308	"🤖 This pull request was automatically generated by "
309	"[flathub-infra/flatpak-external-data-checker]"
310	"(https://github.com/flathub-infra/flatpak-external-data-checker). "
311	"Please [open an issue]"
312	"(https://github.com/flathub-infra/flatpak-external-data-checker/issues/new) "
313	"if you have any questions or complaints. 🤖"
314	)
315
316	AUTOMERGE_DUE_TO_CONFIG = (	1✔
317	"🤖 This PR passed CI, and `automerge-flathubbot-prs` is `true` in "
318	"`flathub.json`, so I'm merging it automatically. 🤖"
319	)
320
321	AUTOMERGE_DUE_TO_BROKEN_URLS = (	1✔
322	"🤖 The currently-published version contains broken URLs, and this PR passed "
323	"CI, so I'm merging it automatically. You can disable this behaviour by setting "
324	"`automerge-flathubbot-prs` to `false` in flathub.json. 🤖"
325	)
326
327
328	def open_pr(	1✔
329	change: CommittedChanges,
330	manifest_checker: t.Optional[manifest.ManifestChecker] = None,
331	fork: t.Optional[bool] = None,
332	pr_labels: t.Optional[t.List[str]] = None,
333	):
334
335	try:	×
336	github_token = os.environ["GITHUB_TOKEN"]	×
337	except KeyError:	×
338	log.error("GITHUB_TOKEN environment variable is not set")	×
339	sys.exit(1)	×
340
341	log.info("Opening pull request for branch %s", change.branch)	×
342	g = Github(github_token)	×
343	user = g.get_user()	×
344
345	origin_url = (	×
346	subprocess.check_output(["git", "remote", "get-url", "origin"])
347	.decode("utf-8")
348	.strip()
349	)
350	origin_repo = g.get_repo(parse_github_url(origin_url))	×
351
352	if fork is True:	×
353	log.debug("creating fork (as requested)")	×
354	repo = user.create_fork(origin_repo)	×
355	elif fork is False:	×
356	log.debug("not creating fork (as requested)")	×
357	repo = origin_repo	×
358	elif origin_repo.permissions.push:	×
359	log.debug("origin repo is writable; not creating fork")	×
360	repo = origin_repo	×
361	else:
362	log.debug("origin repo not writable; creating fork")	×
363	repo = user.create_fork(origin_repo)	×
364
365	remote_url = f"https://{github_token}:x-oauth-basic@github.com/{repo.full_name}"	×
366
367	base = change.base_branch or origin_repo.default_branch	×
368
369	head = "{}:{}".format(repo.owner.login, change.branch)	×
370	pr_message = ((change.body or "") + "\n\n" + DISCLAIMER).strip()	×
371
372	try:	×
373	with open("flathub.json") as f:	×
374	repocfg = json.load(f)	×
375	except FileNotFoundError:	×
376	repocfg = {}	×
377
378	automerge = repocfg.get("automerge-flathubbot-prs")	×
379	# Implicitly automerge if…
380	force_automerge = (	×
381	# …the user has not explicitly disabled automerge…
382	automerge is not False
383	# …and we have a manifest checker (i.e. we're not in a test)…
384	and manifest_checker
385	# …and at least one source is broken and has an update
386	and any(
387	data.type == data.Type.EXTRA_DATA
388	and data.State.BROKEN in data.state
389	and data.new_version
390	for data in manifest_checker.get_outdated_external_data()
391	)
392	)
393
394	prs = origin_repo.get_pulls(state="all", base=base, head=head)	×
395
396	# If the maintainer has closed our last PR or it was merged,
397	# we don't want to open another one.
398	closed_prs = [pr for pr in prs if pr.state == "closed"]	×
399	for pr in closed_prs:	×
400	log.info(	×
401	"Found existing %s PR: %s",
402	"merged" if pr.is_merged() else pr.state,
403	pr.html_url,
404	)
405	return	×
406
407	open_prs = [pr for pr in prs if pr.state == "open"]	×
408	for pr in open_prs:	×
409	log.info("Found open PR: %s", pr.html_url)	×
410
411	if automerge or force_automerge:	×
412	pr_commit = pr.head.repo.get_commit(pr.head.sha)	×
413	if pr_commit.get_combined_status().state == "success" and pr.mergeable:	×
414	log.info("PR passed CI and is mergeable, merging %s", pr.html_url)	×
415	if automerge:	×
416	pr.create_issue_comment(AUTOMERGE_DUE_TO_CONFIG)	×
417	else: # force_automerge
418	pr.create_issue_comment(AUTOMERGE_DUE_TO_BROKEN_URLS)	×
419	pr.merge(merge_method="rebase")	×
420	origin_repo.get_git_ref(f"heads/{pr.head.ref}").delete()	×
421
422	return	×
423
424	check_call(["git", "push", "-u", remote_url, change.branch])	×
425
426	log.info(	×
427	"Creating pull request in %s from head `%s` to base `%s`",
428	origin_repo.html_url,
429	head,
430	base,
431	)
432
433	gh_run_id = os.environ.get("GITHUB_RUN_ID")	×
434	gh_repo_name = os.environ.get("GITHUB_REPOSITORY")	×
435	if gh_run_id and gh_repo_name:	×
436	log.info("Appending GitHub actions log URL to PR message")	×
437	log_url = f"https://github.com/{gh_repo_name}/actions/runs/{gh_run_id}"	×
438	pr_message += f"\n\n[📋 View External data checker logs]({log_url})"	×
439
440	pr = origin_repo.create_pull(	×
441	change.subject,
442	pr_message,
443	base,
444	head,
445	maintainer_can_modify=True,
446	)
447	log.info("Opened pull request %s", pr.html_url)	×
448	if pr_labels:	×
449	log.info("Adding labels to PR: %s", ", ".join(pr_labels))	×
450	pr.set_labels(*pr_labels)	×
451
452
453	def parse_cli_args(cli_args=None):	1✔
454	parser = argparse.ArgumentParser()	1✔
455	parser.add_argument(	1✔
456	"manifest", help="Flatpak manifest to check", type=os.path.abspath
457	)
458	parser.add_argument(	1✔
459	"-v", "--verbose", help="Print debug messages", action="store_true"
460	)
461	parser.add_argument(	1✔
462	"--update",
463	help="Update manifest(s) to refer to new versions of "
464	"external data - also open PRs for changes unless "
465	"--commit-only is specified",
466	action="store_true",
467	)
468	parser.add_argument(	1✔
469	"--commit-only",
470	help="Do not open PRs for updates, only commit changes "
471	"to external data (implies --update)",
472	action="store_true",
473	)
474	parser.add_argument(	1✔
475	"--edit-only",
476	help="Do not commit changes, only update files (implies --update)",
477	action="store_true",
478	)
479	parser.add_argument(	1✔
480	"--check-outdated",
481	help="Exit with non-zero status if outdated sources were found and not updated",
482	action="store_true",
483	)
484	parser.add_argument(	1✔
485	"--filter-type",
486	help="Only check external data of the given type",
487	type=ExternalData.Type,
488	choices=list(ExternalData.Type),
489	)
490
491	fork = parser.add_argument_group(	1✔
492	"control forking behaviour",
493	"By default, %(prog)s pushes directly to the GitHub repo if the GitHub "
494	"token has permission to do so, and creates a fork if not.",
495	).add_mutually_exclusive_group()
496	fork.add_argument(	1✔
497	"--always-fork",
498	action="store_const",
499	const=True,
500	dest="fork",
501	help=(
502	"Always push to a fork, even if the user has write access to the "
503	"upstream repo"
504	),
505	)
506	fork.add_argument(	1✔
507	"--never-fork",
508	action="store_const",
509	const=False,
510	dest="fork",
511	help=(
512	"Never push to a fork, even if this means failing to push to the "
513	"upstream repo"
514	),
515	)
516	parser.add_argument(	1✔
517	"--unsafe",
518	help="Enable unsafe features; use only with manifests from trusted sources",
519	action="store_true",
520	)
521	parser.add_argument(	1✔
522	"--max-manifest-size",
523	help="Maximum manifest file size allowed to load",
524	type=int,
525	default=manifest.MAX_MANIFEST_SIZE,
526	)
527	parser.add_argument(	1✔
528	"--require-important-update",
529	help=(
530	"Require an update to at least one source with is-important or "
531	"is-main-source to save changes to the manifest. If no instances of "
532	"is-important or is-main-source are found, assume normal behaviour and "
533	"always save changes to the manifest. This is useful to avoid PRs "
534	"generated to update a singular unimportant source."
535	),
536	action="store_true",
537	)
538	parser.add_argument(	1✔
539	"--pr-labels",
540	type=str,
541	default="",
542	help="Comma-separated GitHub labels to add to the pull request",
543	)
544	parser.add_argument(	1✔
545	"--commit-to-current-branch",
546	action="store_true",
547	help=(
548	"Commit changes directly to the currently checked out branch"
549	" instead of creating a new branch"
550	),
551	)
552
553	args = parser.parse_args(cli_args)	1✔
554	args.pr_labels = [	1✔
555	label.strip() for label in args.pr_labels.split(",") if label.strip()
556	]
557
558	return args	1✔
559
560
561	async def run_with_args(args: argparse.Namespace) -> t.Tuple[int, int, bool]:	1✔
562	init_logging(logging.DEBUG if args.verbose else logging.INFO)	1✔
563
564	should_update = args.update or args.commit_only or args.edit_only	1✔
565	did_update = False	1✔
566
567	options = manifest.CheckerOptions(	1✔
568	allow_unsafe=args.unsafe,
569	max_manifest_size=args.max_manifest_size,
570	require_important_update=args.require_important_update,
571	)
572
573	manifest_checker = manifest.ManifestChecker(args.manifest, options)	1✔
574
575	await manifest_checker.check(args.filter_type)	1✔
576
577	outdated_num = print_outdated_external_data(manifest_checker)	1✔
578
579	if should_update and outdated_num > 0:	1✔
580	changes = manifest_checker.update_manifests()	1✔
581	if changes and not args.edit_only:	1✔
582	git_checkout = get_manifest_git_checkout(args.manifest)	1✔
583	ensure_git_safe_directory(git_checkout)	1✔
584	with indir(git_checkout):	1✔
585	committed_changes = commit_changes(	1✔
586	changes, commit_to_base=args.commit_to_current_branch
587	)
588	if not committed_changes:	1✔
589	return (-1, -1, False)	1✔
590	if not (args.commit_only or args.commit_to_current_branch):	1✔
UNCOV 591	open_pr(	×
592	committed_changes,
593	manifest_checker=manifest_checker,
594	fork=args.fork,
595	pr_labels=args.pr_labels,
596	)
597	did_update = True	1✔
598
599	errors_num = print_errors(manifest_checker)	1✔
600
601	log.log(	1✔
602	logging.WARNING if errors_num else logging.INFO,
603	"Check finished with %i error(s)",
604	errors_num,
605	)
606
607	return outdated_num, errors_num, did_update	1✔
608
609
610	class ResultCode(IntFlag):	1✔
611	SUCCESS = 0	1✔
612	ERROR = 1	1✔
613	OUTDATED = 2	1✔
614
615
616	def main():	1✔
617	res = ResultCode.SUCCESS	×
618	args = parse_cli_args()	×
619	outdated_num, errors_num, updated = asyncio.run(run_with_args(args))	×
NEW 620	if (outdated_num, errors_num, updated) == (-1, -1, False) or errors_num:	×
621	res \|= ResultCode.ERROR	×
622	if args.check_outdated and not updated and outdated_num > 0:	×
623	res \|= ResultCode.OUTDATED	×
624	sys.exit(res)	×

flathub / flatpak-external-data-checker / 18528321607

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous