10479306856

Committed 20 Aug 2024 09:14PM UTC coverage: 67.503% (-1.7%) from 69.241%

Build # 10479306856

Build Type

push

github

Committed by

kwk

Commit Message

Build PGO for all platforms

Run Details

1049 of 1554 relevant lines covered (67.5%)

0.68 hits per line

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

74.55

/snapshot_manager/snapshot_manager/github_util.py

"""
github_util
"""

import datetime
import enum
import logging
import os
import pathlib
import typing

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 GithubClient:
    dirname = pathlib.Path(os.path.dirname(__file__))

    def __init__(self, config: config.Config, github_token: str = None, **kwargs):
        """
        Keyword Arguments:
            github_token (str, optional): github personal access token.
        """
        self.config = config
        if github_token is None:
            github_token = os.getenv(self.config.github_token_env)
        self.github = github.Github(login_or_token=github_token)
        self.gql = github_graphql.GithubGraphQL(token=github_token, raise_on_error=True)
        self.__label_cache = None
        self.__repo_cache = None

    @classmethod
    def abspath(cls, p: tuple[str, pathlib.Path]) -> pathlib.Path:
        return cls.dirname.joinpath(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:
            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 not None and issues.totalCount > 0:
            logging.info(f"Found today's issue: {issues[0].html_url}")
            return issues[0]
        logging.info("Found no issue for today")
        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 of the <code>llvm</code> project 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>, <code>project/llvm</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 = None, yyyymmdd: str = None) -> str:
        """Constructs the issue title we want to use"""
        if strategy is None:
            strategy = self.config.build_strategy
        if yyyymmdd is None:
            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
        repo = self.gh_repo
        logging.info("Creating issue for today")

        issue = 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)

    @property
    def label_cache(self, refresh: bool = False) -> github.PaginatedList.PaginatedList:
        """Will query the labels of a github repo only once and return it afterwards.

        Args:
            refresh (bool, optional): The cache will be emptied. Defaults to False.

        Returns:
            github.PaginatedList.PaginatedList: An enumerable list of github.Label.Label objects
        """
        if self.__label_cache is None or refresh:
            self.__label_cache = self.gh_repo.get_labels()
        return self.__label_cache

    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.label_cache:
            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 None

        labels = set(labels)
        labels = list(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:
                self.gh_repo.get_label(name=labelname).edit(
                    name=labelname, color=color, description=""
                )
        return res

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

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

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

    def get_project_label_names_on_issue(self, issue: github.Issue.Issue) -> list[str]:
        return self.get_label_names_on_issue(issue, prefix="project/")

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

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

    def create_labels_for_projects(
        self, labels: list[str], **kw_args
    ) -> list[github.Label.Label]:
        return self.create_labels(
            labels=labels, prefix="project/", color="BFDADC", **kw_args
        )

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

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

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

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

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

    def get_comment(
        self, issue: github.Issue.Issue, marker: str
    ) -> github.IssueComment.IssueComment:
        """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

    def create_or_update_comment(
        self, issue: github.Issue.Issue, marker: str, comment_body: str
    ) -> github.IssueComment.IssueComment:
        comment = self.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.GithubException as ex:
            raise ValueError(
                f"Failed to update github comment with marker {marker} and comment body: {comment_body}"
            ) from ex
        return comment

    def remove_labels_safe(
        self, issue: github.Issue.Issue, label_names_to_be_removed: list[str]
    ):
        """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)

    @typing.overload
    def minimize_comment_as_outdated(
        self, comment: github.IssueComment.IssueComment
    ) -> bool: ...

    @typing.overload
    def minimize_comment_as_outdated(self, node_id: str) -> bool: ...

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

        Args:
            object (str | github.IssueComment.IssueComment): The comment to minimize

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

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

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

    @typing.overload
    def unminimize_comment(self, comment: github.IssueComment.IssueComment) -> bool: ...

    @typing.overload
    def unminimize_comment(self, node_id: str) -> bool: ...

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

        Args:
            node_id (str): A comment's `node_id`.

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

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

        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

    @typing.overload
    def add_comment_reaction(
        self, comment: github.IssueComment.IssueComment, reaction: Reaction
    ) -> bool: ...

    @typing.overload
    def add_comment_reaction(self, node_id: str, reaction: Reaction) -> bool: ...

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

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

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

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

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

fedora-llvm-team / llvm-snapshots / 10479306856

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