9659359945

Committed 25 Jun 2024 08:48AM UTC coverage: 67.834% (-1.8%) from 69.585%

Build # 9659359945

Build Type

push

github

Committed by

kwk

Commit Message

Bump urllib3 from 1.26.18 to 1.26.19

Bumps [urllib3](https://github.com/urllib3/urllib3) from 1.26.18 to 1.26.19.
- [Release notes](https://github.com/urllib3/urllib3/releases)
- [Changelog](https://github.com/urllib3/urllib3/blob/1.26.19/CHANGES.rst)
- [Commits](https://github.com/urllib3/urllib3/compare/1.26.18...1.26.19)

---
updated-dependencies:
- dependency-name: urllib3
  dependency-type: indirect
...

Signed-off-by: dependabot[bot] <support@github.com>

Run Details

1046 of 1542 relevant lines covered (67.83%)

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

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

fedora-llvm-team / llvm-snapshots / 9659359945

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