13469286298

Committed 22 Feb 2025 03:42AM UTC coverage: 39.547% (-0.06%) from 39.607%

Build # 13469286298

Build Type

Pull #1098

github

Committed by

web-flow

Commit Message

Merge a62125842 into 00e77f447

Pull Request Pull Request #1098: rebuilder.py: Remove debugging code that was added accidentally

Run Details

10188 of 25762 relevant lines covered (39.55%)

0.4 hits per line

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

99.14

/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 not None and issues.totalCount > 0:
            logging.info(
                f"Found today's ({self.config.yyyymmdd}) issue: {issues[0].html_url}"
            )
            return issues[0]
        logging.info(f"Found no issue for today ({self.config.yyyymmdd})")
        return None

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

{self.config.update_marker}

{self.last_updated_html()}
"""

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

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

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

        strategy = self.config.build_strategy
        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/")

    @classmethod
    def get_project_label_names_on_issue(cls, issue: github.Issue.Issue) -> list[str]:
        return cls.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,
        )

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

fedora-llvm-team / llvm-snapshots / 13469286298

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