14952738270

Committed 11 May 2025 05:34AM UTC coverage: 90.478% (-0.4%) from 90.886%

Build # 14952738270

Build Type

Pull #467

github

Committed by

web-flow

Commit Message

Merge 3b649f7f5 into 5c3846753

Pull Request Pull Request #467: Fix some issues

Run Details

6 of 16 new or added lines in 1 file covered. (37.5%)

3 existing lines in 2 files now uncovered.

2157 of 2384 relevant lines covered (90.48%)

0.9 hits per line

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

58.33

/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: 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 commit_changes(changes: t.List[str]) -> 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

    # Moved to detached HEAD
    log.info("Switching to detached HEAD")
    check_call(["git", "-c", "advice.detachedHead=false", "checkout", "HEAD@{0}"])

    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,
        )

    # 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()
    if base_branch:
        branch = f"update-{base_branch}-{tree[:7]}"
    else:
        branch = f"update-{tree[:7]}"

    try:
        # Check if the branch already exists
        subprocess.run(
            ["git", "rev-parse", "--verify", branch],
            capture_output=True,
            check=True,
        )
    except subprocess.CalledProcessError:
        # If not, create it
        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",
    )

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

flathub / flatpak-external-data-checker / 14952738270

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