Module ogr.services.pagure

class PagureIssue(BaseIssue):
    project: "ogr_pagure.PagureProject"

    def __init__(self, raw_issue, project):
        super().__init__(raw_issue, project)
        self.__dirty = False

    def __update(self):
        if self.__dirty:
            self._raw_issue = self.project._call_project_api("issue", str(self.id))
            self.__dirty = False

    @property
    def title(self) -> str:
        self.__update()
        return self._raw_issue["title"]

    @title.setter
    def title(self, new_title: str) -> None:
        self.__update_info(title=new_title)

    @property
    def private(self) -> bool:
        self.__update()
        return self._raw_issue["private"]

    @property
    def id(self) -> int:
        return self._raw_issue["id"]

    @property
    def status(self) -> IssueStatus:
        self.__update()
        return IssueStatus[self._raw_issue["status"].lower()]

    @property
    def url(self) -> str:
        return self.project._get_project_url(
            "issue",
            str(self.id),
            add_api_endpoint_part=False,
        )

    @property
    def assignee(self) -> str:
        self.__update()
        try:
            return self._raw_issue["assignee"]["name"]
        except Exception:
            return None

    @property
    def description(self) -> str:
        self.__update()
        return self._raw_issue["content"]

    @description.setter
    def description(self, new_description: str) -> None:
        self.__update_info(description=new_description)

    @property
    def author(self) -> str:
        return self._raw_issue["user"]["name"]

    @property
    def created(self) -> datetime.datetime:
        return datetime.datetime.fromtimestamp(int(self._raw_issue["date_created"]))

    @property
    def labels(self) -> list[IssueLabel]:
        return [PagureIssueLabel(label, self) for label in self._raw_issue["tags"]]

    def __str__(self) -> str:
        return "Pagure" + super().__str__()

    def __update_info(
        self,
        title: Optional[str] = None,
        description: Optional[str] = None,
        assignee: Optional[str] = None,
    ) -> None:
        try:
            data = {
                "title": title if title is not None else self.title,
                "issue_content": (
                    description if description is not None else self.description
                ),
            }

            updated_issue = self.project._call_project_api(
                "issue",
                str(self.id),
                method="POST",
                data=data,
            )
            self._raw_issue = updated_issue["issue"]
        except Exception as ex:
            raise PagureAPIException(
                "there was an error while updating the issue",
            ) from ex

    @staticmethod
    def create(
        project: "ogr_pagure.PagureProject",
        title: str,
        body: str,
        private: Optional[bool] = None,
        labels: Optional[list[str]] = None,
        assignees: Optional[list[str]] = None,
    ) -> "Issue":
        if not project.has_issues:
            raise IssueTrackerDisabled()

        payload = {"title": title, "issue_content": body}
        if labels is not None:
            payload["tag"] = ",".join(labels)
        if private:
            payload["private"] = "true"
        if assignees and len(assignees) > 1:
            raise OperationNotSupported("Pagure does not support multiple assignees")

        if assignees:
            payload["assignee"] = assignees[0]

        new_issue = project._call_project_api("new_issue", data=payload, method="POST")[
            "issue"
        ]
        return PagureIssue(new_issue, project)

    @staticmethod
    def get(project: "ogr_pagure.PagureProject", issue_id: int) -> "Issue":
        if not project.has_issues:
            raise IssueTrackerDisabled()

        raw_issue = project._call_project_api("issue", str(issue_id))
        return PagureIssue(raw_issue, project)

    @staticmethod
    def get_list(
        project: "ogr_pagure.PagureProject",
        status: IssueStatus = IssueStatus.open,
        author: Optional[str] = None,
        assignee: Optional[str] = None,
        labels: Optional[list[str]] = None,
    ) -> list["Issue"]:
        if not project.has_issues:
            raise IssueTrackerDisabled()

        payload: dict[str, Union[str, list[str], int]] = {
            "status": status.name.capitalize(),
            "page": 1,
            "per_page": 100,
        }
        if author:
            payload["author"] = author
        if assignee:
            payload["assignee"] = assignee
        if labels:
            payload["tags"] = labels

        raw_issues: list[Any] = []

        while True:
            issues_info = project._call_project_api("issues", params=payload)
            raw_issues += issues_info["issues"]
            if not issues_info["pagination"]["next"]:
                break
            payload["page"] = cast(int, payload["page"]) + 1

        return [PagureIssue(issue_dict, project) for issue_dict in raw_issues]

    def _get_all_comments(self) -> list[IssueComment]:
        self.__update()
        raw_comments = self._raw_issue["comments"]
        return [
            PagureIssueComment(parent=self, raw_comment=raw_comment)
            for raw_comment in raw_comments
        ]

    def comment(self, body: str) -> IssueComment:
        payload = {"comment": body}
        self.project._call_project_api(
            "issue",
            str(self.id),
            "comment",
            data=payload,
            method="POST",
        )
        self.__dirty = True
        return PagureIssueComment(parent=self, body=body, author=self.project._user)

    def close(self) -> "PagureIssue":
        payload = {"status": "Closed"}
        self.project._call_project_api(
            "issue",
            str(self.id),
            "status",
            data=payload,
            method="POST",
        )
        self.__dirty = True
        return self

    def add_assignee(self, *assignees: str) -> None:
        if len(assignees) > 1:
            raise OperationNotSupported("Pagure does not support multiple assignees")
        payload = {"assignee": assignees[0]}
        self.project._call_project_api(
            "issue",
            str(self.id),
            "assign",
            data=payload,
            method="POST",
        )

    def get_comment(self, comment_id: int) -> IssueComment:
        return PagureIssueComment(
            self.project._call_project_api(
                "issue",
                str(self.id),
                "comment",
                str(comment_id),
                method="GET",
            ),
        )

class PagureIssueComment(PagureComment, IssueComment):
    def __str__(self) -> str:
        return "Pagure" + super().__str__()

