diff --git a/.gitignore b/.gitignore index e9cfa15..e63b4a4 100644 --- a/.gitignore +++ b/.gitignore @@ -91,3 +91,5 @@ ENV/ # IDE files .idea/ +# DRONE file +.drone.yml diff --git a/docs/entities.rst b/docs/entities.rst index 8e37c59..8820384 100644 --- a/docs/entities.rst +++ b/docs/entities.rst @@ -17,3 +17,12 @@ This pages documents classes provided by ``gogs_client`` module that represent e .. autoclass:: gogs_client.entities::GogsRepo.Permissions() :members: + +.. autoclass:: gogs_client.entities::GogsRepo.Hook() + :members: + +.. autoclass:: gogs_client.entities::GogsOrg() + :members: + +.. autoclass:: gogs_client.entities::GogsOrg.Team() + :members: \ No newline at end of file diff --git a/docs/updates.rst b/docs/updates.rst index b1b1d13..e196ade 100644 --- a/docs/updates.rst +++ b/docs/updates.rst @@ -8,3 +8,9 @@ Updates .. autoclass:: gogs_client.updates::GogsUserUpdate.Builder() :members: + +.. autoclass:: GogsHookUpdate() + :members: + +.. autoclass:: gogs_client.updates::GogsHookUpdate.Builder() + :members: \ No newline at end of file diff --git a/gogs_client/__init__.py b/gogs_client/__init__.py index 1112302..9f60d7e 100644 --- a/gogs_client/__init__.py +++ b/gogs_client/__init__.py @@ -1,4 +1,4 @@ from gogs_client.auth import Authentication, Token, UsernamePassword -from gogs_client.entities import GogsUser, GogsRepo +from gogs_client.entities import GogsUser, GogsRepo, GogsOrg from gogs_client.interface import GogsApi, ApiFailure, NetworkFailure -from gogs_client.updates import GogsUserUpdate +from gogs_client.updates import GogsUserUpdate, GogsHookUpdate diff --git a/gogs_client/_implementation/http_utils.py b/gogs_client/_implementation/http_utils.py index 4114e63..6dc29e8 100644 --- a/gogs_client/_implementation/http_utils.py +++ b/gogs_client/_implementation/http_utils.py @@ -36,13 +36,13 @@ def options(self, relative_path, params=None, **kwargs): return self.session.options(self.absolute_url(relative_path), params=params, **kwargs) def patch(self, relative_path, data=None, **kwargs): - return self.session.patch(self.absolute_url(relative_path), data=data, **kwargs) + return self.session.patch(self.absolute_url(relative_path), json=data, **kwargs) def post(self, relative_path, data=None, **kwargs): - return self.session.post(self.absolute_url(relative_path), data=data, **kwargs) + return self.session.post(self.absolute_url(relative_path), json=data, **kwargs) def put(self, relative_path, params=None, data=None, **kwargs): - return self.session.put(self.absolute_url(relative_path), params=params, data=data, **kwargs) + return self.session.put(self.absolute_url(relative_path), params=params, json=data, **kwargs) def append_url(base_url, path): diff --git a/gogs_client/entities.py b/gogs_client/entities.py index a117a0f..6f5e191 100644 --- a/gogs_client/entities.py +++ b/gogs_client/entities.py @@ -240,3 +240,329 @@ def pull(self): :rtype: bool """ return self._pull + + class Hook(object): + def __init__(self, hook_id, hook_type, events, active, config): + self._id = hook_id + self._type = hook_type + self._events = events + self._active = active + self._config = config + + @staticmethod + def from_json(parsed_json): + hook_id = json_get(parsed_json, "id") + hook_type = json_get(parsed_json, "type") + events = json_get(parsed_json, "events") + active = json_get(parsed_json, "active") + config = json_get(parsed_json, "config") + + return GogsRepo.Hook(hook_id=hook_id, hook_type=hook_type, events=events, active=active, + config=config) + + def as_dict(self): + fields = { + "id": self._id, + "type": self._type, + "events": self._events, + "config": self._config, + "active": self._active, + } + return {k: v for (k, v) in fields.items() if v is not None} + + @property # named hook_id to avoid conflict with built-in id + def hook_id(self): + """ + The hook's id number + + :rtype: int + """ + return self._id + + @property # named hook_type to avoid conflict with built-in type + def hook_type(self): + """ + The hook's type (gogs, slack, etc.) + + :rtype: str + """ + return self._type + + @property + def events(self): + """ + The events that fire the hook + + :rtype: List[str] + """ + return self._events + + @property + def active(self): + """ + Whether the hook is active + + :rtype: bool + """ + return self._active + + @property + def config(self): + """ + Config of the hook. Contains max. 3 keys: + - content_type + - url + - secret + + :rtype: dict + """ + return self._config + + class DeployKey(object): + def __init__(self, key_id, key, url, title, created_at, read_only): + self._id = key_id + self._key = key + self._url = url + self._title = title + self._created_at = created_at + self._read_only = read_only + + @staticmethod + def from_json(parsed_json): + key_id = json_get(parsed_json, "id") + key = json_get(parsed_json, "key") + url = json_get(parsed_json, "url") + title = json_get(parsed_json, "title") + created_at = json_get(parsed_json, "created_at") + read_only = json_get(parsed_json, "read_only") + + return GogsRepo.DeployKey(key_id=key_id, key=key, url=url, + title=title, created_at=created_at, read_only=read_only) + + def as_dict(self): + fields = { + "id": self._id, + "key": self._key, + "url": self._url, + "title": self._title, + "created_at": self._created_at, + "read_only": self._read_only, + } + return {k: v for (k, v) in fields.items() if v is not None} + + @property # named key_id to avoid conflict with built-in id + def key_id(self): + """ + The key's id number + + :rtype: int + """ + return self._id + + @property + def key(self): + """ + The content of the key + + :rtype: str + """ + return self._key + + @property + def url(self): + """ + Url where the key can be found + + :rtype: str + """ + return self._url + + @property + def title(self): + """ + The name of the key + + :rtype: str + """ + return self._title + + @property + def created_at(self): + """ + Creation date of the key. + :rtype: str + """ + return self._created_at + + @property + def read_only(self): + """ + Whether key is read-only. + :rtype: bool + """ + return self._read_only + +class GogsOrg(object): + """ + An immutable representation of a Gogs Organization. + """ + def __init__(self, org_id, username, full_name, avatar_url, description, website, location): + self._id = org_id + self._username = username + self._full_name = full_name + self._avatar_url = avatar_url + self._description = description + self._website = website + self._location = location + + @staticmethod + def from_json(parsed_json): + org_id = json_get(parsed_json, "id") + username = json_get(parsed_json, "username") + full_name = json_get(parsed_json, "full_name") + avatar_url = json_get(parsed_json, "avatar_url") + description = json_get(parsed_json, "description") + website = json_get(parsed_json, "website") + location = json_get(parsed_json, "location") + return GogsOrg(org_id=org_id, username=username, full_name=full_name, + avatar_url=avatar_url, description=description, + website=website, location=location) + + def as_dict(self): + fields = { + "id": self._id, + "username": self._username, + "full_name": self._full_name, + "avatar_url": self._avatar_url, + "description": self._description, + "website": self._website, + "location": self._location + } + return {k: v for (k, v) in fields.items() if v is not None} + + @property # named org_id to avoid conflict with built-in id + def org_id(self): + """ + The organization's id + + :rtype: int + """ + return self._id + + @property + def username(self): + """ + Organization's username + + :rtype: str + """ + return self._username + + @property + def full_name(self): + """ + Organization's full name + + :rtype: str + """ + return self._full_name + + @property + def avatar_url(self): + """ + Organization's avatar url + + :rtype: str + """ + return self._avatar_url + + @property + def description(self): + """ + Organization's description + + :rtype: str + """ + return self._description + + @property + def website(self): + """ + Organization's website address + + :rtype: str + """ + return self._website + + @property + def location(self): + """ + Organization's location + + :rtype: str + """ + return self._location + + class Team(object): + """ + Team of an organization + """ + def __init__(self, team_id, name, description, permission): + self._id = team_id + self._name = name + self._description = description + self._permission = permission + + @staticmethod + def from_json(parsed_json): + team_id = json_get(parsed_json, "id") + name = json_get(parsed_json, "name") + description = json_get(parsed_json, "description") + permission = json_get(parsed_json, "permission") + return GogsOrg.Team(team_id=team_id, name=name, description=description, permission=permission) + + def as_dict(self): + fields = { + "team_id": self._id, + "name": self._name, + "description": self._description, + "permission": self._permission, + } + return {k: v for (k, v) in fields.items() if v is not None} + + @property # named team_id to avoid conflict with built-in id + def team_id(self): + """ + Team's id + + :rtype: int + """ + return self._id + + @property + def name(self): + """ + Team name + + :rtype: str + """ + return self._name + + @property + def description(self): + """ + Description to the team + + :rtype: str + """ + return self._description + + @property + def permission(self): + """ + Team permission, can be read, write or admin, default is read + + :rtype: int + """ + return self._permission + diff --git a/gogs_client/interface.py b/gogs_client/interface.py index 95ac587..4b3567a 100644 --- a/gogs_client/interface.py +++ b/gogs_client/interface.py @@ -1,7 +1,7 @@ import requests from gogs_client._implementation.http_utils import RelativeHttpRequestor, append_url -from gogs_client.entities import GogsUser, GogsRepo +from gogs_client.entities import GogsUser, GogsRepo, GogsOrg from gogs_client.auth import Token @@ -111,7 +111,8 @@ def ensure_token(self, auth, name, username=None): return self.create_token(auth, name, username) def create_repo(self, auth, name, description=None, private=False, auto_init=False, - gitignore_templates=None, license_template=None, readme_template=None): + gitignore_templates=None, license_template=None, readme_template=None, + organization=None): """ Creates a new repository, and returns the created repository. @@ -123,6 +124,7 @@ def create_repo(self, auth, name, description=None, private=False, auto_init=Fal :param list[str] gitignore_templates: collection of ``.gitignore`` templates to apply :param str license_template: license template to apply :param str readme_template: README template to apply + :param str organization: organization under which repository is created :return: a representation of the created repository :rtype: GogsRepo :raises NetworkFailure: if there is an error communicating with the server @@ -140,7 +142,8 @@ def create_repo(self, auth, name, description=None, private=False, auto_init=Fal "readme": readme_template } data = {k: v for (k, v) in data.items() if v is not None} - response = self._post("/user/repos", auth=auth, data=data) + url = "/org/{0}/repos".format(organization) if organization else "/user/repos" + response = self._post(url, auth=auth, data=data) return GogsRepo.from_json(self._check_ok(response).json()) def repo_exists(self, auth, username, repo_name): @@ -175,6 +178,22 @@ def get_repo(self, auth, username, repo_name): response = self._check_ok(self._get(path, auth=auth)) return GogsRepo.from_json(response.json()) + def get_user_repos(self, auth, username): + """ + Returns the repositories owned by + the user with username ``username``. + + :param auth.Authentication auth: authentication object + :param str username: username of owner of repository + :return: a list of repositories + :rtype: List[GogsRepo] + :raises NetworkFailure: if there is an error communicating with the server + :raises ApiFailure: if the request cannot be serviced + """ + path = "/users/{u}/repos".format(u=username) + response = self._check_ok(self._get(path, auth=auth)) + return [GogsRepo.from_json(repo_json) for repo_json in response.json()] + def delete_repo(self, auth, username, repo_name): """ Deletes the repository with name ``repo_name`` owned by the user with username ``username``. @@ -188,6 +207,43 @@ def delete_repo(self, auth, username, repo_name): path = "/repos/{u}/{r}".format(u=username, r=repo_name) self._check_ok(self._delete(path, auth=auth)) + def migrate_repo(self, auth, clone_addr, + uid, repo_name, auth_username=None, auth_password=None, + mirror=False, private=False, description=None): + """ + Migrate a repository from other Git hosting sources for the authenticated user. + + :param str clone_addr: Remote Git address (HTTP/HTTPS URL or local path) + :param str auth_username: Authorization username + :param str auth_password: Authorization password + :param int uid: User ID who takes ownership of this repository + :param str repo_name: Repository name + :param bool mirror: Repository will be a mirror. Default is false + :param bool private: Repository will be private. Default is false + :param str descriptrion: Repository description + :return: a representation of the migrated repository + :rtype: GogsRepo + :raises NetworkFailure: if there is an error communicating with the server + :raises ApiFailure: if the request cannot be serviced + """ + # "auth_username": auth_username, + # "auth_password": auth_password, + + data = { + "clone_addr": clone_addr, + "uid": uid, + "repo_name": repo_name, + "auth_username": auth_username, + "auth_password": auth_password, + "mirror": mirror, + "private": private, + "description": description, + } + data = {k: v for (k, v) in data.items() if v is not None} + url = "/repos/migrate" + response = self._post(url, auth=auth, data=data) + return GogsRepo.from_json(self._check_ok(response).json()) + def create_user(self, auth, login_name, username, email, password, send_notify=False): """ Creates a new user, and returns the created user. @@ -287,6 +343,257 @@ def delete_user(self, auth, username): path = "/admin/users/{}".format(username) self._check_ok(self._delete(path, auth=auth)) + def get_repo_hooks(self, auth, username, repo_name): + """ + Returns all hooks of repository with name ``repo_name`` owned by + the user with username ``username``. + + :param auth.Authentication auth: authentication object + :param str username: username of owner of repository + :param str repo_name: name of repository + :return: a list of hooks for the specified repository + :rtype: List[GogsRepo.Hooks] + :raises NetworkFailure: if there is an error communicating with the server + :raises ApiFailure: if the request cannot be serviced + """ + path = "/repos/{u}/{r}/hooks".format(u=username, r=repo_name) + response = self._check_ok(self._get(path, auth=auth)) + hooks = [GogsRepo.Hook.from_json(hook) for hook in response.json()] + return hooks + + def create_hook(self, auth, repo_name, hook_type, config, events=["push"], organization=None, active=False): + """ + Creates a new hook, and returns the created hook. + + :param auth.Authentication auth: authentication object, must be admin-level + :param str repo_name: the name of the repo for which we create the hook + :param str hook_type: The type of webhook, either "gogs" or "slack" + :param dict config: Key/value pairs to provide settings for this hook ("url", "content_type", "secret") + :param list events: Determines what events the hook is triggered for. Default: ["push"] + :param str organization: (Optional) Organization of the repo + :param bool active: Determines whether the hook is actually triggered on pushes. Default is false + :return: a representation of the created hook + :rtype: GogsRepo.Hook + :raises NetworkFailure: if there is an error communicating with the server + :raises ApiFailure: if the request cannot be serviced + """ + + data = { + "type": hook_type, + "config": config, + "events": events, + "active": active + } + + url = "/repos/{o}/{r}/hooks".format(o=organization, r=repo_name) if organization else "/repos/{r}/hooks".format(r=repo_name) + response = self._post(url, auth=auth, data=data) + self._check_ok(response) + return GogsRepo.Hook.from_json(response.json()) + + def update_hook(self, auth, repo_name, hook_id, update, organization=None): + """ + Updates hook with id ``hook_id`` according to ``update``. + + :param auth.Authentication auth: authentication object, must be admin-level + :param str repo_name: repo of the hook to update + :param GogsHookUpdate update: a ``GogsHookUpdate`` object describing the requested update + :return: the updated hook + :rtype: GogsRepo.Hook + :raises NetworkFailure: if there is an error communicating with the server + :raises ApiFailure: if the request cannot be serviced + """ + path = "/repos/{o}/{r}/hooks/{i}".format(o=organization, r=repo_name, i=hook_id) if organization else "/repos/{r}/hooks/{i}".format(r=repo_name, i=hook_id) + response = self._check_ok(self._patch(path, auth=auth, data=update.as_dict())) + return GogsRepo.Hook.from_json(response.json()) + + def delete_hook(self, auth, username, repo_name, hook_id): + """ + Deletes the hook with id ``hook_id`` for repo with name ``repo_name`` + owned by the user with username ``username``. + + :param auth.Authentication auth: authentication object + :param str username: username of owner of repository + :param str repo_name: name of repository of hook to delete + :param int hook_id: id of hook to delete + :raises NetworkFailure: if there is an error communicating with the server + :raises ApiFailure: if the request cannot be serviced + """ + path = "/repos/{u}/{r}/hooks/{i}".format(u=username, r=repo_name, i=hook_id) + response = self._check_ok(self._delete(path, auth=auth)) + + def create_organization(self, auth, username, org_name, full_name=None, avatar_url=None, description=None, website=None, location=None): + """ + Creates a new organization, and returns the created one. + + :param auth.Authentication auth: authentication object, must be admin-level + :param str username: [Required] Organization user name + :param str full_name: Full name of organization + :param str description: Description to the organization + :param str website: Official website + :param str location: Organization location + :return: a representation of the created organization + :rtype: GogsOrg + :raises NetworkFailure: if there is an error communicating with the server + :raises ApiFailure: if the request cannot be serviced + """ + + data = { + "username": org_name, + "full_name": full_name, + "description": description, + "website": website, + "location": location + } + + url = "/admin/users/{u}/orgs".format(u=username) + response = self._post(url, auth=auth, data=data) + self._check_ok(response) + return GogsOrg.from_json(response.json()) + + def create_organization_team(self, auth, org_name, name, description=None, permission="read"): + """ + Creates a new team of the organization. + + :param auth.Authentication auth: authentication object, must be admin-level + :param str org_name: [Required] Organization user name + :param str name: Full name of the team + :param str description: Description to the team + :param str permission: Team permission, can be read, write or admin, default is read + :return: a representation of the created team + :rtype: GogsOrg.Team + :raises NetworkFailure: if there is an error communicating with the server + :raises ApiFailure: if the request cannot be serviced + """ + + data = { + "name": name, + "description": description, + "permission": permission + } + + url = "/admin/orgs/{o}/teams".format(o=org_name) + response = self._post(url, auth=auth, data=data) + self._check_ok(response) + return GogsOrg.Team.from_json(response.json()) + + def add_team_membership(self, auth, team_id, username): + """ + Add user to team. + + :param auth.Authentication auth: authentication object, must be admin-level + :param str team_id: [Required] Id of the team + :param str username: Username of the user to be added to team + :raises NetworkFailure: if there is an error communicating with the server + :raises ApiFailure: if the request cannot be serviced + """ + url = "/admin/teams/{t}/members/{u}".format(t=team_id, u=username) + response = self._check_ok(self._put(url, auth=auth)) + + def remove_team_membership(self, auth, team_id, username): + """ + Remove user from team. + + :param auth.Authentication auth: authentication object, must be admin-level + :param str team_id: [Required] Id of the team + :param str username: Username of the user to be removed from the team + :raises NetworkFailure: if there is an error communicating with the server + :raises ApiFailure: if the request cannot be serviced + """ + url = "/admin/teams/{t}/members/{u}".format(t=team_id, u=username) + response = self._check_ok(self._delete(url, auth=auth)) + + def add_repo_to_team(self, auth, team_id, repo_name): + """ + Add or update repo from team. + + :param auth.Authentication auth: authentication object, must be admin-level + :param str team_id: [Required] Id of the team + :param str repo_name: Name of the repo to be added to the team + :raises NetworkFailure: if there is an error communicating with the server + :raises ApiFailure: if the request cannot be serviced + """ + url = "/admin/teams/{t}/repos/{r}".format(t=team_id, r=repo_name) + response = self._check_ok(self._put(url, auth=auth)) + + def remove_repo_from_team(self, auth, team_id, repo_name): + """ + Remove repo from team. + + :param auth.Authentication auth: authentication object, must be admin-level + :param str team_id: [Required] Id of the team + :param str repo_name: Name of the repo to be removed from the team + :raises NetworkFailure: if there is an error communicating with the server + :raises ApiFailure: if the request cannot be serviced + """ + url = "/admin/teams/{t}/repos/{r}".format(t=team_id, r=repo_name) + response = self._check_ok(self._delete(url, auth=auth)) + + def list_deploy_keys(self, auth, username, repo_name): + """ + List deploy keys. + + :param str username: username or organization + :param str repo_name: the name of the repo + :return: a list of deploy keys for the repo + :rtype: List[GogsRepo.DeployKey] + :raises NetworkFailure: if there is an error communicating with the server + :raises ApiFailure: if the request cannot be serviced + """ + response = self._check_ok(self._get("/repos/{u}/{r}/keys".format(u=username, r=repo_name),auth=auth)) + return [GogsRepo.DeployKey.from_json(key_json) for key_json in response.json()] + + def get_deploy_key(self, auth, username, repo_name, key_id): + """ + Get deploy key for specific repo. + + :param str username: username or organization + :param str repo_name: the name of the repo + :param int key_id: the id of the key + :return: the deploy key + :rtype: GogsRepo.DeployKey + :raises NetworkFailure: if there is an error communicating with the server + :raises ApiFailure: if the request cannot be serviced + """ + + response = self._check_ok(self._get("/repos/{u}/{r}/keys/{k}".format(u=username, r=repo_name, k=key_id), auth=auth)) + return GogsRepo.DeployKey.from_json(response.json()) + + def add_deploy_key(self, auth, username, repo_name, title, key): + """ + Get deploy key for specific repo. + + :param str username: username or organization + :param str repo_name: the name of the repo + :param int key_id: the id of the key + :return: the deploy key + :rtype: GogsRepo.DeployKey + :raises NetworkFailure: if there is an error communicating with the server + :raises ApiFailure: if the request cannot be serviced + """ + data = { + "title": title, + "key": key + } + response = self._check_ok(self._post("/repos/{u}/{r}/keys".format(u=username, r=repo_name), auth=auth, data=data)) + return GogsRepo.DeployKey.from_json(response.json()) + + def delete_deploy_key(self, auth, username, repo_name, key_id): + """ + Remove deploy key for specific repo. + + :param str username: username or organization + :param str repo_name: the name of the repo + :param int key_id: the id of the key + :return: the deploy key + :rtype: GogsRepo.DeployKey + :raises NetworkFailure: if there is an error communicating with the server + :raises ApiFailure: if the request cannot be serviced + """ + + response = self._check_ok(self._delete("/repos/{u}/{r}/keys/{k}".format(u=username, r=repo_name, k=key_id), auth=auth)) + return self._check_ok(response) + + # Helper methods def _delete(self, path, auth=None, **kwargs): @@ -321,6 +628,14 @@ def _post(self, path, auth=None, **kwargs): except requests.RequestException as exc: raise NetworkFailure(exc) + def _put(self, path, auth=None, **kwargs): + if auth is not None: + auth.update_kwargs(kwargs) + try: + return self._requestor.put(path, **kwargs) + except requests.RequestException as exc: + raise NetworkFailure(exc) + @staticmethod def _check_ok(response): """ @@ -337,7 +652,7 @@ def _fail(response): """ message = "Status code: {}-{}, url: {}".format(response.status_code, response.reason, response.url) try: - message += ", message:{}".format(response.json()[0]["message"]) + message += ", message:{}".format(response.json()["message"]) except Exception: pass raise ApiFailure(message, response.status_code) diff --git a/gogs_client/updates.py b/gogs_client/updates.py index f31d7e2..e45754a 100644 --- a/gogs_client/updates.py +++ b/gogs_client/updates.py @@ -182,3 +182,76 @@ def build(self): admin=self._admin, allow_git_hook=self._allow_git_hook, allow_import_local=self._allow_import_local) + + +class GogsHookUpdate(object): + """ + An immutable represention of a collection of Gogs hook attributes to update. + + Instances should be created using the :class:`~GogsHookUpdate.Builder` class. + """ + def __init__(self, events, config, active): + """ + :param events: + :param config: + :param active: + """ + + self._events = events + self._config = config + self._active = active + + def as_dict(self): + fields = { + "events": self._events, + "config": self._config, + "active": self._active, + } + return {k: v for (k, v) in fields.items() if v is not None} + + class Builder(object): + def __init__(self): + """ + :param str login_name: login name for authentication source + :param str email: email address of user to update + """ + self._events = None + self._config = None + self._active = None + + def set_events(self, events): + """ + :param list events: + :return: the updated builder + :rtype: GogsHookUpdate.Builder + """ + self._events = events + return self + + def set_config(self, config): + """ + :param dict config: + :return: the updated builder + :rtype: GogsHookUpdate.Builder + """ + self._config = config + return self + + def set_active(self, active): + """ + :param bool active: + :return: the updated builder + :rtype: GogsHookUpdate.Builder + """ + self._active = active + return self + + def build(self): + """ + :return: A :class:`~GogsHookUpdate` instance reflecting the changes added to the builder. + :rtype: GogsHookUpdate + """ + return GogsHookUpdate( + events=self._events, + config=self._config, + active=self._active) diff --git a/tests/interface_test.py b/tests/interface_test.py index a2d0be8..a0cbdab 100644 --- a/tests/interface_test.py +++ b/tests/interface_test.py @@ -36,6 +36,47 @@ def setUp(self): "pull": true } }""" + self.repos_list_json_str = """[{ + "id": 27, + "owner": { + "id": 1, + "username": "unknwon", + "full_name": "", + "email": "u@gogs.io", + "avatar_url": "/avatars/1" + }, + "full_name": "unknwon/Hello-World", + "private": false, + "fork": false, + "html_url": "http://localhost:3000/unknwon/Hello-World", + "clone_url": "http://localhost:3000/unknwon/hello-world.git", + "ssh_url": "jiahuachen@localhost:unknwon/hello-world.git", + "permissions": { + "admin": true, + "push": true, + "pull": true + } + },{ + "id": 28, + "owner": { + "id": 1, + "username": "unknwon", + "full_name": "", + "email": "u@gogs.io", + "avatar_url": "/avatars/1" + }, + "full_name": "unknwon/Hello-World-Again", + "private": false, + "fork": false, + "html_url": "http://localhost:3000/unknwon/Hello-World-Again", + "clone_url": "http://localhost:3000/unknwon/hello-world-again.git", + "ssh_url": "jiahuachen@localhost:unknwon/hello-world-again.git", + "permissions": { + "admin": true, + "push": true, + "pull": true + } + }]""" self.user_json_str = """{ "id": 1, "username": "unknwon", @@ -49,8 +90,98 @@ def setUp(self): }""" self.username_password = gogs_client.UsernamePassword( "auth_username", "password") + self.hook_json_str = """{ + "id": 4, + "type": "gogs", + "config": { + "content_type": "json", + "url": "http://test.io/hook" + }, + "events": [ + "create", + "push", + "issues" + ], + "active": false, + "updated_at": "2017-03-31T12:42:58Z", + "created_at": "2017-03-31T12:42:58Z" + }""" + self.hooks_list_json_str = """[ + { + "id": 4, + "type": "gogs", + "config": { + "content_type": "json", + "url": "http://test.io/hook" + }, + "events": [ + "create", + "push", + "issues" + ], + "active": false, + "updated_at": "2017-03-31T12:42:58Z", + "created_at": "2017-03-31T12:42:58Z" + }, + { + "id": 3, + "type": "gogs", + "config": { + "content_type": "json", + "url": "http://192.168.201.1:8080/hook22/" + }, + "events": [ + "issue_comment" + ], + "active": true, + "updated_at": "2017-03-31T12:47:56Z", + "created_at": "2017-03-31T12:42:54Z" + } + ]""" + self.org_json_str = """{ + "id": 7, + "username": "gogs2", + "full_name": "Gogs2", + "avatar_url": "/avatars/7", + "description": "Gogs is a painless self-hosted Git Service.", + "website": "https://gogs.io", + "location": "USA" + }""" + self.team_json_str = """{ + "id": 12, + "name": "new-team", + "description": "A new team created by API", + "permission": "write" + }""" + self.deploy_key_json_str = """{ + "id": 1, + "key": "ssh-rsa AAAAB3NzaC1yc2EAAAADAQABAAABAQDUbmwBOG5vI8qNCztby5LDc9ozwTuwsqf+1fpuHjT9iQ2Lu9nlKHQJcPSgdrYAcc+88K6o74ayhTAjfajKxkIHnbzZFjidoVZSQDhX5qvl93jvY/Uz390qky0sweW+fspm8pRJL+ofE3QEN5AXAuycq1tgsRT32XC+Ta82Xyv8b3xW+pWbsZzYCzUsZXDe/xWxg1rndXh2BIrmcYf9BMiv9ZJIojJXfuLCeRXl550tDzaMFC0rQ/T5pZjs/lQemtg92MnxnEDi5nhuvDwM4Q8eqCTOXc4BCE7iyIHv+B7rx+0x99ytMh5BSIIGyWTfgTot/AjGVm5aRKJSRFgPBm9N comment with whitespace", + "url": "http://localhost:3000/api/v1/repos/unknwon/project_x/keys/1", + "title": "local", + "created_at": "2015-11-18T15:05:43-05:00", + "read_only": true + }""" + self.deploy_key_json_list = """[{ + "id": 1, + "key": "ssh-rsa AAAAB3NzaC1yc2EAAAADAQABAAABAQDUbmwBOG5vI8qNCztby5LDc9ozwTuwsqf+1fpuHjT9iQ2Lu9nlKHQJcPSgdrYAcc+88K6o74ayhTAjfajKxkIHnbzZFjidoVZSQDhX5qvl93jvY/Uz390qky0sweW+fspm8pRJL+ofE3QEN5AXAuycq1tgsRT32XC+Ta82Xyv8b3xW+pWbsZzYCzUsZXDe/xWxg1rndXh2BIrmcYf9BMiv9ZJIojJXfuLCeRXl550tDzaMFC0rQ/T5pZjs/lQemtg92MnxnEDi5nhuvDwM4Q8eqCTOXc4BCE7iyIHv+B7rx+0x99ytMh5BSIIGyWTfgTot/AjGVm5aRKJSRFgPBm9N comment with whitespace", + "url": "http://localhost:3000/api/v1/repos/unknwon/project_x/keys/1", + "title": "local", + "created_at": "2015-11-18T15:05:43-05:00", + "read_only": true + },{ + "id": 2, + "key": "ssh-rsa AAAAB3NzaC1yc2EAAAADAQABAAABAQDUbmwBOG5vI8qNCztby5LDc9ozwTuwsqf+1fpuHjT9iQ2Lu9nlKHQJcPSgdrYAcc+88K6o74ayhTAjfajKxkIHnbzZFjidoVZSQDhX5qvl93jvY/Uz390qky0sweW+fspm8pRJL+ofE3QEN5AXAuycq1tgsRT32XC+Ta82Xyv8b3xW+pWbsZzYCzUsZXDe/xWxg1rndXh2BIrmcYf9BMiv9ZJIojJXfuLCeRXl550tDzaMFC0rQ/T5pZjs/lQemtg92MnxnEDi5nhuvDwM4Q8eqCTOXc4BCE7iyIHv+B7rx+0x99ytMh5BSIIGyWTfgTot/AjGVm5aRKJSRFgPBm9N comment with whitespace", + "url": "http://localhost:3000/api/v1/repos/unknwon/project_x/keys/1", + "title": "local", + "created_at": "2015-11-18T15:05:43-05:00", + "read_only": true + }]""" self.expected_repo = gogs_client.GogsRepo.from_json(json.loads(self.repo_json_str)) self.expected_user = gogs_client.GogsUser.from_json(json.loads(self.user_json_str)) + self.expected_hook = gogs_client.GogsRepo.Hook.from_json(json.loads(self.hook_json_str)) + self.expected_org = gogs_client.GogsOrg.from_json(json.loads(self.org_json_str)) + self.expected_team = gogs_client.GogsOrg.Team.from_json(json.loads(self.team_json_str)) + self.expected_key = gogs_client.GogsRepo.DeployKey.from_json(json.loads(self.deploy_key_json_str)) self.token = gogs_client.Token.from_json(json.loads(self.token_json_str)) @responses.activate @@ -93,6 +224,15 @@ def test_get_repo1(self): last_call = responses.calls[1] self.assertEqual(last_call.request.url, self.path_with_token(uri2)) + @responses.activate + def test_get_user_repos(self): + uri = self.path("/users/username/repos") + responses.add(responses.GET, uri, body=self.repos_list_json_str, status=200) + repos = self.client.get_user_repos(self.token, "username") + self.assertEqual(len(repos), 2) + self.assert_repos_equal(repos[0], self.expected_repo) + + @responses.activate def test_delete_repo1(self): uri1 = self.path("/repos/username/repo1") @@ -184,7 +324,7 @@ def test_update_user1(self): .build() def callback(request): - data = self.data_of_query(request.body) + data = json.loads(request.body.decode('utf8')) self.assertEqual(data["login_name"], "loginname") self.assertEqual(data["full_name"], "Example User") self.assertEqual(data["email"], "user@example.com") @@ -267,6 +407,149 @@ def test_ensure_auth_token(self): token = self.client.ensure_token(self.username_password, self.token.name) self.assert_tokens_equals(token, self.token) + @responses.activate + def test_create_hook1(self): + uri = self.path("/repos/username/repo1/hooks") + responses.add(responses.POST, uri, body=self.hook_json_str) + hook = self.client.create_hook(self.token, + repo_name="repo1", + hook_type="gogs", + config={ + "content_type": "json2", + "url": "http://test.io/hook" + }, + events=["create", "push", "issues"], + active=False, + organization="username") + self.assert_hooks_equals(hook, self.expected_hook) + self.assertEqual(len(responses.calls), 1) + call = responses.calls[0] + self.assertEqual(call.request.url, self.path_with_token(uri)) + + @responses.activate + def test_update_hook1(self): + update = gogs_client.GogsHookUpdate.Builder()\ + .set_events(["issues_comments"])\ + .set_config({"url": "http://newurl.com/hook"})\ + .set_active(True)\ + .build() + + def callback(request): + data = json.loads(request.body.decode('utf8')) + self.assertEqual(data["config"]["url"], "http://newurl.com/hook") + self.assertEqual(data["events"], ['issues_comments']) + self.assertEqual(data["active"], True) + return 200, {}, self.hook_json_str + uri = self.path("/repos/username/repo1/hooks/4") + responses.add_callback(responses.PATCH, uri, callback=callback) + hook = self.client.update_hook(self.token, "repo1", 4, update, organization="username") + self.assert_hooks_equals(hook, self.expected_hook) + + @responses.activate + def test_list_hooks(self): + uri = self.path("/repos/username/repo1/hooks") + responses.add(responses.GET, uri, body=self.hooks_list_json_str, status=200) + hooks = self.client.get_repo_hooks(self.token, "username", "repo1") + self.assertEqual(len(hooks), 2) + self.assert_hooks_equals(hooks[0], self.expected_hook) + + @responses.activate + def test_delete_hook(self): + uri = self.path("/repos/username/repo1/hooks/4") + responses.add(responses.DELETE, uri, status=204) + hook = self.client.delete_hook(self.token, "username", "repo1", 4) + self.assertEqual(hook, None) + + @responses.activate + def test_create_organization(self): + uri = self.path("/admin/users/username/orgs") + responses.add(responses.POST, uri, body=self.org_json_str) + org = self.client.create_organization(self.token, + username="username", + org_name="gogs2", + full_name="Gogs2", + description="Gogs is a painless self-hosted Git Service.", + website="https://gogs.io", + location="USA") + self.assert_org_equals(org, self.expected_org) + self.assertEqual(len(responses.calls), 1) + call = responses.calls[0] + self.assertEqual(call.request.url, self.path_with_token(uri)) + + @responses.activate + def test_create_organization_team(self): + uri = self.path("/admin/orgs/username/teams") + responses.add(responses.POST, uri, body=self.team_json_str) + team = self.client.create_organization_team(self.token, + org_name="username", + name="new-team", + description="A new team created by API", + permission="write") + self.assert_team_equals(team, self.expected_team) + self.assertEqual(len(responses.calls), 1) + call = responses.calls[0] + self.assertEqual(call.request.url, self.path_with_token(uri)) + + @responses.activate + def test_add_team_membership(self): + uri = self.path("/admin/teams/team/members/username") + responses.add(responses.PUT, uri, status=204) + resp = self.client.add_team_membership(self.token, "team", "username") + self.assertEqual(resp, None) + + @responses.activate + def test_remove_team_membership(self): + uri = self.path("/admin/teams/team/members/username") + responses.add(responses.DELETE, uri, status=204) + resp = self.client.remove_team_membership(self.token, "team", "username") + self.assertEqual(resp, None) + + @responses.activate + def test_add_repo_to_team(self): + uri = self.path("/admin/teams/test_team/repos/repo_name") + responses.add(responses.PUT, uri, status=204) + resp = self.client.add_repo_to_team(self.token, "test_team", "repo_name") + self.assertEqual(resp, None) + + @responses.activate + def test_remove_repo_from_team(self): + uri = self.path("/admin/teams/test_team/repos/repo_name") + responses.add(responses.DELETE, uri, status=204) + resp = self.client.remove_repo_from_team(self.token, "test_team", "repo_name") + self.assertEqual(resp, None) + + @responses.activate + def test_list_deploy_keys(self): + uri = self.path("/repos/username/repo1/keys") + responses.add(responses.GET, uri, body=self.deploy_key_json_list, status=200) + keys = self.client.list_deploy_keys(self.token, "username", "repo1") + self.assertEqual(len(keys), 2) + self.assert_keys_equals(keys[0], self.expected_key) + + @responses.activate + def test_delete_deploy_keys(self): + uri = self.path("/repos/username/repo1/keys/1") + responses.add(responses.DELETE, uri, status=204) + key = self.client.delete_deploy_key(self.token, "username", "repo1", 1) + self.assertEqual(key.status_code, 204) + + @responses.activate + def test_get_deploy_key(self): + uri = self.path("/repos/username/repo1/keys/1") + responses.add(responses.GET, uri, body=self.deploy_key_json_str) + key = self.client.get_deploy_key(self.token, "username", "repo1", 1) + self.assert_keys_equals(key, self.expected_key) + + @responses.activate + def test_add_deploy_key(self): + uri = self.path("/repos/username/repo1/keys") + responses.add(responses.POST, uri, body=self.deploy_key_json_str) + key_title = "My key title" + key_content = "My key content" + key = self.client.add_deploy_key(self.token, "username", "repo1", key_title, key_content) + self.assert_keys_equals(key, self.expected_key) + + # helper methods @staticmethod @@ -310,6 +593,35 @@ def assert_tokens_equals(self, token, expected): self.assertEqual(token.name, expected.name) self.assertEqual(token.token, expected.token) + def assert_hooks_equals(self, hook, expected): + self.assertEqual(hook.hook_id, expected.hook_id) + self.assertEqual(hook.hook_type, expected.hook_type) + self.assertEqual(hook.events, expected.events) + self.assertEqual(hook.config, expected.config) + self.assertEqual(hook.active, expected.active) + + def assert_org_equals(self, org, expected): + self.assertEqual(org.org_id, expected.org_id) + self.assertEqual(org.username, expected.username) + self.assertEqual(org.full_name, expected.full_name) + self.assertEqual(org.avatar_url, expected.avatar_url) + self.assertEqual(org.description, expected.description) + self.assertEqual(org.website, expected.website) + self.assertEqual(org.location, expected.location) + + def assert_team_equals(self, team, expected): + self.assertEqual(team.team_id, expected.team_id) + self.assertEqual(team.name, expected.name) + self.assertEqual(team.description, expected.description) + self.assertEqual(team.permission, expected.permission) + + def assert_keys_equals(self, key, expected): + self.assertEqual(key.key_id, expected.key_id) + self.assertEqual(key.title, expected.title) + self.assertEqual(key.url, expected.url) + self.assertEqual(key.key, expected.key) + self.assertEqual(key.read_only, expected.read_only) + self.assertEqual(key.created_at, expected.created_at) if __name__ == "__main__": unittest.main()