13027349381

Committed 29 Jan 2025 08:21AM UTC coverage: 38.322% (-23.1%) from 61.451%

Build # 13027349381

Build Type

push

github

Committed by

kwk

Commit Message

[debug] Add logging to github client setup

Run Details

1 of 1 new or added line in 1 file covered. (100.0%)

58 existing lines in 1 file now uncovered.

9759 of 25466 relevant lines covered (38.32%)

0.38 hits per line

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

43.95

/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:
            logging.info(
                f"Reading Github token from this environment variable: {self.config.github_token_env}"
            )
            github_token = os.getenv(self.config.github_token_env)
        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 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"""
<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
        repo = self.gh_repo
        logging.info("Creating issue for today")

        issue = 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

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

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

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

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

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

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

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

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

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

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

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

    def create_labels_for_llvm_releases(
        self, labels: list[str], **kw_args
    ) -> list[github.Label.Label]:
        return self.create_labels(
            labels=labels,
            prefix=self.config.label_prefix_llvm_release,
            color="2F3950",
            *kw_args,
        )

    def get_comment(
        self, issue: github.Issue.Issue, marker: str
    ) -> github.IssueComment.IssueComment:
        """Walks through all comments associated with the `issue` and returns the first one that has the `marker` in its body.

        Args:
            issue (github.Issue.Issue): The github issue to look for
            marker (str): The text to look for in the comment's body. (e.g. `"<!--MY MARKER-->"`)

        Returns:
            github.IssueComment.IssueComment: The comment containing the marker or `None`.
        """
        for comment in issue.get_comments():
            if marker in comment.body:
                return comment
        return None

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

    def remove_labels_safe(
        self, issue: github.Issue.Issue, label_names_to_be_removed: list[str]
    ):
        """Removes all of the given labels from the issue.

        Args:
            issue (github.Issue.Issue): The issue from which to remove the labels
            label_names_to_be_removed (list[str]): A list of label names that shall be removed if they exist on the issue.
        """
        current_set = {label.name for label in issue.get_labels()}

        remove_set = set(label_names_to_be_removed)
        intersection = current_set.intersection(remove_set)
        for label in intersection:
            logging.info(f"Removing label '{label}' from issue: {issue.title}")
            issue.remove_from_labels(label)

    @typing.overload
    def minimize_comment_as_outdated(
        self, comment: github.IssueComment.IssueComment
    ) -> bool: ...

    @typing.overload
    def minimize_comment_as_outdated(self, node_id: str) -> bool: ...

    def minimize_comment_as_outdated(
        self,
        object: str | github.IssueComment.IssueComment,
    ) -> bool:
        """Minimizes a comment identified by the `object` argument with the reason `OUTDATED`.

        Args:
            object (str | github.IssueComment.IssueComment): The comment to minimize

        Raises:
            ValueError: If the `object` has a wrong type.

        Returns:
            bool: True if the comment was properly minimized.
        """
        node_id = ""
        if isinstance(object, github.IssueComment.IssueComment):
            node_id = object.raw_data["node_id"]
        elif isinstance(object, str):
            node_id = object
        else:
            raise ValueError(f"invalid comment object passed: {object}")

        res = self.gql.run_from_file(
            variables={
                "classifier": "OUTDATED",
                "id": node_id,
            },
            filename=self.abspath("graphql/minimize_comment.gql"),
        )

        return bool(
            fnc.get(
                "data.minimizeComment.minimizedComment.isMinimized", res, default=False
            )
        )

    @typing.overload
    def unminimize_comment(self, comment: github.IssueComment.IssueComment) -> bool: ...

    @typing.overload
    def unminimize_comment(self, node_id: str) -> bool: ...

    def unminimize_comment(
        self,
        object: str | github.IssueComment.IssueComment,
    ) -> bool:
        """Unminimizes a comment with the given `node_id`.

        Args:
            node_id (str): A comment's `node_id`.

        Returns:
            bool: True if the comment was unminimized
        """

        node_id = ""
        if isinstance(object, github.IssueComment.IssueComment):
            node_id = object.raw_data["node_id"]
        elif isinstance(object, str):
            node_id = object
        else:
            raise ValueError(f"invalid comment object passed: {object}")

        res = self.gql.run_from_file(
            variables={
                "id": node_id,
            },
            filename=self.abspath("graphql/unminimize_comment.gql"),
        )

        is_minimized = fnc.get(
            "data.unminimizeComment.unminimizedComment.isMinimized", res, default=True
        )
        return not is_minimized

    @typing.overload
    def add_comment_reaction(
        self, comment: github.IssueComment.IssueComment, reaction: Reaction
    ) -> bool: ...

    @typing.overload
    def add_comment_reaction(self, node_id: str, reaction: Reaction) -> bool: ...

    def add_comment_reaction(
        self,
        object: str | github.IssueComment.IssueComment,
        reaction: Reaction,
    ) -> bool:
        """Adds a reaction to a comment with the given emoji name

        Args:
            object (str | github.IssueComment.IssueComment): The comment object or node ID to add reaction to.
            reaction (Reaction): The name of the reaction.

        Raises:
            ValueError: If the the `object` has a wrong type.

        Returns:
            bool: True if the comment reaction was added successfully.
        """
        node_id = ""
        if isinstance(object, github.IssueComment.IssueComment):
            node_id = object.raw_data["node_id"]
        elif isinstance(object, str):
            node_id = object
        else:
            raise ValueError(f"invalid comment object passed: {object}")

        res = self.gql.run_from_file(
            variables={
                "comment_id": node_id,
                "reaction": reaction,
            },
            filename=self.abspath("graphql/add_comment_reaction.gql"),
        )

        actual_reaction = fnc.get(
            "data.addReaction.reaction.content", res, default=None
        )
        actual_comment_id = fnc.get("data.addReaction.subject.id", res, default=None)

        return actual_reaction == str(reaction) and actual_comment_id == node_id

    def label_in_testing(self, chroot: str) -> str:
        return f"{self.config.label_prefix_in_testing}{chroot}"

    def label_failed_on(self, chroot: str) -> str:
        return f"{self.config.label_prefix_tests_failed_on}{chroot}"

    def label_tested_on(self, chroot: str) -> str:
        return f"{self.config.label_prefix_tested_on}{chroot}"

    def flip_test_label(
        self, issue: github.Issue.Issue, chroot: str, new_label: str | None
    ):
        """Let's you change the label on an issue for a specific chroot.

         If `new_label` is `None`, then all test labels will be removed.

        Args:
            issue (github.Issue.Issue): The issue to modify
            chroot (str): The chroot for which you want to flip the test label
            new_label (str | None): The new label or `None`.
        """
        in_testing = self.label_in_testing(chroot)
        tested_on = self.label_tested_on(chroot)
        failed_on = self.label_failed_on(chroot)

        all_states = [in_testing, tested_on, failed_on]
        existing_test_labels = [
            label.name for label in issue.get_labels() if label.name in all_states
        ]

        new_label_already_present = False
        for label in existing_test_labels:
            if label != new_label:
                issue.remove_from_labels(label)
            else:
                new_label_already_present = True

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

fedora-llvm-team / llvm-snapshots / 13027349381

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