class PagurePRComment(PagureComment, PRComment):
    def __str__(self) -> str:
        return "Pagure" + super().__str__()

class PagureProject(BaseGitProject):
    service: "ogr_pagure.PagureService"
    access_dict: ClassVar[dict] = {
        AccessLevel.pull: "ticket",
        AccessLevel.triage: "ticket",
        AccessLevel.push: "commit",
        AccessLevel.admin: "commit",
        AccessLevel.maintain: "admin",
        None: "",
    }

    def __init__(
        self,
        repo: str,
        namespace: Optional[str],
        service: "ogr_pagure.PagureService",
        username: Optional[str] = None,
        is_fork: bool = False,
    ) -> None:
        super().__init__(repo, service, namespace)
        self.read_only = service.read_only

        self._is_fork = is_fork
        self._username = username

        self.repo = repo
        self.namespace = namespace

    def __str__(self) -> str:
        fork_info = ""
        if self._is_fork:
            fork_info = f', username="{self._username}", is_fork={self._is_fork}'
        return f'PagureProject(namespace="{self.namespace}", repo="{self.repo}"{fork_info})'

    def __eq__(self, o: object) -> bool:
        if not isinstance(o, PagureProject):
            return False

        return (
            self.repo == o.repo
            and self.namespace == o.namespace
            and self.service == o.service
            and self._username == o._username
            and self._is_fork == o._is_fork
            and self.read_only == o.read_only
        )

    @property
    def _user(self) -> str:
        if not self._username:
            self._username = self.service.user.get_username()
        return self._username

    def _call_project_api(
        self,
        *args,
        add_fork_part: bool = True,
        add_api_endpoint_part: bool = True,
        method: Optional[str] = None,
        params: Optional[dict] = None,
        data: Optional[dict] = None,
    ) -> dict:
        """
        Call project API endpoint.

        Args:
            *args: String parts of the URL, e.g. `"a", "b"` will call `project/a/b`
            add_fork_part: If the project is a fork, use `fork/username` prefix.

                Defaults to `True`.
            add_api_endpoint_part: Add part with API endpoint (`/api/0/`).

                Defaults to `True`.
            method: Method of the HTTP request, e.g. `"GET"`, `"POST"`, etc.
            params: HTTP(S) query parameters in form of a dictionary.
            data: Data to be sent in form of a dictionary.

        Returns:
            Dictionary representing response.
        """
        request_url = self._get_project_url(
            *args,
            add_api_endpoint_part=add_api_endpoint_part,
            add_fork_part=add_fork_part,
        )

        return self.service.call_api(
            url=request_url,
            method=method,
            params=params,
            data=data,
        )

    def _call_project_api_raw(
        self,
        *args,
        add_fork_part: bool = True,
        add_api_endpoint_part: bool = True,
        method: Optional[str] = None,
        params: Optional[dict] = None,
        data: Optional[dict] = None,
    ) -> RequestResponse:
        """
        Call project API endpoint.

        Args:
            *args: String parts of the URL, e.g. `"a", "b"` will call `project/a/b`
            add_fork_part: If the project is a fork, use `fork/username` prefix.

                Defaults to `True`.
            add_api_endpoint_part: Add part with API endpoint (`/api/0/`).

                Defaults to `True`.
            method: Method of the HTTP request, e.g. `"GET"`, `"POST"`, etc.
            params: HTTP(S) query parameters in form of a dictionary.
            data: Data to be sent in form of a dictionary.

        Returns:
            `RequestResponse` object containing response.
        """
        request_url = self._get_project_url(
            *args,
            add_api_endpoint_part=add_api_endpoint_part,
            add_fork_part=add_fork_part,
        )

        return self.service.call_api_raw(
            url=request_url,
            method=method,
            params=params,
            data=data,
        )

    def _get_project_url(self, *args, add_fork_part=True, add_api_endpoint_part=True):
        additional_parts = []
        if self._is_fork and add_fork_part:
            additional_parts += ["fork", self._user]
        return self.service.get_api_url(
            *additional_parts,
            self.namespace,
            self.repo,
            *args,
            add_api_endpoint_part=add_api_endpoint_part,
        )

    def get_project_info(self):
        return self._call_project_api(method="GET")

    def get_branches(self) -> list[str]:
        return_value = self._call_project_api("git", "branches", method="GET")
        return return_value["branches"]

    @property
    def default_branch(self) -> str:
        return_value = self._call_project_api("git", "branches", method="GET")
        return return_value["default"]

    def get_description(self) -> str:
        return self.get_project_info()["description"]

    @property
    def description(self) -> str:
        return self.get_project_info()["description"]

    @description.setter
    def description(self, new_description: str) -> None:
        raise OperationNotSupported("Not possible on Pagure")

    @property
    def has_issues(self) -> bool:
        options = self._call_project_api("options", method="GET")
        return options["settings"]["issue_tracker"]

    def get_owners(self) -> list[str]:
        project = self.get_project_info()
        return project["access_users"]["owner"]

    def who_can_close_issue(self) -> set[str]:
        users: set[str] = set()
        project = self.get_project_info()
        users.update(project["access_users"]["admin"])
        users.update(project["access_users"]["commit"])
        users.update(project["access_users"]["ticket"])
        users.update(project["access_users"]["owner"])
        return users

    def who_can_merge_pr(self) -> set[str]:
        users: set[str] = set()
        project = self.get_project_info()
        users.update(project["access_users"]["admin"])
        users.update(project["access_users"]["commit"])
        users.update(project["access_users"]["owner"])
        return users

    def which_groups_can_merge_pr(self) -> set[str]:
        groups: set[str] = set()
        project = self.get_project_info()
        groups.update(project["access_groups"]["admin"])
        groups.update(project["access_groups"]["commit"])
        return groups

    def can_merge_pr(self, username) -> bool:
        accounts_that_can_merge_pr = self.who_can_merge_pr()

        groups_that_can_merge_pr = self.which_groups_can_merge_pr()
        accounts_that_can_merge_pr.update(
            member
            for group in groups_that_can_merge_pr
            for member in self.service.get_group(group).members
        )

        logger.info(
            f"All users (considering groups) that can merge PR: {accounts_that_can_merge_pr}",
        )
        return username in accounts_that_can_merge_pr

    def request_access(self):
        raise OperationNotSupported("Not possible on Pagure")

    @indirect(PagureIssue.get_list)
    def get_issue_list(
        self,
        status: IssueStatus = IssueStatus.open,
        author: Optional[str] = None,
        assignee: Optional[str] = None,
        labels: Optional[list[str]] = None,
    ) -> list[Issue]:
        pass

    @indirect(PagureIssue.get)
    def get_issue(self, issue_id: int) -> Issue:
        pass

    def delete(self) -> None:
        self._call_project_api_raw("delete", method="POST")

    @indirect(PagureIssue.create)
    def create_issue(
        self,
        title: str,
        body: str,
        private: Optional[bool] = None,
        labels: Optional[list[str]] = None,
        assignees: Optional[list[str]] = None,
    ) -> Issue:
        pass

    @indirect(PagurePullRequest.get_list)
    def get_pr_list(
        self,
        status: PRStatus = PRStatus.open,
        assignee=None,
        author=None,
    ) -> list[PullRequest]:
        pass

    @indirect(PagurePullRequest.get)
    def get_pr(self, pr_id: int) -> PullRequest:
        pass

    @indirect(PagurePullRequest.get_files_diff)
    def get_pr_files_diff(
        self,
        pr_id: int,
        retries: int = 0,
        wait_seconds: int = 3,
    ) -> dict:
        pass

    @if_readonly(return_function=GitProjectReadOnly.create_pr)
    @indirect(PagurePullRequest.create)
    def create_pr(
        self,
        title: str,
        body: str,
        target_branch: str,
        source_branch: str,
        fork_username: Optional[str] = None,
    ) -> PullRequest:
        pass

    @if_readonly(return_function=GitProjectReadOnly.fork_create)
    def fork_create(self, namespace: Optional[str] = None) -> "PagureProject":
        if namespace is not None:
            raise OperationNotSupported(
                "Pagure does not support forking to namespaces.",
            )

        request_url = self.service.get_api_url("fork")
        self.service.call_api(
            url=request_url,
            method="POST",
            data={"repo": self.repo, "namespace": self.namespace, "wait": True},
        )
        fork = self._construct_fork_project()
        logger.debug(f"Forked to {fork.full_repo_name}")
        return fork

    def _construct_fork_project(self) -> "PagureProject":
        return PagureProject(
            service=self.service,
            repo=self.repo,
            namespace=self.namespace,
            username=self._user,
            is_fork=True,
        )

    def get_fork(self, create: bool = True) -> Optional["PagureProject"]:
        if self.is_fork:
            raise OgrException("Cannot create fork from fork.")

        for fork in self.get_forks():
            fork_info = fork.get_project_info()
            if self._user == fork_info["user"]["name"]:
                return fork

        if not self.is_forked():
            if create:
                return self.fork_create()

            logger.info(
                f"Fork of {self.repo}"
                " does not exist and we were asked not to create it.",
            )
            return None
        return self._construct_fork_project()

    def exists(self) -> bool:
        response = self._call_project_api_raw()
        return response.ok

    def is_private(self) -> bool:
        host = urlparse(self.service.instance_url).hostname
        if host in [
            "git.centos.org",
            "git.stg.centos.org",
            "pagure.io",
            "src.fedoraproject.org",
            "src.stg.fedoraproject.org",
        ]:
            # private repositories are not allowed on generally used pagure instances
            return False
        raise OperationNotSupported(
            f"is_private is not implemented for {self.service.instance_url}."
            f"Please open issue in https://github.com/packit/ogr",
        )

    def is_forked(self) -> bool:
        f = self._construct_fork_project()
        return bool(f.exists() and f.parent.exists())

    def get_is_fork_from_api(self) -> bool:
        return bool(self.get_project_info()["parent"])

    @property
    def is_fork(self) -> bool:
        return self._is_fork

    @property
    def parent(self) -> Optional["PagureProject"]:
        if self.get_is_fork_from_api():
            return PagureProject(
                repo=self.repo,
                namespace=self.get_project_info()["parent"]["namespace"],
                service=self.service,
            )
        return None

    def get_git_urls(self) -> dict[str, str]:
        return_value = self._call_project_api("git", "urls")
        return return_value["urls"]

    def add_user(self, user: str, access_level: AccessLevel) -> None:
        self.add_user_or_group(user, access_level, "user")

    def remove_user(self, user: str) -> None:
        self.add_user_or_group(user, None, "user")

    def add_group(self, group: str, access_level: AccessLevel):
        self.add_user_or_group(group, access_level, "group")

    def remove_group(self, group: str) -> None:
        self.add_user_or_group(group, None, "group")

    def add_user_or_group(
        self,
        user: str,
        access_level: Optional[AccessLevel],
        user_type: str,
    ) -> None:
        response = self._call_project_api_raw(
            "git",
            "modifyacls",
            method="POST",
            data={
                "user_type": user_type,
                "name": user,
                "acl": self.access_dict[access_level],
            },
        )

        if response.status_code == 401:
            raise PagureAPIException(
                "You are not allowed to modify ACL's",
                response_code=response.status_code,
            )

    def change_token(self, new_token: str) -> None:
        self.service.change_token(new_token)

    def get_file_content(self, path: str, ref=None) -> str:
        ref = ref or self.default_branch
        result = self._call_project_api_raw(
            "raw",
            ref,
            "f",
            path,
            add_api_endpoint_part=False,
        )

        if not result or result.reason == "NOT FOUND":
            raise FileNotFoundError(f"File '{path}' on {ref} not found")
        if result.reason != "OK":
            raise PagureAPIException(
                f"File '{path}' on {ref} not found due to {result.reason}",
            )
        return result.content.decode()

    def get_sha_from_tag(self, tag_name: str) -> str:
        tags_dict = self.get_tags_dict()
        if tag_name not in tags_dict:
            raise PagureAPIException(f"Tag '{tag_name}' not found.", response_code=404)

        return tags_dict[tag_name].commit_sha

    def commit_comment(
        self,
        commit: str,
        body: str,
        filename: Optional[str] = None,
        row: Optional[int] = None,
    ) -> CommitComment:
        raise OperationNotSupported("Commit comments are not supported on Pagure.")

    def get_commit_comments(self, commit: str) -> list[CommitComment]:
        raise OperationNotSupported("Commit comments are not supported on Pagure.")

    def get_commit_comment(self, commit_sha: str, comment_id: int) -> CommitComment:
        raise OperationNotSupported("Commit comments are not supported on Pagure.")

    @if_readonly(return_function=GitProjectReadOnly.set_commit_status)
    @indirect(PagureCommitFlag.set)
    def set_commit_status(
        self,
        commit: str,
        state: CommitStatus,
        target_url: str,
        description: str,
        context: str,
        percent: Optional[int] = None,
        uid: Optional[str] = None,
        trim: bool = False,
    ) -> "CommitFlag":
        pass

    @indirect(PagureCommitFlag.get)
    def get_commit_statuses(self, commit: str) -> list[CommitFlag]:
        pass

    def get_tags(self) -> list[GitTag]:
        response = self._call_project_api("git", "tags", params={"with_commits": True})
        return [GitTag(name=n, commit_sha=c) for n, c in response["tags"].items()]

    def get_tags_dict(self) -> dict[str, GitTag]:
        response = self._call_project_api("git", "tags", params={"with_commits": True})
        return {n: GitTag(name=n, commit_sha=c) for n, c in response["tags"].items()}

    @indirect(PagureRelease.get_list)
    def get_releases(self) -> list[Release]:
        pass

    @indirect(PagureRelease.get)
    def get_release(self, identifier=None, name=None, tag_name=None) -> PagureRelease:
        pass

    @indirect(PagureRelease.get_latest)
    def get_latest_release(self) -> Optional[PagureRelease]:
        pass

    @indirect(PagureRelease.create)
    def create_release(
        self,
        tag: str,
        name: str,
        message: str,
        ref: Optional[str] = None,
    ) -> Release:
        pass

    def get_forks(self) -> list["PagureProject"]:
        forks_url = self.service.get_api_url("projects")
        projects_response = self.service.call_api(
            url=forks_url,
            params={"fork": True, "pattern": self.repo},
        )
        return [
            PagureProject(
                repo=fork["name"],
                namespace=fork["namespace"],
                service=self.service,
                username=fork["user"]["name"],
                is_fork=True,
            )
            for fork in projects_response["projects"]
        ]

    def get_web_url(self) -> str:
        return f'{self.service.instance_url}/{self.get_project_info()["url_path"]}'

    @property
    def full_repo_name(self) -> str:
        fork = f"fork/{self._user}/" if self.is_fork else ""
        namespace = f"{self.namespace}/" if self.namespace else ""
        return f"{fork}{namespace}{self.repo}"

    def __get_files(
        self,
        path: str,
        ref: Optional[str] = None,
        recursive: bool = False,
    ) -> Iterable[str]:
        subfolders = ["."]

        while subfolders:
            path = subfolders.pop()
            split_path = []
            if path != ".":
                split_path = ["f", *path.split("/")]
            response = self._call_project_api("tree", ref, *split_path)

            for file in response["content"]:
                if file["type"] == "file":
                    yield file["path"]
                elif recursive and file["type"] == "folder":
                    subfolders.append(file["path"])

    def get_files(
        self,
        ref: Optional[str] = None,
        filter_regex: Optional[str] = None,
        recursive: bool = False,
    ) -> list[str]:
        ref = ref or self.default_branch
        paths = list(self.__get_files(".", ref, recursive))
        if filter_regex:
            paths = filter_paths(paths, filter_regex)

        return paths

    def get_sha_from_branch(self, branch: str) -> Optional[str]:
        branches = self._call_project_api(
            "git",
            "branches",
            params={"with_commits": True},
        )["branches"]

        return branches.get(branch)

    def get_contributors(self) -> set[str]:
        raise OperationNotSupported("Pagure doesn't provide list of contributors")

    def users_with_write_access(self) -> set[str]:
        return self._get_users_with_given_access(["commit", "admin", "owner"])

    def get_users_with_given_access(self, access_levels: list[AccessLevel]) -> set[str]:
        access_levels_pagure = [
            self.access_dict[access_level] for access_level in access_levels
        ]

        # for AccessLevel.maintain get the maintainer as well
        if AccessLevel.maintain in access_levels:
            access_levels_pagure.append("owner")

        return self._get_users_with_given_access(access_levels_pagure)

    def _get_users_with_given_access(self, access_levels: list[str]) -> set[str]:
        """
        Get all users (considering groups) with the access levels given by list.

        Arguments:
            access_levels: list of access levels, e.g. ['commit', 'admin']
        """
        users = self._get_user_accounts_with_access(access_levels)

        # group cannot have owner access
        group_accounts = self._get_group_accounts_with_access(
            list(set(access_levels) - {"owner"}),
        )

        users.update(
            member
            for group in group_accounts
            for member in self.service.get_group(group).members
        )

        logger.info(
            f"All users (considering groups) with given access levels: {users}",
        )
        return users

    def _get_entity_accounts_with_access(
        self,
        access_levels: list[str],
        entity_type: str,
    ) -> set[str]:
        """
        Get the entity account names (users or groups) with the access levels given by the set.

        Arguments:
            access_levels: list of access levels, e.g. ['commit', 'admin']
            entity_type: 'users' or 'groups'
        """
        if entity_type not in ("users", "groups"):
            raise OgrException(
                f"Unsupported entity type {entity_type}: only 'users' and 'groups' are allowed.",
            )
        entity_info = self.get_project_info()["access_" + entity_type]
        result = set()
        for access_level in access_levels:
            result.update(entity_info[access_level])

        return result

    def _get_user_accounts_with_access(self, access_levels: list[str]) -> set[str]:
        """
        Get the users with the access levels given by the set.
        """
        return self._get_entity_accounts_with_access(access_levels, "users")

    def _get_group_accounts_with_access(self, access_levels: list[str]) -> set[str]:
        """
        Get the groups with the access levels given by list.
        """
        return self._get_entity_accounts_with_access(access_levels, "groups")

