9188338383

Committed 22 May 2024 08:55AM UTC coverage: 67.632% (-1.6%) from 69.211%

Build # 9188338383

Build Type

push

github

Committed by

kwk

Commit Message

Add a pre-commit workflow

This runs all the checks we run locally for pre-commit but on github.
Thereby we also run tmt linting.

Fixes #77

Run Details

1028 of 1520 relevant lines covered (67.63%)

0.68 hits per line

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

75.23

/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>arch/x86_64</code>, <code>os/fedora-rawhide</code>, <code>project/llvm</code>.
If you manually restart a build in Copr and can bring it to a successful state, we will automatically
remove the aforementioned labels.
</dd>
</dl>
</details>
</p>

{self.config.update_marker}

{self.last_updated_html()}
"""

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

    @property
    def issue_title(self) -> str:
        """Constructs the issue title we want to use"""
        strategy = self.config.build_strategy
        llvm_release = util.get_release_for_yyyymmdd(self.config.yyyymmdd)
        llvm_git_revision = util.get_git_revision_for_yyyymmdd(self.config.yyyymmdd)
        return f"Snapshot for {self.config.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_os_label_names_on_issue(self, issue: github.Issue.Issue) -> list[str]:
        return self.get_label_names_on_issue(issue, prefix="os/")

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

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

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

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

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

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

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

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

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

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

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

fedora-llvm-team / llvm-snapshots / 9188338383

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