13137950566

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

Build # 13137950566

Build Type

Pull #1063

github

Committed by

web-flow

Commit Message

Merge c770c07ef into f5421dcb3

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

Run Details

16 of 16 new or added lines in 3 files covered. (100.0%)

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:
            # We can't proceed without a Github token, otherwise we'll trigger
            # an assertion failure.
            raise MissingToken("Could not retrieve the token")
        auth = github.Auth.Token(github_token)
        self.github = github.Github(auth=auth)
        self.gql = github_graphql.GithubGraphQL(token=github_token, raise_on_error=True)
        self.__label_cache = None
        self.__repo_cache = None

    @classmethod
    def abspath(cls, p: tuple[str, pathlib.Path]) -> pathlib.Path:
        return cls.dirname.joinpath(p)

    @property
    def gh_repo(self) -> github.Repository.Repository:
        if self.__repo_cache is None:
            self.__repo_cache = self.github.get_repo(self.config.github_repo)
        return self.__repo_cache

    def get_todays_github_issue(
        self,
        strategy: str,
        creator: str = "github-actions[bot]",
        github_repo: str | None = None,
    ) -> github.Issue.Issue | None:
        """Returns the github issue (if any) for today's snapshot that was build with the given strategy.

        If no issue was found, `None` is returned.

        Args:
            strategy (str): The build strategy to pick (e.g. "standalone", "big-merge").
            creator (str|None, optional): The author who should have created the issue. Defaults to github-actions[bot]
            repo (str|None, optional): The repo to use. This is only useful for testing purposes. Defaults to None which will result in whatever the github_repo property has.

        Raises:
            ValueError if the strategy is empty

        Returns:
            github.Issue.Issue|None: The found issue or None.
        """
        if not strategy:
            raise ValueError("parameter 'strategy' must not be empty")

        if github_repo is None:
            github_repo = self.config.github_repo

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

fedora-llvm-team / llvm-snapshots / 13137950566

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