class PagurePullRequest(BasePullRequest):
    _target_project: "ogr_pagure.PagureProject"
    _source_project: "ogr_pagure.PagureProject" = None

    def __init__(self, raw_pr, project):
        super().__init__(raw_pr, project)
        self.__dirty = False

    def __update(self):
        if self.__dirty:
            self._raw_pr = self.__call_api()
            self.__dirty = False

    @property
    def title(self) -> str:
        self.__update()
        return self._raw_pr["title"]

    @title.setter
    def title(self, new_title: str) -> None:
        self.update_info(title=new_title)

    @property
    def id(self) -> int:
        return self._raw_pr["id"]

    @property
    def status(self) -> PRStatus:
        self.__update()
        return PRStatus[self._raw_pr["status"].lower()]

    @property
    def url(self) -> str:
        return "/".join(
            [
                self.target_project.service.instance_url,
                self._raw_pr["project"]["url_path"],
                "pull-request",
                str(self.id),
            ],
        )

    @property
    def description(self) -> str:
        self.__update()
        return self._raw_pr["initial_comment"]

    @description.setter
    def description(self, new_description: str) -> None:
        self.update_info(description=new_description)

    @property
    def author(self) -> str:
        return self._raw_pr["user"]["name"]

    @property
    def source_branch(self) -> str:
        return self._raw_pr["branch_from"]

    @property
    def target_branch(self) -> str:
        return self._raw_pr["branch"]

    @property
    def created(self) -> datetime.datetime:
        return datetime.datetime.fromtimestamp(int(self._raw_pr["date_created"]))

    @property
    def diff_url(self) -> str:
        return f"{self.url}#request_diff"

    @property
    def commits_url(self) -> str:
        return f"{self.url}#commit_list"

    @property
    def patch(self) -> bytes:
        request_response = self._target_project._call_project_api_raw(
            "pull-request",
            f"{self.id}.patch",
            add_api_endpoint_part=False,
        )
        if request_response.status_code != 200:
            raise PagureAPIException(
                f"Cannot get patch from {self.url}.patch because {request_response.reason}.",
                response_code=request_response.status_code,
            )
        return request_response.content

    @property
    def head_commit(self) -> str:
        return self._raw_pr["commit_stop"]

    @property
    def source_project(self) -> "ogr_pagure.PagureProject":
        if self._source_project is None:
            source = self._raw_pr["repo_from"]
            source_project_info = {
                "repo": source["name"],
                "namespace": source["namespace"],
            }

            if source["parent"] is not None:
                source_project_info["is_fork"] = True
                source_project_info["username"] = source["user"]["name"]

            self._source_project = self._target_project.service.get_project(
                **source_project_info,
            )

        return self._source_project

    @property
    def closed_by(self) -> Optional[str]:
        closed_by = self._raw_pr["closed_by"]
        return closed_by["name"] if closed_by else None

    @property
    def labels(self) -> list[PRLabel]:
        return [PagurePRLabel(label, self) for label in self._raw_pr["tags"]]

    def __str__(self) -> str:
        return "Pagure" + super().__str__()

    def __call_api(self, *args, **kwargs) -> dict:
        return self._target_project._call_project_api(
            "pull-request",
            str(self.id),
            *args,
            **kwargs,
        )

    @staticmethod
    def create(
        project: "ogr_pagure.PagureProject",
        title: str,
        body: str,
        target_branch: str,
        source_branch: str,
        fork_username: Optional[str] = None,
    ) -> "PullRequest":
        data = {
            "title": title,
            "branch_to": target_branch,
            "branch_from": source_branch,
            "initial_comment": body,
        }

        caller = project
        if project.is_fork:
            data["repo_from"] = project.repo
            data["repo_from_username"] = project._user
            data["repo_from_namespace"] = project.namespace

            # running the call from the parent project
            caller = caller.parent
        elif fork_username:
            fork_project = project.service.get_project(
                username=fork_username,
                repo=project.repo,
                namespace=project.namespace,
                is_fork=True,
            )
            data["repo_from_username"] = fork_username
            data["repo_from"] = fork_project.repo
            data["repo_from_namespace"] = fork_project.namespace

        response = caller._call_project_api(
            "pull-request",
            "new",
            method="POST",
            data=data,
        )
        return PagurePullRequest(response, caller)

    @staticmethod
    def get(project: "ogr_pagure.PagureProject", pr_id: int) -> "PullRequest":
        raw_pr = project._call_project_api("pull-request", str(pr_id))
        return PagurePullRequest(raw_pr, project)

    @staticmethod
    def get_files_diff(
        project: "ogr_pagure.PagureProject",
        pr_id: int,
        retries: int = 0,
        wait_seconds: int = 3,
    ) -> dict:
        """
        Retrieve pull request diff statistics.

        Pagure API tends to return ENOPRSTATS error when a pull request is transitioning
        from open to other states, so you can use `retries` and `wait_seconds` to try to
        mitigate that.


        Args:
            project: Pagure project.
            pr_id: Pull request ID.
            retries: Number of extra attempts.
            wait_seconds: Delay between attempts.
        """
        attempt = 1
        while True:
            try:
                return project._call_project_api(
                    "pull-request",
                    str(pr_id),
                    "diffstats",
                    method="GET",
                )
            except PagureAPIException as ex:  # noqa PERF203
                if "No statistics" in ex.pagure_error:
                    # this may be a race condition, try once more
                    logger.info(
                        f"While retrieving PR diffstats Pagure returned ENOPRSTATS.\n{ex}",
                    )
                    if attempt <= retries:
                        attempt += 1
                        logger.info(
                            f"Trying again; attempt={attempt} after {wait_seconds} seconds",
                        )
                        sleep(wait_seconds)
                        continue
                raise ex

    @staticmethod
    def get_list(
        project: "ogr_pagure.PagureProject",
        status: PRStatus = PRStatus.open,
        assignee=None,
        author=None,
    ) -> list["PullRequest"]:
        payload = {"page": 1, "status": status.name.capitalize()}
        if assignee is not None:
            payload["assignee"] = assignee
        if author is not None:
            payload["author"] = author

        raw_prs = []
        while True:
            page_result = project._call_project_api("pull-requests", params=payload)
            raw_prs += page_result["requests"]
            if not page_result["pagination"]["next"]:
                break

            # mypy don't know that key "page" really contains int...
            payload["page"] += 1  # type: ignore

        return [PagurePullRequest(pr_dict, project) for pr_dict in raw_prs]

    def update_info(
        self,
        title: Optional[str] = None,
        description: Optional[str] = None,
    ) -> "PullRequest":
        try:
            data = {"title": title if title else self.title}

            if description:
                data["initial_comment"] = description

            updated_pr = self.__call_api(method="POST", data=data)
            logger.info("PR updated.")

            self._raw_pr = updated_pr
            return self
        except Exception as ex:
            raise PagureAPIException("there was an error while updating the PR") from ex

    def _get_all_comments(self) -> list[PRComment]:
        self.__update()
        raw_comments = self._raw_pr["comments"]
        return [
            PagurePRComment(parent=self, raw_comment=comment_dict)
            for comment_dict in raw_comments
        ]

    def comment(
        self,
        body: str,
        commit: Optional[str] = None,
        filename: Optional[str] = None,
        row: Optional[int] = None,
    ) -> "PRComment":
        payload: dict[str, Any] = {"comment": body}
        if commit is not None:
            payload["commit"] = commit
        if filename is not None:
            payload["filename"] = filename
        if row is not None:
            payload["row"] = row

        self.__call_api("comment", method="POST", data=payload)
        self.__dirty = True
        return PagurePRComment(
            parent=self,
            body=body,
            author=self.target_project.service.user.get_username(),
        )

    def close(self) -> "PullRequest":
        return_value = self.__call_api("close", method="POST")

        if return_value["message"] != "Pull-request closed!":
            raise PagureAPIException(return_value["message"])

        self.__dirty = True
        return self

    def merge(self) -> "PullRequest":
        return_value = self.__call_api("merge", method="POST")

        if return_value["message"] != "Changes merged!":
            raise PagureAPIException(return_value["message"])

        self.__dirty = True
        return self

    def get_statuses(self) -> list[CommitFlag]:
        self.__update()
        return self.target_project.get_commit_statuses(self._raw_pr["commit_stop"])

    def set_flag(
        self,
        username: str,
        comment: str,
        url: str,
        status: Optional[CommitStatus] = None,
        percent: Optional[int] = None,
        uid: Optional[str] = None,
    ) -> dict:
        """
        Set a flag on a pull-request to display results or status of CI tasks.

        See "Flag a pull-request" at https://pagure.io/api/0/#pull_requests-tab
        for a full description of the parameters.

        Args:
            username: The name of the application to be presented to users
                on the pull request page.
            comment: A short message summarizing the presented results.
            url: A URL to the result of this flag.
            status: The status to be displayed for this flag.
            percent: A percentage of completion compared to the goal.
            uid: A unique identifier used to identify a flag on the pull-request.

        Returns:
            Dictionary with the response received from Pagure.
        """
        data: dict[str, Union[str, int]] = {
            "username": username,
            "comment": comment,
            "url": url,
        }
        if status is not None:
            data["status"] = status.name
        if percent is not None:
            data["percent"] = percent
        if uid is not None:
            data["uid"] = uid
        return self.__call_api("flag", method="POST", data=data)

    def get_comment(self, comment_id: int) -> PRComment:
        for comment in self._get_all_comments():
            if comment.id == comment_id:
                return comment

        raise PagureAPIException(
            f"No comment with id#{comment_id} in PR#{self.id} found.",
            response_code=404,
        )

