8757373822

Committed 19 Apr 2024 05:26PM UTC coverage: 67.204% (-2.1%) from 69.287%

Build # 8757373822

Build Type

Pull #425

github

Committed by

kwk

Commit Message

main: add retest subcommand

See #416

Pull Request Pull Request #425: Retest

Run Details

45 of 126 new or added lines in 5 files covered. (35.71%)

2 existing lines in 1 file now uncovered.

1000 of 1488 relevant lines covered (67.2%)

0.67 hits per line

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

77.1

/snapshot_manager/snapshot_manager/github_util.py

"""
github_util
"""

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

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

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


@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.

    @classmethod
    def all_reactions(cls) -> list["Reaction"]:
        return [s for s in Reaction]


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:
        return f"""
Hello @{self.config.maintainer_handle}!

<p>
This issue exists to let you know that we are about to monitor the builds
of the LLVM 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>arch/x86_64</code>, <code>os/fedora-rawhide</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 create_or_get_todays_github_issue(
        self,
        maintainer_handle: str,
        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(
            assignee=maintainer_handle,
            title=f"Snapshot build for {self.config.yyyymmdd} ({strategy})",
            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_os_label_names_on_issue(self, issue: github.Issue.Issue) -> list[str]:
        return self.get_label_names_on_issue(issue, prefix="os/")

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

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

fedora-llvm-team / llvm-snapshots / 8757373822

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