13137779807

Committed 04 Feb 2025 02:32PM UTC coverage: 38.173% (-0.1%) from 38.322%

Build # 13137779807

Build Type

Pull #1063

github

Committed by

web-flow

Commit Message

Merge 0af3ff7dd into f5421dcb3

Pull Request Pull Request #1063: Skip tests that require access to the token

Run Details

15 of 17 new or added lines in 3 files covered. (88.24%)

46 existing lines in 5 files now uncovered.

9726 of 25479 relevant lines covered (38.17%)

0.38 hits per line

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

38.5

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

fedora-llvm-team / llvm-snapshots / 13137779807

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