class PagureRelease(Release):
    _raw_release: GitTag
    project: "ogr_pagure.PagureProject"

    @property
    def title(self):
        return self.git_tag.name

    @property
    def body(self):
        return ""

    @property
    def git_tag(self) -> GitTag:
        return self._raw_release

    @property
    def tag_name(self) -> str:
        return self._raw_release.name

    @property
    def url(self) -> Optional[str]:
        return ""

    @property
    def created_at(self) -> datetime.datetime:
        return None

    @property
    def tarball_url(self) -> str:
        return ""

    def __str__(self) -> str:
        return "Pagure" + super().__str__()

    @staticmethod
    def get(
        project: "ogr_pagure.PagureProject",
        identifier: Optional[int] = None,
        name: Optional[str] = None,
        tag_name: Optional[str] = None,
    ) -> "Release":
        raise OperationNotSupported()

    @staticmethod
    def get_latest(project: "ogr_pagure.PagureProject") -> Optional["Release"]:
        raise OperationNotSupported("Pagure API does not provide timestamps")

    @staticmethod
    def get_list(project: "ogr_pagure.PagureProject") -> list["Release"]:
        # git tag for Pagure is shown as Release in Pagure UI
        git_tags = project.get_tags()
        return [PagureRelease(git_tag, project) for git_tag in git_tags]

    @staticmethod
    def create(
        project: "ogr_pagure.PagureProject",
        tag: str,
        name: str,
        message: str,
        ref: Optional[str] = None,
    ) -> "Release":
        payload = {
            "tagname": tag,
            "commit_hash": ref,
        }
        if message:
            payload["message"] = message

        response = project._call_project_api("git", "tags", data=payload, method="POST")
        if not response["tag_created"]:
            raise PagureAPIException("Release has not been created")

        return PagureRelease(GitTag(tag, ref), project)

    def edit_release(self, name: str, message: str) -> None:
        raise OperationNotSupported("edit_release not supported on Pagure")

