13865613237

Committed 14 Mar 2025 09:24PM UTC coverage: 60.078% (-0.02%) from 60.1%

Build # 13865613237

Build Type

push

github

Committed by

kwk

Commit Message

fix github search_issues outcome

This is a hack: normally the `PaginagedList[Issue]` type handles this
for us but without this hack no issue being found.

Run Details

3 of 5 new or added lines in 1 file covered. (60.0%)

3 existing lines in 2 files now uncovered.

1082 of 1801 relevant lines covered (60.08%)

0.6 hits per line

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

98.25

/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 MissingToken(Exception):
    """Could not retrieve a Github token."""


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

    def __init__(self, config: config.Config, github_token: str = None, **kwargs):
        """
        Keyword Arguments:
            github_token (str, optional): github personal access token.
        """
        self.config = config
        if github_token is None:
            logging.info(
                f"Reading Github token from this environment variable: {self.config.github_token_env}"
            )
            github_token = os.getenv(self.config.github_token_env)
        if github_token is None or len(github_token) == 0:
            # We can't proceed without a Github token, otherwise we'll trigger
            # an assertion failure.
            raise MissingToken("Could not retrieve the token")
        auth = github.Auth.Token(github_token)
        self.github = github.Github(auth=auth)
        self.gql = github_graphql.GithubGraphQL(token=github_token, raise_on_error=True)
        self._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 None:
            logging.info(f"Found no issue for today ({self.config.yyyymmdd})")
            return None

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

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

{self.config.update_marker}

{self.last_updated_html()}
"""

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

    def issue_title(self, strategy: str = 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
        logging.info("Creating issue for today")

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

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

    @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

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

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

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

    def create_labels_for_error_causes(
        self, labels: list[str], **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_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,
        )

    @classmethod
    def get_comment(
        cls, 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

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

    @classmethod
    def remove_labels_safe(
        cls, issue: github.Issue.Issue, label_names_to_be_removed: list[str]
    ):
        """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:
            if new_label is not None:
                issue.add_to_labels(new_label)

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

fedora-llvm-team / llvm-snapshots / 13865613237

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