18454455609

Committed 13 Oct 2025 03:41AM UTC coverage: 90.579% (+0.2%) from 90.36%

Build # 18454455609

Build Type

Pull #485

github

Committed by

web-flow

Commit Message

Merge 51bfb9a9d into f7f982bd1

Pull Request Pull Request #485: main: Support committing directly to a given branch

Run Details

26 of 27 new or added lines in 1 file covered. (96.3%)

1 existing line in 1 file now uncovered.

2173 of 2399 relevant lines covered (90.58%)

0.91 hits per line

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

61.89

/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], target_branch: t.Optional[str] = None
) -> 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 target_branch:
        log.info("Committing directly to existing branch '%s'", target_branch)
        if not branch_exists(target_branch):
            log.error("Branch '%s' does not exist locally", target_branch)
            return None
        check_call(["git", "checkout", target_branch])
        branch = target_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_email = getpass.getuser() + "@" + "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.email={git_email}",
                "commit",
                "--no-verify",
                "--no-gpg-sign",
                "-am",
                message,
            ],
            check=True,
            env=env,
        )

    if not (target_branch 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-branch",
        type=str,
        help=(
            "Commit changes directly to this local branch. "
            "The branch must already exist"
        ),
    )

    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, target_branch=args.commit_branch
                )
                if not committed_changes:
                    return (-1, -1, False)
                if not (args.commit_only or args.commit_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], target_branch: t.Optional[str] = None
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	×
241
242	if target_branch:	1✔
243	log.info("Committing directly to existing branch '%s'", target_branch)	1✔
244	if not branch_exists(target_branch):	1✔
245	log.error("Branch '%s' does not exist locally", target_branch)	1✔
246	return None	1✔
247	check_call(["git", "checkout", target_branch])	1✔
248	branch = target_branch	1✔
249	tree = None	1✔
250	else:
251	# Moved to detached HEAD
252	log.info("Switching to detached HEAD")	1✔
253	check_call(["git", "-c", "advice.detachedHead=false", "checkout", "HEAD@{0}"])	1✔
254	# Find a stable identifier for the contents of the tree, to avoid
255	# sending the same PR twice.
256	tree = subprocess.check_output(	1✔
257	["git", "rev-parse", "HEAD^{tree}"], text=True
258	).strip()
259	branch = (	1✔
260	f"update-{base_branch}-{tree[:7]}" if base_branch else f"update-{tree[:7]}"
261	)
262
263	retry_commit = False	1✔
264	try:	1✔
265	check_call(["git", "commit", "-am", message])	1✔
266	except subprocess.CalledProcessError:	×
267	retry_commit = True	×
268	pass	×
269
270	if retry_commit:	1✔
271	log.warning("Committing failed. Falling back to a sanitised config")	×
272	git_email = getpass.getuser() + "@" + "localhost"	×
273	assert git_email is not None	×
274	env = {	×
275	"GIT_CONFIG_GLOBAL": "/dev/null",
276	"GIT_CONFIG_NOSYSTEM": "1",
277	"GIT_CONFIG": "''",
278	}
279	subprocess.run(	×
280	[
281	"git",
282	"-c",
283	f"user.email={git_email}",
284	"commit",
285	"--no-verify",
286	"--no-gpg-sign",
287	"-am",
288	message,
289	],
290	check=True,
291	env=env,
292	)
293
294	if not (target_branch or branch_exists(branch)):	1✔
295	check_call(["git", "checkout", "-b", branch])	1✔
296	return CommittedChanges(	1✔
297	subject=subject,
298	body=body,
299	commit=tree,
300	branch=branch,
301	base_branch=base_branch,
302	)
303
304
305	DISCLAIMER = (	1✔
306	"🤖 This pull request was automatically generated by "
307	"[flathub-infra/flatpak-external-data-checker]"
308	"(https://github.com/flathub-infra/flatpak-external-data-checker). "
309	"Please [open an issue]"
310	"(https://github.com/flathub-infra/flatpak-external-data-checker/issues/new) "
311	"if you have any questions or complaints. 🤖"
312	)
313
314	AUTOMERGE_DUE_TO_CONFIG = (	1✔
315	"🤖 This PR passed CI, and `automerge-flathubbot-prs` is `true` in "
316	"`flathub.json`, so I'm merging it automatically. 🤖"
317	)
318
319	AUTOMERGE_DUE_TO_BROKEN_URLS = (	1✔
320	"🤖 The currently-published version contains broken URLs, and this PR passed "
321	"CI, so I'm merging it automatically. You can disable this behaviour by setting "
322	"`automerge-flathubbot-prs` to `false` in flathub.json. 🤖"
323	)
324
325
326	def open_pr(	1✔
327	change: CommittedChanges,
328	manifest_checker: t.Optional[manifest.ManifestChecker] = None,
329	fork: t.Optional[bool] = None,
330	pr_labels: t.Optional[t.List[str]] = None,
331	):
332
333	try:	×
334	github_token = os.environ["GITHUB_TOKEN"]	×
335	except KeyError:	×
336	log.error("GITHUB_TOKEN environment variable is not set")	×
337	sys.exit(1)	×
338
339	log.info("Opening pull request for branch %s", change.branch)	×
340	g = Github(github_token)	×
341	user = g.get_user()	×
342
343	origin_url = (	×
344	subprocess.check_output(["git", "remote", "get-url", "origin"])
345	.decode("utf-8")
346	.strip()
347	)
348	origin_repo = g.get_repo(parse_github_url(origin_url))	×
349
350	if fork is True:	×
351	log.debug("creating fork (as requested)")	×
352	repo = user.create_fork(origin_repo)	×
353	elif fork is False:	×
354	log.debug("not creating fork (as requested)")	×
355	repo = origin_repo	×
356	elif origin_repo.permissions.push:	×
357	log.debug("origin repo is writable; not creating fork")	×
358	repo = origin_repo	×
359	else:
360	log.debug("origin repo not writable; creating fork")	×
361	repo = user.create_fork(origin_repo)	×
362
363	remote_url = f"https://{github_token}:x-oauth-basic@github.com/{repo.full_name}"	×
364
365	base = change.base_branch or origin_repo.default_branch	×
366
367	head = "{}:{}".format(repo.owner.login, change.branch)	×
368	pr_message = ((change.body or "") + "\n\n" + DISCLAIMER).strip()	×
369
370	try:	×
371	with open("flathub.json") as f:	×
372	repocfg = json.load(f)	×
373	except FileNotFoundError:	×
374	repocfg = {}	×
375
376	automerge = repocfg.get("automerge-flathubbot-prs")	×
377	# Implicitly automerge if…
378	force_automerge = (	×
379	# …the user has not explicitly disabled automerge…
380	automerge is not False
381	# …and we have a manifest checker (i.e. we're not in a test)…
382	and manifest_checker
383	# …and at least one source is broken and has an update
384	and any(
385	data.type == data.Type.EXTRA_DATA
386	and data.State.BROKEN in data.state
387	and data.new_version
388	for data in manifest_checker.get_outdated_external_data()
389	)
390	)
391
392	prs = origin_repo.get_pulls(state="all", base=base, head=head)	×
393
394	# If the maintainer has closed our last PR or it was merged,
395	# we don't want to open another one.
396	closed_prs = [pr for pr in prs if pr.state == "closed"]	×
397	for pr in closed_prs:	×
398	log.info(	×
399	"Found existing %s PR: %s",
400	"merged" if pr.is_merged() else pr.state,
401	pr.html_url,
402	)
403	return	×
404
405	open_prs = [pr for pr in prs if pr.state == "open"]	×
406	for pr in open_prs:	×
407	log.info("Found open PR: %s", pr.html_url)	×
408
409	if automerge or force_automerge:	×
410	pr_commit = pr.head.repo.get_commit(pr.head.sha)	×
411	if pr_commit.get_combined_status().state == "success" and pr.mergeable:	×
412	log.info("PR passed CI and is mergeable, merging %s", pr.html_url)	×
413	if automerge:	×
414	pr.create_issue_comment(AUTOMERGE_DUE_TO_CONFIG)	×
415	else: # force_automerge
416	pr.create_issue_comment(AUTOMERGE_DUE_TO_BROKEN_URLS)	×
417	pr.merge(merge_method="rebase")	×
418	origin_repo.get_git_ref(f"heads/{pr.head.ref}").delete()	×
419
420	return	×
421
422	check_call(["git", "push", "-u", remote_url, change.branch])	×
423
424	log.info(	×
425	"Creating pull request in %s from head `%s` to base `%s`",
426	origin_repo.html_url,
427	head,
428	base,
429	)
430
431	gh_run_id = os.environ.get("GITHUB_RUN_ID")	×
432	gh_repo_name = os.environ.get("GITHUB_REPOSITORY")	×
433	if gh_run_id and gh_repo_name:	×
434	log.info("Appending GitHub actions log URL to PR message")	×
435	log_url = f"https://github.com/{gh_repo_name}/actions/runs/{gh_run_id}"	×
436	pr_message += f"\n\n[📋 View External data checker logs]({log_url})"	×
437
438	pr = origin_repo.create_pull(	×
439	change.subject,
440	pr_message,
441	base,
442	head,
443	maintainer_can_modify=True,
444	)
445	log.info("Opened pull request %s", pr.html_url)	×
446	if pr_labels:	×
447	log.info("Adding labels to PR: %s", ", ".join(pr_labels))	×
448	pr.set_labels(*pr_labels)	×
449
450
451	def parse_cli_args(cli_args=None):	1✔
452	parser = argparse.ArgumentParser()	1✔
453	parser.add_argument(	1✔
454	"manifest", help="Flatpak manifest to check", type=os.path.abspath
455	)
456	parser.add_argument(	1✔
457	"-v", "--verbose", help="Print debug messages", action="store_true"
458	)
459	parser.add_argument(	1✔
460	"--update",
461	help="Update manifest(s) to refer to new versions of "
462	"external data - also open PRs for changes unless "
463	"--commit-only is specified",
464	action="store_true",
465	)
466	parser.add_argument(	1✔
467	"--commit-only",
468	help="Do not open PRs for updates, only commit changes "
469	"to external data (implies --update)",
470	action="store_true",
471	)
472	parser.add_argument(	1✔
473	"--edit-only",
474	help="Do not commit changes, only update files (implies --update)",
475	action="store_true",
476	)
477	parser.add_argument(	1✔
478	"--check-outdated",
479	help="Exit with non-zero status if outdated sources were found and not updated",
480	action="store_true",
481	)
482	parser.add_argument(	1✔
483	"--filter-type",
484	help="Only check external data of the given type",
485	type=ExternalData.Type,
486	choices=list(ExternalData.Type),
487	)
488
489	fork = parser.add_argument_group(	1✔
490	"control forking behaviour",
491	"By default, %(prog)s pushes directly to the GitHub repo if the GitHub "
492	"token has permission to do so, and creates a fork if not.",
493	).add_mutually_exclusive_group()
494	fork.add_argument(	1✔
495	"--always-fork",
496	action="store_const",
497	const=True,
498	dest="fork",
499	help=(
500	"Always push to a fork, even if the user has write access to the "
501	"upstream repo"
502	),
503	)
504	fork.add_argument(	1✔
505	"--never-fork",
506	action="store_const",
507	const=False,
508	dest="fork",
509	help=(
510	"Never push to a fork, even if this means failing to push to the "
511	"upstream repo"
512	),
513	)
514	parser.add_argument(	1✔
515	"--unsafe",
516	help="Enable unsafe features; use only with manifests from trusted sources",
517	action="store_true",
518	)
519	parser.add_argument(	1✔
520	"--max-manifest-size",
521	help="Maximum manifest file size allowed to load",
522	type=int,
523	default=manifest.MAX_MANIFEST_SIZE,
524	)
525	parser.add_argument(	1✔
526	"--require-important-update",
527	help=(
528	"Require an update to at least one source with is-important or "
529	"is-main-source to save changes to the manifest. If no instances of "
530	"is-important or is-main-source are found, assume normal behaviour and "
531	"always save changes to the manifest. This is useful to avoid PRs "
532	"generated to update a singular unimportant source."
533	),
534	action="store_true",
535	)
536	parser.add_argument(	1✔
537	"--pr-labels",
538	type=str,
539	default="",
540	help="Comma-separated GitHub labels to add to the pull request",
541	)
542	parser.add_argument(	1✔
543	"--commit-branch",
544	type=str,
545	help=(
546	"Commit changes directly to this local branch. "
547	"The branch must already exist"
548	),
549	)
550
551	args = parser.parse_args(cli_args)	1✔
552	args.pr_labels = [	1✔
553	label.strip() for label in args.pr_labels.split(",") if label.strip()
554	]
555
556	return args	1✔
557
558
559	async def run_with_args(args: argparse.Namespace) -> t.Tuple[int, int, bool]:	1✔
560	init_logging(logging.DEBUG if args.verbose else logging.INFO)	1✔
561
562	should_update = args.update or args.commit_only or args.edit_only	1✔
563	did_update = False	1✔
564
565	options = manifest.CheckerOptions(	1✔
566	allow_unsafe=args.unsafe,
567	max_manifest_size=args.max_manifest_size,
568	require_important_update=args.require_important_update,
569	)
570
571	manifest_checker = manifest.ManifestChecker(args.manifest, options)	1✔
572
573	await manifest_checker.check(args.filter_type)	1✔
574
575	outdated_num = print_outdated_external_data(manifest_checker)	1✔
576
577	if should_update and outdated_num > 0:	1✔
578	changes = manifest_checker.update_manifests()	1✔
579	if changes and not args.edit_only:	1✔
580	git_checkout = get_manifest_git_checkout(args.manifest)	1✔
581	ensure_git_safe_directory(git_checkout)	1✔
582	with indir(git_checkout):	1✔
583	committed_changes = commit_changes(	1✔
584	changes, target_branch=args.commit_branch
585	)
586	if not committed_changes:	1✔
587	return (-1, -1, False)	1✔
588	if not (args.commit_only or args.commit_branch):	1✔
UNCOV 589	open_pr(	×
590	committed_changes,
591	manifest_checker=manifest_checker,
592	fork=args.fork,
593	pr_labels=args.pr_labels,
594	)
595	did_update = True	1✔
596
597	errors_num = print_errors(manifest_checker)	1✔
598
599	log.log(	1✔
600	logging.WARNING if errors_num else logging.INFO,
601	"Check finished with %i error(s)",
602	errors_num,
603	)
604
605	return outdated_num, errors_num, did_update	1✔
606
607
608	class ResultCode(IntFlag):	1✔
609	SUCCESS = 0	1✔
610	ERROR = 1	1✔
611	OUTDATED = 2	1✔
612
613
614	def main():	1✔
615	res = ResultCode.SUCCESS	×
616	args = parse_cli_args()	×
617	outdated_num, errors_num, updated = asyncio.run(run_with_args(args))	×
NEW 618	if (outdated_num, errors_num, updated) == (-1, -1, False) or errors_num:	×
619	res \|= ResultCode.ERROR	×
620	if args.check_outdated and not updated and outdated_num > 0:	×
621	res \|= ResultCode.OUTDATED	×
622	sys.exit(res)	×

flathub / flatpak-external-data-checker / 18454455609

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