@use_for_service("pagure")
@use_for_service("src.fedoraproject.org")
@use_for_service("src.stg.fedoraproject.org")
@use_for_service("pkgs.fedoraproject.org")
@use_for_service("pkgs.stg.fedoraproject.org")
@use_for_service("git.centos.org")
@use_for_service("git.stg.centos.org")
class PagureService(BaseGitService):
    def __init__(
        self,
        token: Optional[str] = None,
        instance_url: str = "https://src.fedoraproject.org",
        read_only: bool = False,
        insecure: bool = False,
        max_retries: Union[int, urllib3.util.Retry] = 5,
        **kwargs,
    ) -> None:
        super().__init__()
        self.instance_url = instance_url
        self._token = token
        self.read_only = read_only

        self.session = requests.session()

        adapter = requests.adapters.HTTPAdapter(max_retries=max_retries)

        self.insecure = insecure
        if self.insecure:
            self.session.mount("http://", adapter)
        else:
            self.session.mount("https://", adapter)

        self.header = {"Authorization": "token " + self._token} if self._token else {}

        if kwargs:
            logger.warning(f"Ignored keyword arguments: {kwargs}")

    def __str__(self) -> str:
        token_str = (
            f", token='{self._token[:1]}***{self._token[-1:]}'" if self._token else ""
        )
        insecure_str = ", insecure=True" if self.insecure else ""
        readonly_str = ", read_only=True" if self.read_only else ""

        return (
            f"PagureService(instance_url='{self.instance_url}'"
            f"{token_str}"
            f"{readonly_str}"
            f"{insecure_str})"
        )

    def __eq__(self, o: object) -> bool:
        if not issubclass(o.__class__, PagureService):
            return False

        return (
            self._token == o._token  # type: ignore
            and self.read_only == o.read_only  # type: ignore
            and self.instance_url == o.instance_url  # type: ignore
            and self.insecure == o.insecure  # type: ignore
            and self.header == o.header  # type: ignore
        )

    def __hash__(self) -> int:
        return hash(str(self))

    def get_project(self, **kwargs) -> "PagureProject":
        if "username" in kwargs:
            return PagureProject(service=self, **kwargs)

        return PagureProject(
            service=self,
            username=self.user.get_username(),
            **kwargs,
        )

    def get_project_from_url(self, url: str) -> "PagureProject":
        repo_url = parse_git_repo(potential_url=url)
        if not repo_url:
            raise OgrException(f"Cannot parse project url: '{url}'")

        if not repo_url.is_fork:
            repo_url.username = None

        return self.get_project(
            repo=repo_url.repo,
            namespace=repo_url.namespace,
            is_fork=repo_url.is_fork,
            username=repo_url.username,
        )

    @property
    def user(self) -> "PagureUser":
        return PagureUser(service=self)

    def call_api(
        self,
        url: str,
        method: Optional[str] = None,
        params: Optional[dict] = None,
        data=None,
    ) -> dict:
        """
        Call API endpoint.

        Args:
            url: URL to be called.
            method: Method of the HTTP request, e.g. `"GET"`, `"POST"`, etc.
            params: HTTP(S) query parameters in form of a dictionary.
            data: Data to be sent in form of a dictionary.

        Returns:
            Dictionary representing response.

        Raises:
            PagureAPIException, if error occurs.
        """
        response = self.call_api_raw(url=url, method=method, params=params, data=data)

        if response.status_code == 404:
            error_msg = (
                response.json_content["error"]
                if response.json_content and "error" in response.json_content
                else None
            )
            raise PagureAPIException(
                f"Page '{url}' not found when calling Pagure API.",
                pagure_error=error_msg,
                response_code=response.status_code,
            )

        if not response.json_content:
            logger.debug(response.content)
            raise PagureAPIException(
                "Error while decoding JSON: {0}",
                response_code=response.status_code,
            )

        if not response.ok:
            logger.error(response.json_content)
            if "error" in response.json_content:
                error_msg = response.json_content["error"]
                error_msg_ext = response.json_content.get("errors", "")
                msg = f"Pagure API returned an error when calling '{url}': {error_msg}"
                if error_msg_ext:
                    msg += f" - {error_msg_ext}"
                raise PagureAPIException(
                    msg,
                    pagure_error=error_msg,
                    pagure_response=response.json_content,
                    response_code=response.status_code,
                )
            raise PagureAPIException(
                f"Problem with Pagure API when calling '{url}'",
                response_code=response.status_code,
            )

        return response.json_content

    def call_api_raw(
        self,
        url: str,
        method: Optional[str] = None,
        params: Optional[dict] = None,
        data=None,
    ):
        """
        Call API endpoint and returns raw response.

        Args:
            url: URL to be called.
            method: Method of the HTTP request, e.g. `"GET"`, `"POST"`, etc.
            params: HTTP(S) query parameters in form of a dictionary.
            data: Data to be sent in form of a dictionary.

        Returns:
            `RequestResponse` object that represents the response from the API
            endpoint.
        """

        method = method or "GET"
        try:
            response = self.get_raw_request(
                method=method,
                url=url,
                params=params,
                data=data,
            )

        except requests.exceptions.ConnectionError as er:
            logger.error(er)
            raise OgrNetworkError(f"Cannot connect to url: '{url}'.") from er

        if response.status_code >= 500:
            raise GitForgeInternalError(
                f"Pagure API returned {response.status_code} status for `{url}`"
                f" with reason: `{response.reason}`",
            )

        return response

    def get_raw_request(
        self,
        url,
        method="GET",
        params=None,
        data=None,
        header=None,
    ) -> RequestResponse:
        """
        Call API endpoint and wrap the response in `RequestResponse` type.

        Args:
            url: URL to be called.
            method: Method of the HTTP request, e.g. `"GET"`, `"POST"`, etc.

                Defaults to `"GET"`.
            params: HTTP(S) query parameters in form of a dictionary.
            data: Data to be sent in form of a dictionary.
            header: Header of the HTTP request.

        Returns:
            `RequestResponse` object representing the response.

        Raises:
            ValueError, if JSON cannot be retrieved.
        """

        response = self.session.request(
            method=method,
            url=url,
            params=params,
            headers=header or self.header,
            data=data,
            verify=not self.insecure,
        )

        json_output = None
        try:
            json_output = response.json()
        except ValueError:
            logger.debug(response.text)

        return RequestResponse(
            status_code=response.status_code,
            ok=response.ok,
            content=response.content,
            json=json_output,
            reason=response.reason,
        )

    @property
    def api_url(self):
        """URL to the Pagure API."""
        return f"{self.instance_url}/api/0/"

    def get_api_url(self, *args, add_api_endpoint_part: bool = True) -> str:
        """
        Get a URL from its parts.

        Args:
            *args: String parts of the URL, e.g. `"a", "b"` will call `project/a/b`
            add_api_endpoint_part: Add part with API endpoint (`/api/0/`).

                Defaults to `True`.

        Returns:
            String
        """
        args_list: list[str] = []

        args_list += filter(lambda x: x is not None, args)

        if add_api_endpoint_part:
            return self.api_url + "/".join(args_list)
        return f"{self.instance_url}/" + "/".join(args_list)

    def get_api_version(self) -> str:
        """
        Returns:
            Version of the Pagure API.
        """
        request_url = self.get_api_url("version")
        return_value = self.call_api(request_url)
        return return_value["version"]

    def get_error_codes(self):
        """
        Returns:
            Dictionary with all error codes.
        """
        request_url = self.get_api_url("error_codes")
        return self.call_api(request_url)

    def change_token(self, token: str):
        self._token = token
        self.header = {"Authorization": "token " + self._token}

    def __handle_project_create_fail(
        self,
        exception: PagureAPIException,
        namespace: str,
    ) -> None:
        if (
            exception.pagure_response
            and exception.pagure_response["errors"]["namespace"][0]
            == "Not a valid choice"
        ):
            request_url = self.get_api_url("group", namespace)

            try:
                self.call_api(request_url, data={"projects": False})
            except PagureAPIException as ex:
                raise OgrException(f"Namespace doesn't exist ({namespace}).") from ex

            raise OgrException(
                "Cannot create project in given namespace (permissions).",
            )

        raise exception

    def project_create(
        self,
        repo: str,
        namespace: Optional[str] = None,
        description: Optional[str] = None,
    ) -> PagureProject:
        request_url = self.get_api_url("new")

        parameters = {"name": repo, "description": description, "wait": True}
        if not description:
            parameters["description"] = repo
        if namespace:
            parameters["namespace"] = namespace

        try:
            self.call_api(request_url, "POST", data=parameters)
        except PagureAPIException as ex:
            self.__handle_project_create_fail(ex, namespace)
        return PagureProject(repo=repo, namespace=namespace, service=self)

    def list_projects(
        self,
        namespace: Optional[str] = None,
        user: Optional[str] = None,
        search_pattern: Optional[str] = None,
        language: Optional[str] = None,
    ) -> list[GitProject]:
        raise OperationNotSupported

    def get_group(self, group_name: str) -> PagureGroup:
        """
        Get a Pagure group by name.
        """
        url = self.get_api_url("group", group_name)
        return PagureGroup(group_name, self.call_api(url))

