diff --git a/behave/features/git-pr.feature b/behave/features/git-pr.feature index 4fa1ddd6..8b04b9a9 100644 --- a/behave/features/git-pr.feature +++ b/behave/features/git-pr.feature @@ -29,6 +29,7 @@ Scenario: List pull requests Allow edit : no Author : Admin \(admin@example.com\) Source : Admin/test-GitPkgA, branch: factory, commit: .* + Target : pool/test-GitPkgA, branch: factory, commit: .* Description : some text """ And stderr is @@ -39,7 +40,6 @@ Scenario: List pull requests * URL: http://localhost:{context.podman.container.ports[gitea_http]} * User: Admin - Total entries: 1 """ @@ -65,7 +65,6 @@ Scenario: Search pull requests * URL: http://localhost:{context.podman.container.ports[gitea_http]} * User: Admin - Total entries: 1 """ @@ -85,6 +84,7 @@ Scenario: Get a pull request Allow edit : no Author : Admin \(admin@example.com\) Source : Admin/test-GitPkgA, branch: factory, commit: .* + Target : pool/test-GitPkgA, branch: factory, commit: .* Description : some text """ And stderr is diff --git a/osc/commandline_git.py b/osc/commandline_git.py index b56cf14f..3162a488 100644 --- a/osc/commandline_git.py +++ b/osc/commandline_git.py @@ -128,12 +128,12 @@ def complete_pr(prefix, parsed_args, **kwargs): gitea_conf = gitea_api.Config(conf) gitea_login = gitea_conf.get_login(name=login) gitea_conn = gitea_api.Connection(gitea_login) - data = gitea_api.PullRequest.search( + pr_obj_list = gitea_api.PullRequest.search( gitea_conn, state="open", - ).json() - data.sort(key=gitea_api.PullRequest.cmp) - return [f"{entry['repository']['full_name']}#{entry['number']}" for entry in data] + ) + pr_obj_list.sort() + return [pr_obj.id for pr_obj in pr_obj_list] def complete_checkout_pr(prefix, parsed_args, **kwargs): @@ -147,14 +147,14 @@ def complete_checkout_pr(prefix, parsed_args, **kwargs): gitea_conf = gitea_api.Config(conf) gitea_login = gitea_conf.get_login(name=login) gitea_conn = gitea_api.Connection(gitea_login) - data = gitea_api.PullRequest.list( + pr_obj_list = gitea_api.PullRequest.list( gitea_conn, owner=owner, repo=repo, state="open", - ).json() - data.sort(key=gitea_api.PullRequest.cmp) - return [f"{entry['number']}" for entry in data] + ) + pr_obj_list.sort() + return [str(pr_obj.number) for pr_obj in pr_obj_list] class GitObsMainCommand(osc.commandline_common.MainCommand): diff --git a/osc/commands/fork.py b/osc/commands/fork.py index 2992c9f3..296d47cc 100644 --- a/osc/commands/fork.py +++ b/osc/commands/fork.py @@ -147,17 +147,17 @@ class ForkCommand(osc.commandline.OscCommand): if branch: fork_branch = branch else: - repo_data = gitea_api.Repo.get(gitea_conn, owner, repo).json() - branch = repo_data["default_branch"] + repo_obj = gitea_api.Repo.get(gitea_conn, owner, repo) + branch = repo_obj.default_branch fork_branch = branch # check if the scmsync branch exists in the source repo - parent_branch_data = gitea_api.Branch.get(gitea_conn, owner, repo, fork_branch).json() + parent_branch_obj = gitea_api.Branch.get(gitea_conn, owner, repo, fork_branch) try: - repo_data = gitea_api.Fork.create(gitea_conn, owner, repo, new_repo_name=args.new_repo_name).json() - fork_owner = repo_data["owner"]["login"] - fork_repo = repo_data["name"] + repo_obj = gitea_api.Fork.create(gitea_conn, owner, repo, new_repo_name=args.new_repo_name) + fork_owner = repo_obj.owner + fork_repo = repo_obj.repo print(f" * Fork created: {fork_owner}/{fork_repo}") except gitea_api.ForkExists as e: fork_owner = e.fork_owner @@ -196,10 +196,10 @@ class ForkCommand(osc.commandline.OscCommand): print(f" * scmsync URL: {fork_scmsync}") # check if the scmsync branch exists in the forked repo - fork_branch_data = gitea_api.Branch.get(gitea_conn, fork_owner, fork_repo, fork_branch).json() + fork_branch_obj = gitea_api.Branch.get(gitea_conn, fork_owner, fork_repo, fork_branch) - parent_commit = parent_branch_data["commit"]["id"] - fork_commit = fork_branch_data["commit"]["id"] + parent_commit = parent_branch_obj.commit + fork_commit = fork_branch_obj.commit if parent_commit != fork_commit: print() print(f"{tty.colorize('ERROR', 'red,bold')}: The branch in the forked repo is out of sync with the parent") diff --git a/osc/commands_git/login_add.py b/osc/commands_git/login_add.py index 0cbfc455..17464479 100644 --- a/osc/commands_git/login_add.py +++ b/osc/commands_git/login_add.py @@ -38,8 +38,15 @@ class LoginAddCommand(osc.commandline_git.GitObsCommand): if not re.match(r"^[0-9a-f]{40}$", args.token): self.parser.error("Invalid token format, 40 hexadecimal characters expected") - login = gitea_api.Login(name=args.name, url=args.url, user=args.user, token=args.token, ssh_key=args.ssh_key, default=args.set_as_default) - self.gitea_conf.add_login(login) + login_obj = gitea_api.Login( + name=args.name, + url=args.url, + user=args.user, + token=args.token, + ssh_key=args.ssh_key, + default=args.set_as_default, + ) + self.gitea_conf.add_login(login_obj) print("Added entry:") - print(login.to_human_readable_string()) + print(login_obj.to_human_readable_string()) diff --git a/osc/commands_git/login_list.py b/osc/commands_git/login_list.py index 49b0a811..ce7699af 100644 --- a/osc/commands_git/login_list.py +++ b/osc/commands_git/login_list.py @@ -13,6 +13,6 @@ class LoginListCommand(osc.commandline_git.GitObsCommand): self.parser.add_argument("--show-tokens", action="store_true", help="Show tokens in the output") def run(self, args): - for login in self.gitea_conf.list_logins(): - print(login.to_human_readable_string(show_token=args.show_tokens)) + for login_obj in self.gitea_conf.list_logins(): + print(login_obj.to_human_readable_string(show_token=args.show_tokens)) print() diff --git a/osc/commands_git/login_remove.py b/osc/commands_git/login_remove.py index 23ea80b3..0716905f 100644 --- a/osc/commands_git/login_remove.py +++ b/osc/commands_git/login_remove.py @@ -21,7 +21,7 @@ class LoginRemoveCommand(osc.commandline_git.GitObsCommand): print(f" * Config path: {self.gitea_conf.path}", file=sys.stderr) print("", file=sys.stderr) - login = self.gitea_conf.remove_login(args.name) + login_obj = self.gitea_conf.remove_login(args.name) print("Removed entry:") - print(login.to_human_readable_string()) + print(login_obj.to_human_readable_string()) diff --git a/osc/commands_git/login_update.py b/osc/commands_git/login_update.py index 5497aac8..a27653d6 100644 --- a/osc/commands_git/login_update.py +++ b/osc/commands_git/login_update.py @@ -31,19 +31,19 @@ class LoginUpdateCommand(osc.commandline_git.GitObsCommand): # TODO: try to authenticate to verify that the updated entry works - original_login = self.gitea_conf.get_login(args.name) + original_login_obj = self.gitea_conf.get_login(args.name) print("Original entry:") - print(original_login.to_human_readable_string()) + print(original_login_obj.to_human_readable_string()) if args.new_token == "-": print(file=sys.stderr) while not args.new_token or args.new_token == "-": - args.new_token = getpass.getpass(prompt=f"Enter a new Gitea token for user '{args.new_user or original_login.user}': ") + args.new_token = getpass.getpass(prompt=f"Enter a new Gitea token for user '{args.new_user or original_login_obj.user}': ") if not re.match(r"^[0-9a-f]{40}$", args.new_token): self.parser.error("Invalid token format, 40 hexadecimal characters expected") - updated_login = self.gitea_conf.update_login( + updated_login_obj = self.gitea_conf.update_login( args.name, new_name=args.new_name, new_url=args.new_url, @@ -54,4 +54,4 @@ class LoginUpdateCommand(osc.commandline_git.GitObsCommand): ) print("") print("Updated entry:") - print(updated_login.to_human_readable_string()) + print(updated_login_obj.to_human_readable_string()) diff --git a/osc/commands_git/pr_checkout.py b/osc/commands_git/pr_checkout.py index f144ae42..55adad62 100644 --- a/osc/commands_git/pr_checkout.py +++ b/osc/commands_git/pr_checkout.py @@ -34,20 +34,16 @@ class PullRequestCheckoutCommand(osc.commandline_git.GitObsCommand): git = gitea_api.Git(".") owner, repo = git.get_owner_repo() - pr = gitea_api.PullRequest.get(self.gitea_conn, owner, repo, args.pull).json() - - head_ssh_url = pr["head"]["repo"]["ssh_url"] - head_owner = pr["head"]["repo"]["owner"]["login"] - head_branch = pr["head"]["ref"] + pr_obj = gitea_api.PullRequest.get(self.gitea_conn, owner, repo, args.pull) try: - git.add_remote(head_owner, head_ssh_url) + git.add_remote(pr_obj.head_owner, pr_obj.head_ssh_url) except subprocess.CalledProcessError as e: # TODO: check if the remote url matches if e.returncode != 3: # returncode 3 means that the remote exists; see `man git-remote` raise - git.fetch(head_owner) + git.fetch(pr_obj.head_owner) local_branch = git.fetch_pull_request(args.pull, force=args.force) @@ -55,9 +51,9 @@ class PullRequestCheckoutCommand(osc.commandline_git.GitObsCommand): git.set_config("lfs.remote.searchall", "1") # configure branch for `git push` - git.set_config(f"branch.{local_branch}.remote", head_owner) - git.set_config(f"branch.{local_branch}.pushRemote", head_owner) - git.set_config(f"branch.{local_branch}.merge", f"refs/heads/{head_branch}") + git.set_config(f"branch.{local_branch}.remote", pr_obj.head_owner) + git.set_config(f"branch.{local_branch}.pushRemote", pr_obj.head_owner) + git.set_config(f"branch.{local_branch}.merge", f"refs/heads/{pr_obj.head_branch}") # allow `git push` with no arguments to push to a remote branch that is named differently than the local branch git.set_config("push.default", "upstream") diff --git a/osc/commands_git/pr_create.py b/osc/commands_git/pr_create.py index ce85daad..236b0ede 100644 --- a/osc/commands_git/pr_create.py +++ b/osc/commands_git/pr_create.py @@ -142,7 +142,7 @@ class PullRequestCreateCommand(osc.commandline_git.GitObsCommand): git = gitea_api.Git(".") local_owner, local_repo = git.get_owner_repo() local_branch = git.current_branch - local_rev = git.get_branch_head(local_branch) + local_commit = git.get_branch_head(local_branch) # remote git repo - source if use_local_git: @@ -153,12 +153,12 @@ class PullRequestCreateCommand(osc.commandline_git.GitObsCommand): source_owner = args.source_owner source_repo = args.source_repo source_branch = args.source_branch - source_repo_data = gitea_api.Repo.get(self.gitea_conn, source_owner, source_repo).json() - source_branch_data = gitea_api.Branch.get(self.gitea_conn, source_owner, source_repo, source_branch).json() - source_rev = source_branch_data["commit"]["id"] + source_repo_obj = gitea_api.Repo.get(self.gitea_conn, source_owner, source_repo) + source_branch_obj = gitea_api.Branch.get(self.gitea_conn, source_owner, source_repo, source_branch) # remote git repo - target - target_owner, target_repo = source_repo_data["parent"]["full_name"].split("/") + target_owner = source_repo_obj.parent_obj.owner + target_repo = source_repo_obj.parent_obj.repo if args.target_branch: target_branch = args.target_branch @@ -168,21 +168,20 @@ class PullRequestCreateCommand(osc.commandline_git.GitObsCommand): else: target_branch = source_branch - target_branch_data = gitea_api.Branch.get(self.gitea_conn, target_owner, target_repo, target_branch).json() - target_rev = target_branch_data["commit"]["id"] + target_branch_obj = gitea_api.Branch.get(self.gitea_conn, target_owner, target_repo, target_branch) print("Creating a pull request ...", file=sys.stderr) if use_local_git: - print(f" * Local git: branch: {local_branch}, rev: {local_rev}", file=sys.stderr) - print(f" * Source: {source_owner}/{source_repo}, branch: {source_branch}, rev: {source_rev}", file=sys.stderr) - print(f" * Target: {target_owner}/{target_repo}, branch: {target_branch}, rev: {target_rev}", file=sys.stderr) + print(f" * Local git: branch: {local_branch}, commit: {local_commit}", file=sys.stderr) + print(f" * Source: {source_owner}/{source_repo}, branch: {source_branch_obj.name}, commit: {source_branch_obj.commit}", file=sys.stderr) + print(f" * Target: {target_owner}/{target_repo}, branch: {target_branch_obj.name}, commit: {target_branch_obj.commit}", file=sys.stderr) - if use_local_git and local_rev != source_rev: + if use_local_git and local_commit != source_branch_obj.commit: from osc.output import tty print(f"{tty.colorize('ERROR', 'red,bold')}: Local commit doesn't correspond with the latest commit in the remote source branch") sys.exit(1) - if source_rev == target_rev: + if source_branch_obj.commit == target_branch_obj.commit: from osc.output import tty print(f"{tty.colorize('ERROR', 'red,bold')}: Source and target are identical, make and push changes to the remote source repo first") sys.exit(1) @@ -221,7 +220,7 @@ class PullRequestCreateCommand(osc.commandline_git.GitObsCommand): title = title.strip() description = description.strip() - pull = gitea_api.PullRequest.create( + pr_obj = gitea_api.PullRequest.create( self.gitea_conn, target_owner=target_owner, target_repo=target_repo, @@ -231,8 +230,8 @@ class PullRequestCreateCommand(osc.commandline_git.GitObsCommand): source_branch=source_branch, title=title, description=description, - ).json() + ) print("", file=sys.stderr) print("Pull request created:", file=sys.stderr) - print(gitea_api.PullRequest.to_human_readable_string(pull)) + print(pr_obj.to_human_readable_string()) diff --git a/osc/commands_git/pr_get.py b/osc/commands_git/pr_get.py index cbb0d4fc..d4480d05 100644 --- a/osc/commands_git/pr_get.py +++ b/osc/commands_git/pr_get.py @@ -34,14 +34,14 @@ class PullRequestGetCommand(osc.commandline_git.GitObsCommand): failed_entries = [] for owner, repo, pull in args.owner_repo_pull: try: - pr = gitea_api.PullRequest.get(self.gitea_conn, owner, repo, int(pull)).json() + pr_obj = gitea_api.PullRequest.get(self.gitea_conn, owner, repo, int(pull)) num_entries += 1 except gitea_api.GiteaException as e: if e.status == 404: failed_entries.append(f"{owner}/{repo}#{pull}") continue raise - print(gitea_api.PullRequest.to_human_readable_string(pr)) + print(pr_obj.to_human_readable_string()) if args.patch: print("") diff --git a/osc/commands_git/pr_list.py b/osc/commands_git/pr_list.py index 35d771e5..362efa85 100644 --- a/osc/commands_git/pr_list.py +++ b/osc/commands_git/pr_list.py @@ -51,20 +51,19 @@ class PullRequestListCommand(osc.commandline_git.GitObsCommand): total_entries = 0 for owner, repo in args.owner_repo: - data = gitea_api.PullRequest.list(self.gitea_conn, owner, repo, state=args.state).json() + pr_obj_list = gitea_api.PullRequest.list(self.gitea_conn, owner, repo, state=args.state) if args.no_draft: - data = [i for i in data if not i["draft"]] + pr_obj_list = [i for i in pr_obj_list if not i.draft] if args.target_branches: - data = [i for i in data if i["base"]["ref"] in args.target_branches] - - review_states = args.review_states or ["REQUEST_REVIEW"] + pr_obj_list = [i for i in pr_obj_list if i.base_branch in args.target_branches] if args.reviewers: - new_data = [] - for entry in data: - all_reviews = gitea_api.PullRequest.get_reviews(self.gitea_conn, owner, repo, entry["number"]).json() + review_states = args.review_states or ["REQUEST_REVIEW"] + new_pr_obj_list = [] + for pr_obj in pr_obj_list: + all_reviews = gitea_api.PullRequest.get_reviews(self.gitea_conn, owner, repo, pr_obj.number).json() user_reviews = {i["user"]["login"]: i["state"] for i in all_reviews if i["user"] and i["state"] in review_states} team_reviews = {i["team"]["name"]: i["state"] for i in all_reviews if i["team"] and i["state"] in review_states} @@ -72,16 +71,15 @@ class PullRequestListCommand(osc.commandline_git.GitObsCommand): team_reviewers = [i[1:] for i in args.reviewers if i.startswith("@")] if set(user_reviews) & set(user_reviewers) or set(team_reviews) & set(team_reviewers): - print(set(user_reviews) & set(user_reviewers), set(team_reviews) & set(team_reviewers)) - new_data.append(entry) + new_pr_obj_list.append(pr_obj) - data = new_data + pr_obj_list = new_pr_obj_list - total_entries += len(data) - - text = gitea_api.PullRequest.list_to_human_readable_string(data, sort=True) - if text: - print(text) - print("", file=sys.stderr) + if pr_obj_list: + total_entries += len(pr_obj_list) + pr_obj_list.sort() + for pr_obj in pr_obj_list: + print(pr_obj.to_human_readable_string()) + print() print(f"Total entries: {total_entries}", file=sys.stderr) diff --git a/osc/commands_git/pr_review.py b/osc/commands_git/pr_review.py index f88504c0..fef88366 100644 --- a/osc/commands_git/pr_review.py +++ b/osc/commands_git/pr_review.py @@ -1,6 +1,7 @@ import os import subprocess import sys +from typing import Generator from typing import Optional import osc.commandline_git @@ -28,6 +29,7 @@ DECLINE_REVIEW_TEMPLATE = """ class PullRequestReviewCommand(osc.commandline_git.GitObsCommand): """ + Interactive review of pull requests """ name = "review" @@ -50,11 +52,10 @@ class PullRequestReviewCommand(osc.commandline_git.GitObsCommand): else: # keep only the list of pull request IDs, throw search results away # because the search returns issues instead of pull requests - data = gitea_api.PullRequest.search(self.gitea_conn, review_requested=True).json() - # TODO: priority ordering? - data = sorted(data, key=gitea_api.PullRequest.cmp) - pull_request_ids = [f"{i['repository']['full_name']}#{i['number']}" for i in data] - del data + pr_obj_list = gitea_api.PullRequest.search(self.gitea_conn, review_requested=True) + pr_obj_list.sort() + pull_request_ids = [pr_obj.id for pr_obj in pr_obj_list] + del pr_obj_list skipped_drafts = 0 @@ -62,15 +63,15 @@ class PullRequestReviewCommand(osc.commandline_git.GitObsCommand): self.print_gitea_settings() owner, repo, number = gitea_api.PullRequest.split_id(pr_id) - pr_data = gitea_api.PullRequest.get(self.gitea_conn, owner, repo, number).json() + pr_obj = gitea_api.PullRequest.get(self.gitea_conn, owner, repo, number) - if pr_data["draft"]: + if pr_obj.draft: # we don't want to review drafts, they will change skipped_drafts += 1 continue self.clone_git(owner, repo, number) - self.view(owner, repo, number, pr_index=pr_index, pr_count=len(pull_request_ids), pr_data=pr_data) + self.view(owner, repo, number, pr_index=pr_index, pr_count=len(pull_request_ids), pr_obj=pr_obj) while True: # TODO: print at least some context because the PR details disappear after closing less @@ -96,7 +97,7 @@ class PullRequestReviewCommand(osc.commandline_git.GitObsCommand): self.comment(owner, repo, number) break elif reply == "v": - self.view(owner, repo, number, pr_index=pr_index, pr_count=len(pull_request_ids), pr_data=pr_data) + self.view(owner, repo, number, pr_index=pr_index, pr_count=len(pull_request_ids), pr_obj=pr_obj) elif reply == "s": break elif reply == "x": @@ -110,6 +111,7 @@ class PullRequestReviewCommand(osc.commandline_git.GitObsCommand): def approve(self, owner: str, repo: str, number: int): from osc import gitea_api + gitea_api.PullRequest.approve_review(self.gitea_conn, owner, repo, number) def decline(self, owner: str, repo: str, number: int): @@ -148,8 +150,8 @@ class PullRequestReviewCommand(osc.commandline_git.GitObsCommand): def clone_git(self, owner: str, repo: str, number: int): from osc import gitea_api - repo_data = gitea_api.Repo.get(self.gitea_conn, owner, repo).json() - clone_url = repo_data["ssh_url"] + repo_obj = gitea_api.Repo.get(self.gitea_conn, owner, repo) + clone_url = repo_obj.ssh_url # TODO: it might be good to have a central cache for the git repos to speed cloning up path = self.get_git_repo_path(owner, repo, number) @@ -161,14 +163,23 @@ class PullRequestReviewCommand(osc.commandline_git.GitObsCommand): git.clone(clone_url, directory=path, quiet=False) git.fetch_pull_request(number, force=True) - def view(self, owner: str, repo: str, number: int, *, pr_index: int, pr_count: int, pr_data: Optional[dict] = None): + def view( + self, + owner: str, + repo: str, + number: int, + *, + pr_index: int, + pr_count: int, + pr_obj: Optional["PullRequest"] = None, + ): from osc import gitea_api from osc.core import highlight_diff from osc.output import sanitize_text from osc.output import tty - if pr_data is None: - pr_data = gitea_api.PullRequest.get(self.gitea_conn, owner, repo, number).json() + if pr_obj is None: + pr_obj = gitea_api.PullRequest.get(self.gitea_conn, owner, repo, number) # the process works with bytes rather than with strings # because the diffs may contain character sequences that cannot be decoded as utf-8 strings @@ -181,14 +192,13 @@ class PullRequestReviewCommand(osc.commandline_git.GitObsCommand): proc.stdin.write(b"\n") # pr details - pr = gitea_api.PullRequest.to_human_readable_string(pr_data) - proc.stdin.write(pr.encode("utf-8")) + proc.stdin.write(pr_obj.to_human_readable_string().encode("utf-8")) proc.stdin.write(b"\n") proc.stdin.write(b"\n") # patch proc.stdin.write(tty.colorize("Patch:\n", "bold").encode("utf-8")) - patch = gitea_api.PullRequest.get_patch(self.gitea_conn, owner, repo, number).data + patch = gitea_api.PullRequest.get_patch(self.gitea_conn, owner, repo, number) patch = sanitize_text(patch) patch = highlight_diff(patch) proc.stdin.write(patch) @@ -196,7 +206,7 @@ class PullRequestReviewCommand(osc.commandline_git.GitObsCommand): # tardiff proc.stdin.write(tty.colorize("Archive diffs:\n", "bold").encode("utf-8")) - tardiff_chunks = self.tardiff(owner, repo, number, pr_data=pr_data) + tardiff_chunks = self.tardiff(owner, repo, number, pr_obj=pr_obj) for chunk in tardiff_chunks: chunk = sanitize_text(chunk) chunk = highlight_diff(chunk) @@ -213,20 +223,17 @@ class PullRequestReviewCommand(osc.commandline_git.GitObsCommand): path = os.path.expanduser(path) return path - def tardiff(self, owner: str, repo: str, number: int, *, pr_data: dict): + def tardiff(self, owner: str, repo: str, number: int, *, pr_obj: ".PullRequest") -> Generator[bytes, None, None]: from osc import gitea_api - src_commit = pr_data["head"]["sha"] - dst_commit = pr_data["base"]["sha"] - path = self.get_git_repo_path(owner, repo, number) git = gitea_api.Git(path) # the repo might be outdated, make sure the commits are available git.fetch() - src_archives = git.lfs_ls_files(ref=src_commit) - dst_archives = git.lfs_ls_files(ref=dst_commit) + src_archives = git.lfs_ls_files(ref=pr_obj.head_commit) + dst_archives = git.lfs_ls_files(ref=pr_obj.base_commit) def map_archives_by_name(archives: list): result = {} @@ -248,10 +255,10 @@ class PullRequestReviewCommand(osc.commandline_git.GitObsCommand): dst_archive = dst_archives_by_name.get(name, (None, None)) if src_archive[0]: - td.add_archive(src_archive[0], src_archive[1], git.lfs_cat_file(src_archive[0], ref=src_commit)) + td.add_archive(src_archive[0], src_archive[1], git.lfs_cat_file(src_archive[0], ref=pr_obj.head_commit)) if dst_archive[0]: - td.add_archive(dst_archive[0], dst_archive[1], git.lfs_cat_file(dst_archive[0], ref=dst_commit)) + td.add_archive(dst_archive[0], dst_archive[1], git.lfs_cat_file(dst_archive[0], ref=pr_obj.base_commit)) # TODO: max output length / max lines; in such case, it would be great to list all the changed files at least yield from td.diff_archives(*dst_archive, *src_archive) diff --git a/osc/commands_git/pr_search.py b/osc/commands_git/pr_search.py index b5a0c581..ee02ddc9 100644 --- a/osc/commands_git/pr_search.py +++ b/osc/commands_git/pr_search.py @@ -59,7 +59,7 @@ class PullRequestSearchCommand(osc.commandline_git.GitObsCommand): self.print_gitea_settings() - data = gitea_api.PullRequest.search( + pr_obj_list = gitea_api.PullRequest.search( self.gitea_conn, state=args.state, title=args.title, @@ -69,11 +69,12 @@ class PullRequestSearchCommand(osc.commandline_git.GitObsCommand): created=args.created, mentioned=args.mentioned, review_requested=args.review_requested, - ).json() + ) - text = gitea_api.PullRequest.list_to_human_readable_string(data, sort=True) - if text: - print(text) - print("", file=sys.stderr) + if pr_obj_list: + pr_obj_list.sort() + for pr_obj in pr_obj_list: + print(pr_obj.to_human_readable_string()) + print() - print(f"Total entries: {len(data)}", file=sys.stderr) + print(f"Total entries: {len(pr_obj_list)}", file=sys.stderr) diff --git a/osc/commands_git/pr_set.py b/osc/commands_git/pr_set.py index 13279f2d..32babbf1 100644 --- a/osc/commands_git/pr_set.py +++ b/osc/commands_git/pr_set.py @@ -37,13 +37,12 @@ class PullRequestSetCommand(osc.commandline_git.GitObsCommand): from osc.output import tty self.print_gitea_settings() - print(args) num_entries = 0 failed_entries = [] for owner, repo, pull in args.owner_repo_pull: try: - pr = gitea_api.PullRequest.set( + pr_obj = gitea_api.PullRequest.set( self.gitea_conn, owner, repo, @@ -51,7 +50,7 @@ class PullRequestSetCommand(osc.commandline_git.GitObsCommand): title=args.title, description=args.description, allow_maintainer_edit=args.allow_maintainer_edit, - ).json() + ) num_entries += 1 except gitea_api.GiteaException as e: if e.status == 404: @@ -59,7 +58,7 @@ class PullRequestSetCommand(osc.commandline_git.GitObsCommand): continue raise - print(gitea_api.PullRequest.to_human_readable_string(pr)) + print(pr_obj.to_human_readable_string()) print() print(f"Total modified entries: {num_entries}", file=sys.stderr) diff --git a/osc/commands_git/repo_fork.py b/osc/commands_git/repo_fork.py index 8d90e2cd..217ac71e 100644 --- a/osc/commands_git/repo_fork.py +++ b/osc/commands_git/repo_fork.py @@ -29,10 +29,9 @@ class RepoForkCommand(osc.commandline_git.GitObsCommand): for owner, repo in args.owner_repo: print(f"Forking git repo {owner}/{repo} ...", file=sys.stderr) try: - response = gitea_api.Fork.create(self.gitea_conn, owner, repo, new_repo_name=args.new_repo_name) - repo = response.json() - fork_owner = repo["owner"]["login"] - fork_repo = repo["name"] + repo_obj = gitea_api.Fork.create(self.gitea_conn, owner, repo, new_repo_name=args.new_repo_name) + fork_owner = repo_obj.owner + fork_repo = repo_obj.repo print(f" * Fork created: {fork_owner}/{fork_repo}", file=sys.stderr) num_entries += 1 except gitea_api.ForkExists as e: diff --git a/osc/commands_git/ssh_key_add.py b/osc/commands_git/ssh_key_add.py index e441154a..54e17652 100644 --- a/osc/commands_git/ssh_key_add.py +++ b/osc/commands_git/ssh_key_add.py @@ -35,6 +35,6 @@ class SSHKeyAddCommand(osc.commandline_git.GitObsCommand): with open(os.path.expanduser(args.key_path)) as f: key = f.read().strip() - response = gitea_api.SSHKey.create(self.gitea_conn, key) + ssh_key_obj = gitea_api.SSHKey.create(self.gitea_conn, key) print("Added entry:") - print(gitea_api.SSHKey.to_human_readable_string(response.json())) + print(ssh_key_obj.to_human_readable_string()) diff --git a/osc/commands_git/ssh_key_list.py b/osc/commands_git/ssh_key_list.py index 1dcd7b97..585cdcbd 100644 --- a/osc/commands_git/ssh_key_list.py +++ b/osc/commands_git/ssh_key_list.py @@ -16,6 +16,7 @@ class SSHKeyListCommand(osc.commandline_git.GitObsCommand): self.print_gitea_settings() - for i in gitea_api.SSHKey.list(self.gitea_conn).json(): - print(gitea_api.SSHKey.to_human_readable_string(i)) + ssh_key_obj_list = gitea_api.SSHKey.list(self.gitea_conn) + for ssh_key_obj in ssh_key_obj_list: + print(ssh_key_obj.to_human_readable_string()) print() diff --git a/osc/commands_git/ssh_key_remove.py b/osc/commands_git/ssh_key_remove.py index ad58c522..5c5456b9 100644 --- a/osc/commands_git/ssh_key_remove.py +++ b/osc/commands_git/ssh_key_remove.py @@ -23,8 +23,8 @@ class SSHKeyRemoveCommand(osc.commandline_git.GitObsCommand): self.print_gitea_settings() print(f"Removing ssh key with id='{args.id}' ...", file=sys.stderr) - response = gitea_api.SSHKey.get(self.gitea_conn, args.id) + ssh_key_obj = gitea_api.SSHKey.get(self.gitea_conn, args.id) gitea_api.SSHKey.delete(self.gitea_conn, args.id) print("Removed entry:") - print(gitea_api.SSHKey.to_human_readable_string(response.json())) + print(ssh_key_obj.to_human_readable_string()) diff --git a/osc/gitea_api/branch.py b/osc/gitea_api/branch.py index cc75da4b..b5156e92 100644 --- a/osc/gitea_api/branch.py +++ b/osc/gitea_api/branch.py @@ -1,13 +1,24 @@ +from typing import List from typing import Optional from .connection import Connection from .connection import GiteaHTTPResponse -from .exceptions import BranchDoesNotExist from .exceptions import BranchExists -from .exceptions import GiteaException class Branch: + def __init__(self, data: dict, *, response: Optional[GiteaHTTPResponse] = None): + self._data = data + self._response = response + + @property + def commit(self) -> str: + return self._data["commit"]["id"] + + @property + def name(self) -> str: + return self._data["name"] + @classmethod def get( cls, @@ -15,7 +26,7 @@ class Branch: owner: str, repo: str, branch: str, - ) -> GiteaHTTPResponse: + ) -> "Branch": """ Retrieve details about a repository branch. @@ -25,7 +36,9 @@ class Branch: :param branch: Name of the branch. """ url = conn.makeurl("repos", owner, repo, "branches", branch) - return conn.request("GET", url, context={"owner": owner, "repo": repo}) + response = conn.request("GET", url, context={"owner": owner, "repo": repo}) + obj = cls(response.json(), response=response) + return obj @classmethod def list( @@ -33,7 +46,7 @@ class Branch: conn: Connection, owner: str, repo: str, - ) -> GiteaHTTPResponse: + ) -> List["Branch"]: """ Retrieve details about all repository branches. @@ -46,7 +59,9 @@ class Branch: } url = conn.makeurl("repos", owner, repo, "branches", query=q) # XXX: returns 'null' when there are no branches; an empty list would be a better API - return conn.request("GET", url) + response = conn.request("GET", url) + obj_list = [cls(i, response=response) for i in response.json() or []] + return obj_list @classmethod def create( @@ -58,7 +73,7 @@ class Branch: old_ref_name: Optional[str] = None, new_branch_name: str, exist_ok: bool = False, - ) -> GiteaHTTPResponse: + ) -> "Branch": """ Create a new branch in a repository. @@ -75,8 +90,10 @@ class Branch: } url = conn.makeurl("repos", owner, repo, "branches") try: - return conn.request("POST", url, json_data=json_data, context={"owner": owner, "repo": repo, "branch": new_branch_name}) - except BranchExists as e: + response = conn.request("POST", url, json_data=json_data, context={"owner": owner, "repo": repo, "branch": new_branch_name}) + obj = cls(response.json(), response=response) + return obj + except BranchExists: if not exist_ok: raise return cls.get(conn, owner, repo, new_branch_name) diff --git a/osc/gitea_api/fork.py b/osc/gitea_api/fork.py index 911923a4..1fe20605 100644 --- a/osc/gitea_api/fork.py +++ b/osc/gitea_api/fork.py @@ -1,8 +1,9 @@ +from typing import List from typing import Optional from .connection import Connection -from .connection import GiteaHTTPResponse from .exceptions import ForkExists +from .repo import Repo class Fork: @@ -12,7 +13,7 @@ class Fork: conn: Connection, owner: str, repo: str, - ) -> GiteaHTTPResponse: + ) -> List["Repo"]: """ List forks of a repository. @@ -24,7 +25,9 @@ class Fork: "limit": -1, } url = conn.makeurl("repos", owner, repo, "forks", query=q) - return conn.request("GET", url) + response = conn.request("GET", url) + obj_list = [Repo(i, response=response) for i in response.json()] + return obj_list @classmethod def create( @@ -36,7 +39,7 @@ class Fork: new_repo_name: Optional[str] = None, target_org: Optional[str] = None, exist_ok: bool = False, - ) -> GiteaHTTPResponse: + ) -> Repo: """ Fork a repository. @@ -47,17 +50,16 @@ class Fork: :param target_org: Name of the organization, if forking into organization. :param exist_ok: A ``ForkExists`` exception is raised when the target exists. Set to ``True`` to avoid throwing the exception. """ - json_data = { "name": new_repo_name, "organization": target_org, } url = conn.makeurl("repos", owner, repo, "forks") try: - return conn.request("POST", url, json_data=json_data) + response = conn.request("POST", url, json_data=json_data) + obj = Repo(response.json(), response=response) + return obj except ForkExists as e: if not exist_ok: raise - - from . import Repo # pylint: disable=import-outside-toplevel return Repo.get(conn, e.fork_owner, e.fork_repo) diff --git a/osc/gitea_api/pr.py b/osc/gitea_api/pr.py index f9a929f8..78419318 100644 --- a/osc/gitea_api/pr.py +++ b/osc/gitea_api/pr.py @@ -1,3 +1,4 @@ +import functools import re from typing import List from typing import Optional @@ -5,17 +6,20 @@ from typing import Tuple from .connection import Connection from .connection import GiteaHTTPResponse +from .user import User +@functools.total_ordering class PullRequest: - @classmethod - def cmp(cls, entry: dict): - if "base" in entry: - # a proper pull request - return entry["base"]["repo"]["full_name"], entry["number"] - else: - # an issue without pull request details - return entry["repository"]["full_name"], entry["number"] + def __init__(self, data: dict, *, response: Optional[GiteaHTTPResponse] = None): + self._data = data + self._response = response + + def __eq__(self, other): + (self.base_owner, self.base_repo, self.number) == (other.base_owner, other.base_repo, other.number) + + def __lt__(self, other): + (self.base_owner, self.base_repo, self.number) < (other.base_owner, other.base_repo, other.number) @classmethod def split_id(cls, pr_id: str) -> Tuple[str, str, int]: @@ -27,52 +31,149 @@ class PullRequest: raise ValueError(f"Invalid pull request id: {pr_id}") return match.group(1), match.group(2), int(match.group(3)) - @classmethod - def to_human_readable_string(cls, entry: dict): + @property + def is_pull_request(self): + # determine if we're working with a proper pull request or an issue without pull request details + return "base" in self._data + + @property + def id(self) -> str: + return f"{self.base_owner}/{self.base_repo}#{self.number}" + + @property + def number(self) -> int: + return self._data["number"] + + @property + def title(self) -> str: + return self._data["title"] + + @property + def body(self) -> str: + return self._data["body"] + + @property + def state(self) -> str: + return self._data["state"] + + @property + def user(self) -> str: + return self._data["user"]["login"] + + @property + def user_obj(self) -> User: + return User(self._data["user"]) + + @property + def draft(self) -> Optional[bool]: + if not self.is_pull_request: + return None + return self._data["draft"] + + @property + def merged(self) -> Optional[bool]: + if not self.is_pull_request: + return None + return self._data["merged"] + + @property + def allow_maintainer_edit(self) -> Optional[bool]: + if not self.is_pull_request: + return None + return self._data["allow_maintainer_edit"] + + @property + def base_owner(self) -> Optional[str]: + if not self.is_pull_request: + return self._data["repository"]["owner"] + return self._data["base"]["repo"]["owner"]["login"] + + @property + def base_repo(self) -> str: + if not self.is_pull_request: + return self._data["repository"]["name"] + return self._data["base"]["repo"]["name"] + + @property + def base_branch(self) -> Optional[str]: + if not self.is_pull_request: + return None + return self._data["base"]["ref"] + + @property + def base_commit(self) -> Optional[str]: + if not self.is_pull_request: + return None + return self._data["base"]["sha"] + + @property + def base_ssh_url(self) -> Optional[str]: + if not self.is_pull_request: + return None + return self._data["base"]["repo"]["ssh_url"] + + @property + def head_owner(self) -> Optional[str]: + if not self.is_pull_request: + return None + return self._data["head"]["repo"]["owner"]["login"] + + @property + def head_repo(self) -> Optional[str]: + if not self.is_pull_request: + return None + return self._data["head"]["repo"]["name"] + + @property + def head_branch(self) -> Optional[str]: + if not self.is_pull_request: + return None + return self._data["head"]["ref"] + + @property + def head_commit(self) -> Optional[str]: + if not self.is_pull_request: + return None + return self._data["head"]["sha"] + + @property + def head_ssh_url(self) -> Optional[str]: + if not self.is_pull_request: + return None + return self._data["head"]["repo"]["ssh_url"] + + @property + def url(self) -> str: + # HACK: search API returns issues, the URL needs to be transformed to a pull request URL + return re.sub(r"^(.*)/api/v1/repos/(.+/.+)/issues/([0-9]+)$", r"\1/\2/pulls/\3", self._data["url"]) + + def to_human_readable_string(self): from osc.output import KeyValueTable - from . import User def yes_no(value): return "yes" if value else "no" - if "base" in entry: - # a proper pull request - entry_id = f"{entry['base']['repo']['full_name']}#{entry['number']}" - is_pull_request = True - else: - # an issue without pull request details - entry_id = f"{entry['repository']['full_name']}#{entry['number']}" - is_pull_request = False - - # HACK: search API returns issues, the URL needs to be transformed to a pull request URL - entry_url = entry["url"] - entry_url = re.sub(r"^(.*)/api/v1/repos/(.+/.+)/issues/([0-9]+)$", r"\1/\2/pulls/\3", entry_url) - table = KeyValueTable() - table.add("ID", entry_id, color="bold") - table.add("URL", f"{entry_url}") - table.add("Title", f"{entry['title']}") - table.add("State", entry["state"]) - if is_pull_request: - table.add("Draft", yes_no(entry["draft"])) - table.add("Merged", yes_no(entry["merged"])) - table.add("Allow edit", yes_no(entry["allow_maintainer_edit"])) - table.add("Author", f"{User.to_login_full_name_email_string(entry['user'])}") - if is_pull_request: - table.add("Source", f"{entry['head']['repo']['full_name']}, branch: {entry['head']['ref']}, commit: {entry['head']['sha']}") - table.add("Description", entry["body"]) + table.add("ID", self.id, color="bold") + table.add("URL", self.url) + table.add("Title", self.title) + table.add("State", self.state) + if self.is_pull_request: + table.add("Draft", yes_no(self.draft)) + table.add("Merged", yes_no(self.merged)) + table.add("Allow edit", yes_no(self.allow_maintainer_edit)) + table.add("Author", f"{self.user_obj.login_full_name_email}") + if self.is_pull_request: + table.add( + "Source", f"{self.head_owner}/{self.head_repo}, branch: {self.head_branch}, commit: {self.head_commit}" + ) + table.add( + "Target", f"{self.base_owner}/{self.base_repo}, branch: {self.base_branch}, commit: {self.base_commit}" + ) + table.add("Description", self.body) return str(table) - @classmethod - def list_to_human_readable_string(cls, entries: List, sort: bool = False): - if sort: - entries = sorted(entries, key=cls.cmp) - result = [] - for entry in entries: - result.append(cls.to_human_readable_string(entry)) - return "\n\n".join(result) - @classmethod def create( cls, @@ -85,7 +186,7 @@ class PullRequest: source_branch: str, title: str, description: Optional[str] = None, - ) -> GiteaHTTPResponse: + ) -> "PullRequest": """ Create a pull request to ``owner``/``repo`` to the ``base`` branch. The pull request comes from a fork. The fork repo name is determined from gitea database. @@ -106,7 +207,9 @@ class PullRequest: "title": title, "body": description, } - return conn.request("POST", url, json_data=data) + response = conn.request("POST", url, json_data=data) + obj = cls(response.json(), response=response) + return obj @classmethod def get( @@ -115,7 +218,7 @@ class PullRequest: owner: str, repo: str, number: int, - ) -> GiteaHTTPResponse: + ) -> "PullRequest": """ Get a pull request. @@ -125,7 +228,9 @@ class PullRequest: :param number: Number of the pull request in the repo. """ url = conn.makeurl("repos", owner, repo, "pulls", str(number)) - return conn.request("GET", url) + response = conn.request("GET", url) + obj = cls(response.json(), response=response) + return obj @classmethod def set( @@ -138,7 +243,7 @@ class PullRequest: title: Optional[str] = None, description: Optional[str] = None, allow_maintainer_edit: Optional[bool] = None, - ) -> GiteaHTTPResponse: + ) -> "PullRequest": """ Change a pull request. @@ -156,7 +261,9 @@ class PullRequest: "allow_maintainer_edit": allow_maintainer_edit, } url = conn.makeurl("repos", owner, repo, "pulls", str(number)) - return conn.request("PATCH", url, json_data=json_data) + response = conn.request("PATCH", url, json_data=json_data) + obj = cls(response.json(), response=response) + return obj @classmethod def list( @@ -166,7 +273,7 @@ class PullRequest: repo: str, *, state: Optional[str] = "open", - ) -> GiteaHTTPResponse: + ) -> List["PullRequest"]: """ List pull requests in a repo. @@ -183,7 +290,9 @@ class PullRequest: "limit": -1, } url = conn.makeurl("repos", owner, repo, "pulls", query=q) - return conn.request("GET", url) + response = conn.request("GET", url) + obj_list = [cls(i, response=response) for i in response.json()] + return obj_list @classmethod def search( @@ -198,7 +307,7 @@ class PullRequest: created: bool = False, mentioned: bool = False, review_requested: bool = False, - ) -> GiteaHTTPResponse: + ) -> List["PullRequest"]: """ Search pull requests. :param conn: Gitea ``Connection`` instance. @@ -225,7 +334,9 @@ class PullRequest: "limit": 10**6, } url = conn.makeurl("repos", "issues", "search", query=q) - return conn.request("GET", url) + response = conn.request("GET", url) + obj_list = [cls(i, response=response) for i in response.json()] + return obj_list @classmethod def get_patch( @@ -234,7 +345,7 @@ class PullRequest: owner: str, repo: str, number: int, - ) -> GiteaHTTPResponse: + ) -> "bytes": """ Get a patch associated with a pull request. @@ -244,7 +355,8 @@ class PullRequest: :param number: Number of the pull request in the repo. """ url = conn.makeurl("repos", owner, repo, "pulls", f"{number}.patch") - return conn.request("GET", url) + response = conn.request("GET", url) + return response.data @classmethod def add_comment( diff --git a/osc/gitea_api/repo.py b/osc/gitea_api/repo.py index 87849edb..2d32b533 100644 --- a/osc/gitea_api/repo.py +++ b/osc/gitea_api/repo.py @@ -10,6 +10,40 @@ from .user import User class Repo: + def __init__(self, data: dict, *, response: Optional[GiteaHTTPResponse] = None): + self._data = data + self._response = response + + @property + def owner(self) -> str: + return self._data["owner"]["login"] + + @property + def owner_obj(self) -> User: + return User(self._data["owner"]) + + @property + def repo(self) -> str: + return self._data["name"] + + @property + def parent_obj(self) -> Optional["Repo"]: + if not self._data["parent"]: + return None + return Repo(self._data["parent"]) + + @property + def clone_url(self) -> str: + return self._data["clone_url"] + + @property + def ssh_url(self) -> str: + return self._data["ssh_url"] + + @property + def default_branch(self) -> str: + return self._data["default_branch"] + @classmethod def split_id(cls, repo_id: str) -> Tuple[str, str]: """ @@ -26,7 +60,7 @@ class Repo: conn: Connection, owner: str, repo: str, - ) -> GiteaHTTPResponse: + ) -> "Repo": """ Retrieve details about a repository. @@ -35,7 +69,9 @@ class Repo: :param repo: Name of the repo. """ url = conn.makeurl("repos", owner, repo) - return conn.request("GET", url) + response = conn.request("GET", url) + obj = cls(response.json(), response=response) + return obj @classmethod def clone( @@ -71,26 +107,27 @@ class Repo: # it's perfectly fine to use os.path.join() here because git can take an absolute path directory_abspath = os.path.join(cwd, directory) - repo_data = cls.get(conn, owner, repo).json() - clone_url = repo_data["clone_url"] if anonymous else repo_data["ssh_url"] + repo_obj = cls.get(conn, owner, repo) + + clone_url = repo_obj.clone_url if anonymous else repo_obj.ssh_url remotes = {} if add_remotes: - user = User.get(conn).json() - if repo_data["owner"]["login"] == user["login"]: + user_obj = User.get(conn) + if repo_obj.owner == user_obj.login: # we're cloning our own repo, setting remote to the parent (if exists) - parent = repo_data["parent"] - if parent: - remotes["parent"] = parent["clone_url"] if anonymous else parent["ssh_url"] + if repo_obj.parent_obj: + remotes["parent"] = repo_obj.parent_obj.clone_url if anonymous else repo_obj.parent_obj.ssh_url else: # we're cloning someone else's repo, setting remote to our fork (if exists) from . import Fork - forks = Fork.list(conn, owner, repo).json() - forks = [i for i in forks if i["owner"]["login"] == user["login"]] - if forks: - assert len(forks) == 1 - fork = forks[0] - remotes["fork"] = fork["clone_url"] if anonymous else fork["ssh_url"] + + fork_obj_list = Fork.list(conn, owner, repo) + fork_obj_list = [fork_obj for fork_obj in fork_obj_list if fork_obj.owner == user_obj.login] + if fork_obj_list: + assert len(fork_obj_list) == 1 + fork_obj = fork_obj_list[0] + remotes["fork"] = fork_obj.clone_url if anonymous else fork_obj.ssh_url ssh_args = [] env = os.environ.copy() @@ -131,7 +168,14 @@ class Repo: # store used ssh args (GIT_SSH_COMMAND) in the local git config # to allow seamlessly running ``git push`` and other commands if ssh_args: - cmd = ["git", "-C", directory_abspath, "config", "core.sshCommand", f"echo 'Using core.sshCommand: {env['GIT_SSH_COMMAND']}' >&2; {env['GIT_SSH_COMMAND']}"] + cmd = [ + "git", + "-C", + directory_abspath, + "config", + "core.sshCommand", + f"echo 'Using core.sshCommand: {env['GIT_SSH_COMMAND']}' >&2; {env['GIT_SSH_COMMAND']}", + ] subprocess.run(cmd, cwd=cwd, check=True) return directory_abspath diff --git a/osc/gitea_api/ssh_key.py b/osc/gitea_api/ssh_key.py index 4a0c2236..7124b5c7 100644 --- a/osc/gitea_api/ssh_key.py +++ b/osc/gitea_api/ssh_key.py @@ -1,3 +1,4 @@ +from typing import List from typing import Optional from .connection import Connection @@ -5,8 +6,24 @@ from .connection import GiteaHTTPResponse class SSHKey: + def __init__(self, data: dict, *, response: Optional[GiteaHTTPResponse] = None): + self._data = data + self._response = response + + @property + def id(self) -> int: + return self._data["id"] + + @property + def key(self) -> str: + return self._data["key"] + + @property + def title(self) -> str: + return self._data["title"] + @classmethod - def get(cls, conn: Connection, id: int) -> GiteaHTTPResponse: + def get(cls, conn: Connection, id: int) -> "SSHKey": """ Get an authenticated user's public key by its ``id``. @@ -14,10 +31,12 @@ class SSHKey: :param id: key numeric id """ url = conn.makeurl("user", "keys", str(id)) - return conn.request("GET", url) + response = conn.request("GET", url) + obj = cls(response.json(), response=response) + return obj @classmethod - def list(cls, conn: Connection) -> GiteaHTTPResponse: + def list(cls, conn: Connection) -> List["SSHKey"]: """ List the authenticated user's public keys. @@ -27,11 +46,14 @@ class SSHKey: "limit": -1, } url = conn.makeurl("user", "keys", query=q) - return conn.request("GET", url) + response = conn.request("GET", url) + obj_list = [cls(i, response=response) for i in response.json()] + return obj_list @classmethod def _split_key(cls, key): import re + return re.split(" +", key, maxsplit=2) @classmethod @@ -61,7 +83,7 @@ class SSHKey: raise InvalidSshPublicKey() @classmethod - def create(cls, conn: Connection, key: str, title: Optional[str] = None) -> GiteaHTTPResponse: + def create(cls, conn: Connection, key: str, title: Optional[str] = None) -> "SSHKey": """ Create a public key. @@ -80,10 +102,12 @@ class SSHKey: "key": key, "title": title, } - return conn.request("POST", url, json_data=data) + response = conn.request("POST", url, json_data=data) + obj = cls(response.json(), response=response) + return obj @classmethod - def delete(cls, conn: Connection, id: int): + def delete(cls, conn: Connection, id: int) -> GiteaHTTPResponse: """ Delete a public key @@ -94,11 +118,11 @@ class SSHKey: url = conn.makeurl("user", "keys", str(id)) return conn.request("DELETE", url) - @classmethod - def to_human_readable_string(cls, data): + def to_human_readable_string(self) -> str: from osc.output import KeyValueTable + table = KeyValueTable() - table.add("ID", f"{data['id']}", color="bold") - table.add("Title", f"{data['title']}") - table.add("Key", f"{data['key']}") + table.add("ID", f"{self.id}", color="bold") + table.add("Title", f"{self.title}") + table.add("Key", f"{self.key}") return str(table) diff --git a/osc/gitea_api/user.py b/osc/gitea_api/user.py index 01c08678..cc9677c2 100644 --- a/osc/gitea_api/user.py +++ b/osc/gitea_api/user.py @@ -1,29 +1,47 @@ +from typing import Optional + from .connection import Connection from .connection import GiteaHTTPResponse class User: - @classmethod - def to_full_name_email_string(cls, data): - full_name = data["full_name"] - email = data["email"] - if full_name: - return f"{full_name} <{email}>" - return email + def __init__(self, data: dict, *, response: Optional[GiteaHTTPResponse] = None): + self._data = data + self._response = response - @classmethod - def to_login_full_name_email_string(cls, data): - return f"{data['login']} ({cls.to_full_name_email_string(data)})" + @property + def login(self) -> str: + return self._data["login"] + + @property + def full_name(self) -> str: + return self._data["full_name"] + + @property + def email(self) -> str: + return self._data["email"] + + @property + def full_name_email(self) -> str: + if self.full_name: + return f"{self.full_name} <{self.email}>" + return self.email + + @property + def login_full_name_email(self) -> str: + return f"{self.login} ({self.full_name_email})" @classmethod def get( cls, conn: Connection, - ) -> GiteaHTTPResponse: + ) -> "Self": """ Retrieve details about the current user. :param conn: Gitea ``Connection`` instance. """ url = conn.makeurl("user") - return conn.request("GET", url) + response = conn.request("GET", url) + obj = cls(response.json(), response=response) + return obj diff --git a/tests/test_gitea_api_branch.py b/tests/test_gitea_api_branch.py new file mode 100644 index 00000000..05d26598 --- /dev/null +++ b/tests/test_gitea_api_branch.py @@ -0,0 +1,20 @@ +import unittest + +from osc.gitea_api import Branch + + +class TestGiteaApiPullRequest(unittest.TestCase): + def test_object(self): + data = { + "name": "branch", + "commit": { + "id": "commit", + }, + } + obj = Branch(data) + self.assertEqual(obj.name, "branch") + self.assertEqual(obj.commit, "commit") + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_gitea_api_pr.py b/tests/test_gitea_api_pr.py new file mode 100644 index 00000000..1fd53b7e --- /dev/null +++ b/tests/test_gitea_api_pr.py @@ -0,0 +1,116 @@ +import unittest + +from osc.gitea_api import PullRequest + + +class TestGiteaApiPullRequest(unittest.TestCase): + def test_object_pull_request(self): + data = { + "number": 1, + "url": "https://example.com/base-owner/base-repo", + "title": "title", + "body": "body", + "state": "state", + "user": { + "login": "alice", + "full_name": "Alice", + "email": "alice@example.com", + }, + "allow_maintainer_edit": False, + "draft": False, + "merged": False, + "base": { + "ref": "base-branch", + "sha": "base-commit", + "repo": { + "owner": { + "login": "base-owner", + }, + "name": "base-repo", + "ssh_url": "base-ssh-url", + }, + }, + "head": { + "ref": "head-branch", + "sha": "head-commit", + "repo": { + "owner": { + "login": "head-owner", + }, + "name": "head-repo", + "ssh_url": "head-ssh-url", + }, + }, + } + obj = PullRequest(data) + self.assertEqual(obj.is_pull_request, True) + self.assertEqual(obj.id, "base-owner/base-repo#1") + self.assertEqual(obj.url, "https://example.com/base-owner/base-repo") + self.assertEqual(obj.number, 1) + self.assertEqual(obj.title, "title") + self.assertEqual(obj.body, "body") + self.assertEqual(obj.state, "state") + self.assertEqual(obj.user, "alice") + self.assertEqual(obj.user_obj.login, "alice") + self.assertEqual(obj.draft, False) + self.assertEqual(obj.merged, False) + self.assertEqual(obj.allow_maintainer_edit, False) + + self.assertEqual(obj.base_owner, "base-owner") + self.assertEqual(obj.base_repo, "base-repo") + self.assertEqual(obj.base_branch, "base-branch") + self.assertEqual(obj.base_commit, "base-commit") + self.assertEqual(obj.base_ssh_url, "base-ssh-url") + + self.assertEqual(obj.head_owner, "head-owner") + self.assertEqual(obj.head_repo, "head-repo") + self.assertEqual(obj.head_branch, "head-branch") + self.assertEqual(obj.head_commit, "head-commit") + self.assertEqual(obj.head_ssh_url, "head-ssh-url") + + def test_object_issue(self): + data = { + "number": 1, + "url": "https://example.com/base-owner/base-repo", + "title": "title", + "body": "body", + "state": "state", + "user": { + "login": "alice", + "full_name": "Alice", + "email": "alice@example.com", + }, + "repository": { + "owner": "base-owner", + "name": "base-repo", + }, + } + obj = PullRequest(data) + self.assertEqual(obj.is_pull_request, False) + self.assertEqual(obj.id, "base-owner/base-repo#1") + self.assertEqual(obj.url, "https://example.com/base-owner/base-repo") + self.assertEqual(obj.number, 1) + self.assertEqual(obj.title, "title") + self.assertEqual(obj.body, "body") + self.assertEqual(obj.state, "state") + self.assertEqual(obj.user, "alice") + self.assertEqual(obj.user_obj.login, "alice") + self.assertEqual(obj.draft, None) + self.assertEqual(obj.merged, None) + self.assertEqual(obj.allow_maintainer_edit, None) + + self.assertEqual(obj.base_owner, "base-owner") + self.assertEqual(obj.base_repo, "base-repo") + self.assertEqual(obj.base_branch, None) + self.assertEqual(obj.base_commit, None) + self.assertEqual(obj.base_ssh_url, None) + + self.assertEqual(obj.head_owner, None) + self.assertEqual(obj.head_repo, None) + self.assertEqual(obj.head_branch, None) + self.assertEqual(obj.head_commit, None) + self.assertEqual(obj.head_ssh_url, None) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_gitea_api_repo.py b/tests/test_gitea_api_repo.py new file mode 100644 index 00000000..b5d003d0 --- /dev/null +++ b/tests/test_gitea_api_repo.py @@ -0,0 +1,41 @@ +import unittest + +from osc.gitea_api import Repo + + +class TestGiteaApiRepo(unittest.TestCase): + def test_object(self): + data = { + "owner": { + "login": "owner", + }, + "name": "repo", + "clone_url": "https://example.com/owner/repo", + "ssh_url": "gitea:example.com:owner/repo", + "default_branch": "default-branch", + "parent": { + "owner": { + "login": "parent-owner", + }, + "name": "parent-repo", + "clone_url": "https://example.com/parent-owner/parent-repo", + "ssh_url": "gitea:example.com:parent-owner/parent-repo", + }, + } + obj = Repo(data) + self.assertEqual(obj.owner, "owner") + self.assertEqual(obj.owner_obj.login, "owner") + self.assertEqual(obj.repo, "repo") + self.assertEqual(obj.clone_url, "https://example.com/owner/repo") + self.assertEqual(obj.ssh_url, "gitea:example.com:owner/repo") + self.assertEqual(obj.default_branch, "default-branch") + + self.assertEqual(obj.parent_obj.owner, "parent-owner") + self.assertEqual(obj.parent_obj.owner_obj.login, "parent-owner") + self.assertEqual(obj.parent_obj.repo, "parent-repo") + self.assertEqual(obj.parent_obj.clone_url, "https://example.com/parent-owner/parent-repo") + self.assertEqual(obj.parent_obj.ssh_url, "gitea:example.com:parent-owner/parent-repo") + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_gitea_api_ssh_key.py b/tests/test_gitea_api_ssh_key.py new file mode 100644 index 00000000..bf32d931 --- /dev/null +++ b/tests/test_gitea_api_ssh_key.py @@ -0,0 +1,20 @@ +import unittest + +from osc.gitea_api import SSHKey + + +class TestGiteaApiSSHKey(unittest.TestCase): + def test_object(self): + data = { + "id": 1, + "key": "ssh-rsa ZXhhbXBsZS1zc2gta2V5Cg==", + "title": "key title", + } + obj = SSHKey(data) + self.assertEqual(obj.id, 1) + self.assertEqual(obj.key, "ssh-rsa ZXhhbXBsZS1zc2gta2V5Cg==") + self.assertEqual(obj.title, "key title") + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_gitea_api_user.py b/tests/test_gitea_api_user.py new file mode 100644 index 00000000..32e58af5 --- /dev/null +++ b/tests/test_gitea_api_user.py @@ -0,0 +1,22 @@ +import unittest + +from osc.gitea_api import User + + +class TestGiteaApiUser(unittest.TestCase): + def test_object(self): + data = { + "login": "alice", + "full_name": "Alice", + "email": "alice@example.com", + } + obj = User(data) + self.assertEqual(obj.login, "alice") + self.assertEqual(obj.full_name, "Alice") + self.assertEqual(obj.email, "alice@example.com") + self.assertEqual(obj.full_name_email, "Alice ") + self.assertEqual(obj.login_full_name_email, "alice (Alice )") + + +if __name__ == "__main__": + unittest.main()