14795010914

Committed 02 May 2025 12:16PM UTC coverage: 58.86% (+0.02%) from 58.84%

Build # 14795010914

Build Type

push

github

Committed by

web-flow

Commit Message

Add mypy, autoflake and ruff pre-commit checks (#1331)

* Add mypy pre-commit hook

This adds mypy, autoflake, and ruff as a pre-commit check and addresses all the complaints it
has. This solved a lot of real issues plus some boilerplate.

* Add autoflake pre-commit hook

This adds autoflake as a pre-commit check and addresses all the complaints it has.

* Add ruff pre-commit hook

This adds ruff as a pre-commit check and addresses all the complaints it has. This solved a lot of real issues plus some boilerplate.

Other commits:

* Provide tests with secret
* Remove unused kwargs
* Deal with cases in which the XML attribute lookup is None
* [pre-commit] autoupdate
* Fix: [WARNING] Unexpected key(s) present on https://github.com/psf/black-pre-commit-mirror => black: force-exclude
* Annotate test fixtures
* Remove premature github label cache
* Add annotation to load-tests protocol
* Add comment about munch.Munch not being typed https://github.com/Infinidat/munch/issues/84
* Annotate subparsers parameters
* Remove unused session_headers property from GithubGraphQL

Run Details

176 of 234 new or added lines in 21 files covered. (75.21%)

18 existing lines in 7 files now uncovered.

1239 of 2105 relevant lines covered (58.86%)

0.59 hits per line

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

98.58

/snapshot_manager/snapshot_manager/github_util.py

"""
github_util
"""

import datetime
import enum
import logging
import os
import pathlib

import fnc
import github
import github.GithubException
import github.Issue
import github.IssueComment
import github.Label
import github.PaginatedList
import github.Repository

import snapshot_manager.build_status as build_status
import snapshot_manager.config as config
import snapshot_manager.github_graphql as github_graphql
import snapshot_manager.util as util


@enum.unique
class Reaction(enum.StrEnum):
    """An enum to represent the possible comment reactions"""

    THUMBS_UP = "THUMBS_UP"  # Represents the :+1: emoji.
    THUMBS_DOWN = "THUMBS_DOWN"  # Represents the :-1: emoji.
    LAUGH = "LAUGH"  # Represents the :laugh: emoji.
    HOORAY = "HOORAY"  # Represents the :hooray: emoji.
    CONFUSED = "CONFUSED"  # Represents the :confused: emoji.
    HEART = "HEART"  # Represents the :heart: emoji.
    ROCKET = "ROCKET"  # Represents the :rocket: emoji.
    EYES = "EYES"  # Represents the :eyes: emoji.


class MissingToken(Exception):
    """Could not retrieve a Github token."""


class GithubClient:
    dirname = pathlib.Path(os.path.dirname(__file__))

    def __init__(
        self,
        config: config.Config,
        github_token: str | None = None,
    ):
        """
        Keyword Arguments:
            config (config.Config): A config object to be used to get the name of the github token environment variable.
            github_token (str, optional): github personal access token.
        """
        self.config = config
        if github_token != "" or github_token is not None:
            logging.info(
                f"Reading Github token from this environment variable: {self.config.github_token_env}"
            )
            github_token = os.getenv(self.config.github_token_env)
        if github_token is None or len(github_token) == 0:
            # We can't proceed without a Github token, otherwise we'll trigger
            # an assertion failure.
            raise MissingToken("Could not retrieve the token")
        auth = github.Auth.Token(github_token)
        self.github = github.Github(auth=auth)
        self.gql = github_graphql.GithubGraphQL(token=github_token, raise_on_error=True)
        self.__repo_cache: github.Repository.Repository | None = None

    @classmethod
    def abspath(cls, p: str | pathlib.Path) -> pathlib.Path:
        return cls.dirname.joinpath(str(p))

    @property
    def gh_repo(self) -> github.Repository.Repository:
        if self.__repo_cache is None:
            self.__repo_cache = self.github.get_repo(self.config.github_repo)
        return self.__repo_cache

    def get_todays_github_issue(
        self,
        strategy: str,
        creator: str = "github-actions[bot]",
        github_repo: str | None = None,
    ) -> github.Issue.Issue | None:
        """Returns the github issue (if any) for today's snapshot that was build with the given strategy.

        If no issue was found, `None` is returned.

        Args:
            strategy (str): The build strategy to pick (e.g. "standalone", "big-merge").
            creator (str|None, optional): The author who should have created the issue. Defaults to github-actions[bot]
            repo (str|None, optional): The repo to use. This is only useful for testing purposes. Defaults to None which will result in whatever the github_repo property has.

        Raises:
            ValueError if the strategy is empty

        Returns:
            github.Issue.Issue|None: The found issue or None.
        """
        if not strategy or strategy.strip() == "":
            raise ValueError("parameter 'strategy' must not be empty")

        if github_repo is None:
            github_repo = self.config.github_repo

        # See https://docs.github.com/en/search-github/searching-on-github/searching-issues-and-pull-requests
        # label:broken_snapshot_detected
        query = f"is:issue repo:{github_repo} author:{creator} label:strategy/{strategy} {self.config.yyyymmdd} in:title"
        issues = self.github.search_issues(query)
        if issues is None:
            logging.info(f"Found no issue for today ({self.config.yyyymmdd})")
            return None

        # This is a hack: normally the PaginagedList[Issue] type handles this
        # for us but without this hack no issue being found.
        issues.get_page(0)
        if issues.totalCount > 0:
            logging.info(
                f"Found today's ({self.config.yyyymmdd}) issue: {issues[0].html_url}"
            )
            return issues[0]
        return None

    @property
    def initial_comment(self) -> str:
        llvm_release = util.get_release_for_yyyymmdd(self.config.yyyymmdd)
        llvm_git_revision = util.get_git_revision_for_yyyymmdd(self.config.yyyymmdd)
        return f"""
<p>
This issue exists to let you know that we are about to monitor the builds
of the LLVM (v{llvm_release}, <a href="https://github.com/llvm/llvm-project/commit/{llvm_git_revision}">llvm/llvm-project@ {llvm_git_revision[:7]}</a>) snapshot for <a href="{self.config.copr_monitor_url}">{self.config.yyyymmdd}</a>.
<details>
<summary>At certain intervals the CI system will update this very comment over time to reflect the progress of builds.</summary>
<dl>
<dt>Log analysis</dt>
<dd>For example if a build fails on the <code>fedora-rawhide-x86_64</code> platform,
we'll analyze the build log (if any) to identify the cause of the failure. The cause can be any of <code>{build_status.ErrorCause.list()}</code>.
For each cause we will list the packages and the relevant log excerpts.</dd>
<dt>Use of labels</dt>
<dd>Let's assume a unit test test in upstream LLVM was broken.
We will then add these labels to this issue: <code>error/test</code>, <code>build_failed_on/fedora-rawhide-x86_64</code>.
If you manually restart a build in Copr and can bring it to a successful state, we will automatically
remove the aforementioned labels.
</dd>
</dl>
</details>
</p>

{self.config.update_marker}

{self.last_updated_html()}
"""

    @classmethod
    def last_updated_html(cls) -> str:
        return f"<p><b>Last updated: {datetime.datetime.now().isoformat()}</b></p>"

    def issue_title(self, strategy: str = "", yyyymmdd: str = "") -> str:
        """Constructs the issue title we want to use"""
        if strategy == "":
            strategy = self.config.build_strategy
        if yyyymmdd == "":
            yyyymmdd = self.config.yyyymmdd
        llvm_release = util.get_release_for_yyyymmdd(yyyymmdd)
        llvm_git_revision = util.get_git_revision_for_yyyymmdd(yyyymmdd)
        return f"Snapshot for {yyyymmdd}, v{llvm_release}, {llvm_git_revision[:7]} ({strategy})"

    def create_or_get_todays_github_issue(
        self,
        creator: str = "github-actions[bot]",
    ) -> tuple[github.Issue.Issue, bool]:
        issue = self.get_todays_github_issue(
            strategy=self.config.build_strategy,
            creator=creator,
            github_repo=self.config.github_repo,
        )
        if issue is not None:
            return (issue, False)

        strategy = self.config.build_strategy
        logging.info("Creating issue for today")

        issue = self.gh_repo.create_issue(
            title=self.issue_title(), body=self.initial_comment
        )
        self.create_labels_for_strategies(labels=[strategy])

        issue.add_to_labels(f"strategy/{strategy}")
        return (issue, True)

    def get_labels(self) -> github.PaginatedList.PaginatedList[github.Label.Label]:
        """Returns the labels of a github repo.

        Returns:
            github.PaginatedList.PaginatedList: An enumerable list of github.Label.Label objects
        """
        return self.gh_repo.get_labels()

    def is_label_in_cache(self, name: str, color: str) -> bool:
        """Returns True if the label exists in the cache.

        Args:
            name (str): Name of the label to look for
            color (str): Color string of the label to look for

        Returns:
            bool: True if the label is in the cache
        """
        for label in self.get_labels():
            if label.name == name and label.color == color:
                return True
        return False

    def create_labels(
        self,
        prefix: str,
        color: str,
        labels: list[str] = [],
    ) -> list[github.Label.Label]:
        """Iterates over the given labels and creates or edits each label in the list
        with the given prefix and color."""
        if labels is None or len(labels) == 0:
            return []

        labels = list(set(labels))
        labels.sort()
        res = []
        for label in labels:
            labelname = label
            if not labelname.startswith(prefix):
                labelname = f"{prefix}{label}"
            if self.is_label_in_cache(name=labelname, color=color):
                continue
            logging.info(
                f"Creating label: repo={self.config.github_repo} name={labelname} color={color}",
            )
            try:
                res.append(self.gh_repo.create_label(color=color, name=labelname))
            except:  # noqa: E722
                self.gh_repo.get_label(name=labelname).edit(
                    name=labelname, color=color, description=""
                )
        return res

    @classmethod
    def get_label_names_on_issue(
        cls, issue: github.Issue.Issue, prefix: str
    ) -> list[str]:
        return [
            label.name for label in issue.get_labels() if label.name.startswith(prefix)
        ]

    @classmethod
    def get_error_label_names_on_issue(cls, issue: github.Issue.Issue) -> list[str]:
        return cls.get_label_names_on_issue(issue, prefix="error/")

    @classmethod
    def get_build_failed_on_names_on_issue(cls, issue: github.Issue.Issue) -> list[str]:
        return cls.get_label_names_on_issue(issue, prefix="build_failed_on/")

    def create_labels_for_error_causes(
        self,
        labels: list[str],
    ) -> list[github.Label.Label]:
        return self.create_labels(
            labels=labels,
            prefix="error/",
            color="FBCA04",
        )

    def create_labels_for_build_failed_on(
        self,
        labels: list[str],
    ) -> list[github.Label.Label]:
        return self.create_labels(
            labels=labels,
            prefix="build_failed_on/",
            color="F9D0C4",
        )

    def create_labels_for_strategies(
        self,
        labels: list[str],
    ) -> list[github.Label.Label]:
        return self.create_labels(
            labels=labels,
            prefix="strategy/",
            color="FFFFFF",
        )

    def create_labels_for_in_testing(
        self,
        labels: list[str],
    ) -> list[github.Label.Label]:
        return self.create_labels(
            labels=labels,
            prefix=self.config.label_prefix_in_testing,
            color="FEF2C0",
        )

    def create_labels_for_tested_on(
        self,
        labels: list[str],
    ) -> list[github.Label.Label]:
        return self.create_labels(
            labels=labels,
            prefix=self.config.label_prefix_tested_on,
            color="0E8A16",
        )

    def create_labels_for_tests_failed_on(
        self,
        labels: list[str],
    ) -> list[github.Label.Label]:
        return self.create_labels(
            labels=labels,
            prefix=self.config.label_prefix_tests_failed_on,
            color="D93F0B",
        )

    def create_labels_for_llvm_releases(
        self,
        labels: list[str],
    ) -> list[github.Label.Label]:
        return self.create_labels(
            labels=labels,
            prefix=self.config.label_prefix_llvm_release,
            color="2F3950",
        )

    @classmethod
    def get_comment(
        cls, issue: github.Issue.Issue, marker: str
    ) -> github.IssueComment.IssueComment | None:
        """Walks through all comments associated with the `issue` and returns the first one that has the `marker` in its body.

        Args:
            issue (github.Issue.Issue): The github issue to look for
            marker (str): The text to look for in the comment's body. (e.g. `"<!--MY MARKER-->"`)

        Returns:
            github.IssueComment.IssueComment: The comment containing the marker or `None`.
        """
        for comment in issue.get_comments():
            if marker in comment.body:
                return comment
        return None

    @classmethod
    def create_or_update_comment(
        cls, issue: github.Issue.Issue, marker: str, comment_body: str
    ) -> github.IssueComment.IssueComment:
        comment = cls.get_comment(issue=issue, marker=marker)
        if comment is None:
            return issue.create_comment(body=comment_body)
        try:
            comment.edit(body=comment_body)
        except github.GithubException as ex:
            raise ValueError(
                f"Failed to update github comment with marker {marker} and comment body: {comment_body}"
            ) from ex
        return comment

    @classmethod
    def remove_labels_safe(
        cls, issue: github.Issue.Issue, label_names_to_be_removed: list[str]
    ) -> None:
        """Removes all of the given labels from the issue.

        Args:
            issue (github.Issue.Issue): The issue from which to remove the labels
            label_names_to_be_removed (list[str]): A list of label names that shall be removed if they exist on the issue.
        """
        current_set = {label.name for label in issue.get_labels()}

        remove_set = set(label_names_to_be_removed)
        intersection = current_set.intersection(remove_set)
        for label in intersection:
            logging.info(f"Removing label '{label}' from issue: {issue.title}")
            issue.remove_from_labels(label)

    def minimize_comment_as_outdated(
        self,
        issue_comment_or_node_id: github.IssueComment.IssueComment | str,
    ) -> bool:
        """Minimizes a comment identified by the `issue_comment_or_node_id` argument with the reason `OUTDATED`.

        Args:
            issue_comment_or_node_id (str | github.IssueComment.IssueComment): The comment object or its node ID to add minimize.

        Raises:
            ValueError: If the `issue_comment_or_node_id` has a wrong type.

        Returns:
            bool: True if the comment was properly minimized.
        """
        node_id = ""
        if isinstance(issue_comment_or_node_id, github.IssueComment.IssueComment):
            node_id = issue_comment_or_node_id.raw_data["node_id"]
        elif isinstance(issue_comment_or_node_id, str):
            node_id = issue_comment_or_node_id
        else:
            raise ValueError(
                f"invalid comment object passed: {issue_comment_or_node_id}"
            )

        res = self.gql.run_from_file(
            variables={
                "classifier": "OUTDATED",
                "id": node_id,
            },
            filename=self.abspath("graphql/minimize_comment.gql"),
        )

        return bool(
            fnc.get(
                "data.minimizeComment.minimizedComment.isMinimized", res, default=False
            )
        )

    def unminimize_comment(
        self,
        issue_comment_or_node_id: github.IssueComment.IssueComment | str,
    ) -> bool:
        """Unminimizes a comment with the given `issue_comment_or_node_id`.

        Args:
            issue_comment_or_node_id (str): The comment object or its node ID to add unminimize.

        Raises:
            ValueError: If the `issue_comment_or_node_id` has a wrong type.

        Returns:
            bool: True if the comment was unminimized
        """

        node_id = ""
        if isinstance(issue_comment_or_node_id, github.IssueComment.IssueComment):
            node_id = issue_comment_or_node_id.raw_data["node_id"]
        elif isinstance(issue_comment_or_node_id, str):
            node_id = issue_comment_or_node_id
        else:
            raise ValueError(
                f"invalid comment object passed: {issue_comment_or_node_id}"
            )

        res = self.gql.run_from_file(
            variables={
                "id": node_id,
            },
            filename=self.abspath("graphql/unminimize_comment.gql"),
        )

        is_minimized = fnc.get(
            "data.unminimizeComment.unminimizedComment.isMinimized", res, default=True
        )
        return not is_minimized

    def add_comment_reaction(
        self,
        issue_comment_or_node_id: github.IssueComment.IssueComment | str,
        reaction: Reaction,
    ) -> bool:
        """Adds a reaction to a comment with the given emoji name

        Args:
            issue_comment_or_node_id (github.IssueComment.IssueComment|str): The comment object or its node ID to add reaction to.
            reaction (Reaction): The name of the reaction.

        Raises:
            ValueError: If the the `issue_comment_or_node_id` has a wrong type.

        Returns:
            bool: True if the comment reaction was added successfully.
        """
        node_id = ""
        if isinstance(issue_comment_or_node_id, github.IssueComment.IssueComment):
            node_id = issue_comment_or_node_id.raw_data["node_id"]
        elif isinstance(issue_comment_or_node_id, str):
            node_id = issue_comment_or_node_id
        else:
            raise ValueError(
                f"invalid comment object passed: {issue_comment_or_node_id}"
            )

        res = self.gql.run_from_file(
            variables={
                "comment_id": node_id,
                "reaction": reaction,
            },
            filename=self.abspath("graphql/add_comment_reaction.gql"),
        )

        actual_reaction = fnc.get(
            "data.addReaction.reaction.content", res, default=None
        )
        actual_comment_id = fnc.get("data.addReaction.subject.id", res, default=None)

        return str(actual_reaction) == str(reaction) and str(actual_comment_id) == str(
            node_id
        )

    def label_in_testing(self, chroot: str) -> str:
        return f"{self.config.label_prefix_in_testing}{chroot}"

    def label_failed_on(self, chroot: str) -> str:
        return f"{self.config.label_prefix_tests_failed_on}{chroot}"

    def label_tested_on(self, chroot: str) -> str:
        return f"{self.config.label_prefix_tested_on}{chroot}"

    def flip_test_label(
        self, issue: github.Issue.Issue, chroot: str, new_label: str | None
    ) -> None:
        """Let's you change the label on an issue for a specific chroot.

         If `new_label` is `None`, then all test labels will be removed.

        Args:
            issue (github.Issue.Issue): The issue to modify
            chroot (str): The chroot for which you want to flip the test label
            new_label (str | None): The new label or `None`.
        """
        in_testing = self.label_in_testing(chroot)
        tested_on = self.label_tested_on(chroot)
        failed_on = self.label_failed_on(chroot)

        all_states = [in_testing, tested_on, failed_on]
        existing_test_labels = [
            label.name for label in issue.get_labels() if label.name in all_states
        ]

        new_label_already_present = False
        for label in existing_test_labels:
            if label != new_label:
                issue.remove_from_labels(label)
            else:
                new_label_already_present = True

        if not new_label_already_present:
            if new_label is not None:
                issue.add_to_labels(new_label)

1	"""
2	github_util
3	"""
4
5	import datetime	1✔
6	import enum	1✔
7	import logging	1✔
8	import os	1✔
9	import pathlib	1✔
10
11	import fnc	1✔
12	import github	1✔
13	import github.GithubException	1✔
14	import github.Issue	1✔
15	import github.IssueComment	1✔
16	import github.Label	1✔
17	import github.PaginatedList	1✔
18	import github.Repository	1✔
19
20	import snapshot_manager.build_status as build_status	1✔
21	import snapshot_manager.config as config	1✔
22	import snapshot_manager.github_graphql as github_graphql	1✔
23	import snapshot_manager.util as util	1✔
24
25
26	@enum.unique	1✔
27	class Reaction(enum.StrEnum):	1✔
28	"""An enum to represent the possible comment reactions"""
29
30	THUMBS_UP = "THUMBS_UP" # Represents the :+1: emoji.	1✔
31	THUMBS_DOWN = "THUMBS_DOWN" # Represents the :-1: emoji.	1✔
32	LAUGH = "LAUGH" # Represents the :laugh: emoji.	1✔
33	HOORAY = "HOORAY" # Represents the :hooray: emoji.	1✔
34	CONFUSED = "CONFUSED" # Represents the :confused: emoji.	1✔
35	HEART = "HEART" # Represents the :heart: emoji.	1✔
36	ROCKET = "ROCKET" # Represents the :rocket: emoji.	1✔
37	EYES = "EYES" # Represents the :eyes: emoji.	1✔
38
39
40	class MissingToken(Exception):	1✔
41	"""Could not retrieve a Github token."""
42
43
44	class GithubClient:	1✔
45	dirname = pathlib.Path(os.path.dirname(__file__))	1✔
46
47	def __init__(	1✔
48	self,
49	config: config.Config,
50	github_token: str \| None = None,
51	):
52	"""
53	Keyword Arguments:
54	config (config.Config): A config object to be used to get the name of the github token environment variable.
55	github_token (str, optional): github personal access token.
56	"""
57	self.config = config	1✔
58	if github_token != "" or github_token is not None:	1✔
59	logging.info(	1✔
60	f"Reading Github token from this environment variable: {self.config.github_token_env}"
61	)
62	github_token = os.getenv(self.config.github_token_env)	1✔
63	if github_token is None or len(github_token) == 0:	1✔
64	# We can't proceed without a Github token, otherwise we'll trigger
65	# an assertion failure.
UNCOV 66	raise MissingToken("Could not retrieve the token")	×
67	auth = github.Auth.Token(github_token)	1✔
68	self.github = github.Github(auth=auth)	1✔
69	self.gql = github_graphql.GithubGraphQL(token=github_token, raise_on_error=True)	1✔
70	self.__repo_cache: github.Repository.Repository \| None = None	1✔
71
72	@classmethod	1✔
73	def abspath(cls, p: str \| pathlib.Path) -> pathlib.Path:	1✔
74	return cls.dirname.joinpath(str(p))	1✔
75
76	@property	1✔
77	def gh_repo(self) -> github.Repository.Repository:	1✔
78	if self.__repo_cache is None:	1✔
79	self.__repo_cache = self.github.get_repo(self.config.github_repo)	1✔
80	return self.__repo_cache	1✔
81
82	def get_todays_github_issue(	1✔
83	self,
84	strategy: str,
85	creator: str = "github-actions[bot]",
86	github_repo: str \| None = None,
87	) -> github.Issue.Issue \| None:
88	"""Returns the github issue (if any) for today's snapshot that was build with the given strategy.
89
90	If no issue was found, `None` is returned.
91
92	Args:
93	strategy (str): The build strategy to pick (e.g. "standalone", "big-merge").
94	creator (str\|None, optional): The author who should have created the issue. Defaults to github-actions[bot]
95	repo (str\|None, optional): The repo to use. This is only useful for testing purposes. Defaults to None which will result in whatever the github_repo property has.
96
97	Raises:
98	ValueError if the strategy is empty
99
100	Returns:
101	github.Issue.Issue\|None: The found issue or None.
102	"""
103	if not strategy or strategy.strip() == "":	1✔
104	raise ValueError("parameter 'strategy' must not be empty")	1✔
105
106	if github_repo is None:	1✔
107	github_repo = self.config.github_repo	1✔
108
109	# See https://docs.github.com/en/search-github/searching-on-github/searching-issues-and-pull-requests
110	# label:broken_snapshot_detected
111	query = f"is:issue repo:{github_repo} author:{creator} label:strategy/{strategy} {self.config.yyyymmdd} in:title"	1✔
112	issues = self.github.search_issues(query)	1✔
113	if issues is None:	1✔
114	logging.info(f"Found no issue for today ({self.config.yyyymmdd})")	1✔
115	return None	1✔
116
117	# This is a hack: normally the PaginagedList[Issue] type handles this
118	# for us but without this hack no issue being found.
119	issues.get_page(0)	1✔
120	if issues.totalCount > 0:	1✔
121	logging.info(	1✔
122	f"Found today's ({self.config.yyyymmdd}) issue: {issues[0].html_url}"
123	)
124	return issues[0]	1✔
125	return None	×
126
127	@property	1✔
128	def initial_comment(self) -> str:	1✔
129	llvm_release = util.get_release_for_yyyymmdd(self.config.yyyymmdd)	1✔
130	llvm_git_revision = util.get_git_revision_for_yyyymmdd(self.config.yyyymmdd)	1✔
131	return f"""	1✔
132	<p>
133	This issue exists to let you know that we are about to monitor the builds
134	of the LLVM (v{llvm_release}, <a href="https://github.com/llvm/llvm-project/commit/{llvm_git_revision}">llvm/llvm-project@ {llvm_git_revision[:7]}</a>) snapshot for <a href="{self.config.copr_monitor_url}">{self.config.yyyymmdd}</a>.
135	<details>
136	<summary>At certain intervals the CI system will update this very comment over time to reflect the progress of builds.</summary>
137	<dl>
138	<dt>Log analysis</dt>
139	<dd>For example if a build fails on the <code>fedora-rawhide-x86_64</code> platform,
140	we'll analyze the build log (if any) to identify the cause of the failure. The cause can be any of <code>{build_status.ErrorCause.list()}</code>.
141	For each cause we will list the packages and the relevant log excerpts.</dd>
142	<dt>Use of labels</dt>
143	<dd>Let's assume a unit test test in upstream LLVM was broken.
144	We will then add these labels to this issue: <code>error/test</code>, <code>build_failed_on/fedora-rawhide-x86_64</code>.
145	If you manually restart a build in Copr and can bring it to a successful state, we will automatically
146	remove the aforementioned labels.
147	</dd>
148	</dl>
149	</details>
150	</p>
151
152	{self.config.update_marker}
153
154	{self.last_updated_html()}
155	"""
156
157	@classmethod	1✔
158	def last_updated_html(cls) -> str:	1✔
159	return f"<p><b>Last updated: {datetime.datetime.now().isoformat()}</b></p>"	1✔
160
161	def issue_title(self, strategy: str = "", yyyymmdd: str = "") -> str:	1✔
162	"""Constructs the issue title we want to use"""
163	if strategy == "":	1✔
164	strategy = self.config.build_strategy	1✔
165	if yyyymmdd == "":	1✔
166	yyyymmdd = self.config.yyyymmdd	1✔
167	llvm_release = util.get_release_for_yyyymmdd(yyyymmdd)	1✔
168	llvm_git_revision = util.get_git_revision_for_yyyymmdd(yyyymmdd)	1✔
169	return f"Snapshot for {yyyymmdd}, v{llvm_release}, {llvm_git_revision[:7]} ({strategy})"	1✔
170
171	def create_or_get_todays_github_issue(	1✔
172	self,
173	creator: str = "github-actions[bot]",
174	) -> tuple[github.Issue.Issue, bool]:
175	issue = self.get_todays_github_issue(	1✔
176	strategy=self.config.build_strategy,
177	creator=creator,
178	github_repo=self.config.github_repo,
179	)
180	if issue is not None:	1✔
181	return (issue, False)	1✔
182
183	strategy = self.config.build_strategy	1✔
184	logging.info("Creating issue for today")	1✔
185
186	issue = self.gh_repo.create_issue(	1✔
187	title=self.issue_title(), body=self.initial_comment
188	)
189	self.create_labels_for_strategies(labels=[strategy])	1✔
190
191	issue.add_to_labels(f"strategy/{strategy}")	1✔
192	return (issue, True)	1✔
193
194	def get_labels(self) -> github.PaginatedList.PaginatedList[github.Label.Label]:	1✔
195	"""Returns the labels of a github repo.
196
197	Returns:
198	github.PaginatedList.PaginatedList: An enumerable list of github.Label.Label objects
199	"""
NEW 200	return self.gh_repo.get_labels()	×
201
202	def is_label_in_cache(self, name: str, color: str) -> bool:	1✔
203	"""Returns True if the label exists in the cache.
204
205	Args:
206	name (str): Name of the label to look for
207	color (str): Color string of the label to look for
208
209	Returns:
210	bool: True if the label is in the cache
211	"""
212	for label in self.get_labels():	1✔
213	if label.name == name and label.color == color:	1✔
214	return True	1✔
215	return False	1✔
216
217	def create_labels(	1✔
218	self,
219	prefix: str,
220	color: str,
221	labels: list[str] = [],
222	) -> list[github.Label.Label]:
223	"""Iterates over the given labels and creates or edits each label in the list
224	with the given prefix and color."""
225	if labels is None or len(labels) == 0:	1✔
226	return []	1✔
227
228	labels = list(set(labels))	1✔
229	labels.sort()	1✔
230	res = []	1✔
231	for label in labels:	1✔
232	labelname = label	1✔
233	if not labelname.startswith(prefix):	1✔
234	labelname = f"{prefix}{label}"	1✔
235	if self.is_label_in_cache(name=labelname, color=color):	1✔
236	continue	1✔
237	logging.info(	1✔
238	f"Creating label: repo={self.config.github_repo} name={labelname} color={color}",
239	)
240	try:	1✔
241	res.append(self.gh_repo.create_label(color=color, name=labelname))	1✔
242	except: # noqa: E722	1✔
243	self.gh_repo.get_label(name=labelname).edit(	1✔
244	name=labelname, color=color, description=""
245	)
246	return res	1✔
247
248	@classmethod	1✔
249	def get_label_names_on_issue(	1✔
250	cls, issue: github.Issue.Issue, prefix: str
251	) -> list[str]:
252	return [	1✔
253	label.name for label in issue.get_labels() if label.name.startswith(prefix)
254	]
255
256	@classmethod	1✔
257	def get_error_label_names_on_issue(cls, issue: github.Issue.Issue) -> list[str]:	1✔
258	return cls.get_label_names_on_issue(issue, prefix="error/")	1✔
259
260	@classmethod	1✔
261	def get_build_failed_on_names_on_issue(cls, issue: github.Issue.Issue) -> list[str]:	1✔
262	return cls.get_label_names_on_issue(issue, prefix="build_failed_on/")	1✔
263
264	def create_labels_for_error_causes(	1✔
265	self,
266	labels: list[str],
267	) -> list[github.Label.Label]:
268	return self.create_labels(	1✔
269	labels=labels,
270	prefix="error/",
271	color="FBCA04",
272	)
273
274	def create_labels_for_build_failed_on(	1✔
275	self,
276	labels: list[str],
277	) -> list[github.Label.Label]:
278	return self.create_labels(	1✔
279	labels=labels,
280	prefix="build_failed_on/",
281	color="F9D0C4",
282	)
283
284	def create_labels_for_strategies(	1✔
285	self,
286	labels: list[str],
287	) -> list[github.Label.Label]:
288	return self.create_labels(	1✔
289	labels=labels,
290	prefix="strategy/",
291	color="FFFFFF",
292	)
293
294	def create_labels_for_in_testing(	1✔
295	self,
296	labels: list[str],
297	) -> list[github.Label.Label]:
298	return self.create_labels(	1✔
299	labels=labels,
300	prefix=self.config.label_prefix_in_testing,
301	color="FEF2C0",
302	)
303
304	def create_labels_for_tested_on(	1✔
305	self,
306	labels: list[str],
307	) -> list[github.Label.Label]:
308	return self.create_labels(	1✔
309	labels=labels,
310	prefix=self.config.label_prefix_tested_on,
311	color="0E8A16",
312	)
313
314	def create_labels_for_tests_failed_on(	1✔
315	self,
316	labels: list[str],
317	) -> list[github.Label.Label]:
318	return self.create_labels(	1✔
319	labels=labels,
320	prefix=self.config.label_prefix_tests_failed_on,
321	color="D93F0B",
322	)
323
324	def create_labels_for_llvm_releases(	1✔
325	self,
326	labels: list[str],
327	) -> list[github.Label.Label]:
328	return self.create_labels(	1✔
329	labels=labels,
330	prefix=self.config.label_prefix_llvm_release,
331	color="2F3950",
332	)
333
334	@classmethod	1✔
335	def get_comment(	1✔
336	cls, issue: github.Issue.Issue, marker: str
337	) -> github.IssueComment.IssueComment \| None:
338	"""Walks through all comments associated with the `issue` and returns the first one that has the `marker` in its body.
339
340	Args:
341	issue (github.Issue.Issue): The github issue to look for
342	marker (str): The text to look for in the comment's body. (e.g. `"<!--MY MARKER-->"`)
343
344	Returns:
345	github.IssueComment.IssueComment: The comment containing the marker or `None`.
346	"""
347	for comment in issue.get_comments():	1✔
348	if marker in comment.body:	1✔
349	return comment	1✔
350	return None	1✔
351
352	@classmethod	1✔
353	def create_or_update_comment(	1✔
354	cls, issue: github.Issue.Issue, marker: str, comment_body: str
355	) -> github.IssueComment.IssueComment:
356	comment = cls.get_comment(issue=issue, marker=marker)	1✔
357	if comment is None:	1✔
358	return issue.create_comment(body=comment_body)	1✔
359	try:	1✔
360	comment.edit(body=comment_body)	1✔
361	except github.GithubException as ex:	1✔
362	raise ValueError(	1✔
363	f"Failed to update github comment with marker {marker} and comment body: {comment_body}"
364	) from ex
365	return comment	1✔
366
367	@classmethod	1✔
368	def remove_labels_safe(	1✔
369	cls, issue: github.Issue.Issue, label_names_to_be_removed: list[str]
370	) -> None:
371	"""Removes all of the given labels from the issue.
372
373	Args:
374	issue (github.Issue.Issue): The issue from which to remove the labels
375	label_names_to_be_removed (list[str]): A list of label names that shall be removed if they exist on the issue.
376	"""
377	current_set = {label.name for label in issue.get_labels()}	1✔
378
379	remove_set = set(label_names_to_be_removed)	1✔
380	intersection = current_set.intersection(remove_set)	1✔
381	for label in intersection:	1✔
382	logging.info(f"Removing label '{label}' from issue: {issue.title}")	1✔
383	issue.remove_from_labels(label)	1✔
384
385	def minimize_comment_as_outdated(	1✔
386	self,
387	issue_comment_or_node_id: github.IssueComment.IssueComment \| str,
388	) -> bool:
389	"""Minimizes a comment identified by the `issue_comment_or_node_id` argument with the reason `OUTDATED`.
390
391	Args:
392	issue_comment_or_node_id (str \| github.IssueComment.IssueComment): The comment object or its node ID to add minimize.
393
394	Raises:
395	ValueError: If the `issue_comment_or_node_id` has a wrong type.
396
397	Returns:
398	bool: True if the comment was properly minimized.
399	"""
400	node_id = ""	1✔
401	if isinstance(issue_comment_or_node_id, github.IssueComment.IssueComment):	1✔
402	node_id = issue_comment_or_node_id.raw_data["node_id"]	1✔
403	elif isinstance(issue_comment_or_node_id, str):	1✔
404	node_id = issue_comment_or_node_id	1✔
405	else:
406	raise ValueError(	1✔
407	f"invalid comment object passed: {issue_comment_or_node_id}"
408	)
409
410	res = self.gql.run_from_file(	1✔
411	variables={
412	"classifier": "OUTDATED",
413	"id": node_id,
414	},
415	filename=self.abspath("graphql/minimize_comment.gql"),
416	)
417
418	return bool(	1✔
419	fnc.get(
420	"data.minimizeComment.minimizedComment.isMinimized", res, default=False
421	)
422	)
423
424	def unminimize_comment(	1✔
425	self,
426	issue_comment_or_node_id: github.IssueComment.IssueComment \| str,
427	) -> bool:
428	"""Unminimizes a comment with the given `issue_comment_or_node_id`.
429
430	Args:
431	issue_comment_or_node_id (str): The comment object or its node ID to add unminimize.
432
433	Raises:
434	ValueError: If the `issue_comment_or_node_id` has a wrong type.
435
436	Returns:
437	bool: True if the comment was unminimized
438	"""
439
440	node_id = ""	1✔
441	if isinstance(issue_comment_or_node_id, github.IssueComment.IssueComment):	1✔
442	node_id = issue_comment_or_node_id.raw_data["node_id"]	1✔
443	elif isinstance(issue_comment_or_node_id, str):	1✔
444	node_id = issue_comment_or_node_id	1✔
445	else:
446	raise ValueError(	1✔
447	f"invalid comment object passed: {issue_comment_or_node_id}"
448	)
449
450	res = self.gql.run_from_file(	1✔
451	variables={
452	"id": node_id,
453	},
454	filename=self.abspath("graphql/unminimize_comment.gql"),
455	)
456
457	is_minimized = fnc.get(	1✔
458	"data.unminimizeComment.unminimizedComment.isMinimized", res, default=True
459	)
460	return not is_minimized	1✔
461
462	def add_comment_reaction(	1✔
463	self,
464	issue_comment_or_node_id: github.IssueComment.IssueComment \| str,
465	reaction: Reaction,
466	) -> bool:
467	"""Adds a reaction to a comment with the given emoji name
468
469	Args:
470	issue_comment_or_node_id (github.IssueComment.IssueComment\|str): The comment object or its node ID to add reaction to.
471	reaction (Reaction): The name of the reaction.
472
473	Raises:
474	ValueError: If the the `issue_comment_or_node_id` has a wrong type.
475
476	Returns:
477	bool: True if the comment reaction was added successfully.
478	"""
479	node_id = ""	1✔
480	if isinstance(issue_comment_or_node_id, github.IssueComment.IssueComment):	1✔
481	node_id = issue_comment_or_node_id.raw_data["node_id"]	1✔
482	elif isinstance(issue_comment_or_node_id, str):	1✔
483	node_id = issue_comment_or_node_id	1✔
484	else:
485	raise ValueError(	1✔
486	f"invalid comment object passed: {issue_comment_or_node_id}"
487	)
488
489	res = self.gql.run_from_file(	1✔
490	variables={
491	"comment_id": node_id,
492	"reaction": reaction,
493	},
494	filename=self.abspath("graphql/add_comment_reaction.gql"),
495	)
496
497	actual_reaction = fnc.get(	1✔
498	"data.addReaction.reaction.content", res, default=None
499	)
500	actual_comment_id = fnc.get("data.addReaction.subject.id", res, default=None)	1✔
501
502	return str(actual_reaction) == str(reaction) and str(actual_comment_id) == str(	1✔
503	node_id
504	)
505
506	def label_in_testing(self, chroot: str) -> str:	1✔
507	return f"{self.config.label_prefix_in_testing}{chroot}"	1✔
508
509	def label_failed_on(self, chroot: str) -> str:	1✔
510	return f"{self.config.label_prefix_tests_failed_on}{chroot}"	1✔
511
512	def label_tested_on(self, chroot: str) -> str:	1✔
513	return f"{self.config.label_prefix_tested_on}{chroot}"	1✔
514
515	def flip_test_label(	1✔
516	self, issue: github.Issue.Issue, chroot: str, new_label: str \| None
517	) -> None:
518	"""Let's you change the label on an issue for a specific chroot.
519
520	If `new_label` is `None`, then all test labels will be removed.
521
522	Args:
523	issue (github.Issue.Issue): The issue to modify
524	chroot (str): The chroot for which you want to flip the test label
525	new_label (str \| None): The new label or `None`.
526	"""
527	in_testing = self.label_in_testing(chroot)	1✔
528	tested_on = self.label_tested_on(chroot)	1✔
529	failed_on = self.label_failed_on(chroot)	1✔
530
531	all_states = [in_testing, tested_on, failed_on]	1✔
532	existing_test_labels = [	1✔
533	label.name for label in issue.get_labels() if label.name in all_states
534	]
535
536	new_label_already_present = False	1✔
537	for label in existing_test_labels:	1✔
538	if label != new_label:	1✔
539	issue.remove_from_labels(label)	1✔
540	else:
541	new_label_already_present = True	1✔
542
543	if not new_label_already_present:	1✔
544	if new_label is not None:	1✔
545	issue.add_to_labels(new_label)	1✔

fedora-llvm-team / llvm-snapshots / 14795010914

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