class PagureUser(BaseGitUser):
    service: "ogr_pagure.PagureService"

    def __init__(self, service: "ogr_pagure.PagureService") -> None:
        super().__init__(service=service)

    def __str__(self) -> str:
        return f'PagureUser(username="{self.get_username()}")'

    def get_username(self) -> str:
        request_url = self.service.get_api_url("-", "whoami")

        return_value = self.service.call_api(url=request_url, method="POST", data={})
        return return_value["username"]

    def get_projects(self) -> list["PagureProject"]:
        user_url = self.service.get_api_url("user", self.get_username())
        raw_projects = self.service.call_api(user_url)["repos"]

        return [
            PagureProject(
                repo=project["name"],
                namespace=project["namespace"],
                service=self.service,
            )
            for project in raw_projects
        ]

    def get_forks(self) -> list["PagureProject"]:
        user_url = self.service.get_api_url("user", self.get_username())
        raw_forks = self.service.call_api(user_url)["forks"]

        return [
            PagureProject(
                repo=fork["name"],
                namespace=fork["namespace"],
                service=self.service,
                is_fork=True,
            )
            for fork in raw_forks
        ]

    def get_email(self) -> str:
        # Not supported by Pagure
        raise OperationNotSupported(
            "Pagure does not support retrieving of user's email address",
        )