diff --git a/README.md b/README.md index 425ba73..3269a95 100644 --- a/README.md +++ b/README.md @@ -1,10 +1,10 @@ # py-gitea -A very simple API client for Gitea > 1.16.1 +A very simple API client for Gitea > 1.16.1 This has been somewhat tested (and used), so most things should work as expected. -Note that not the full Swagger-API is accessible. The whole implementation is focused +Note that not the full Swagger-API is accessible. The whole implementation is focused on making access and working with Organizations, Teams, Repositories and Users as pain free as possible. @@ -27,24 +27,25 @@ print("Gitea Version: " + gitea.get_version()) print("API-Token belongs to user: " + gitea.get_user().username) ``` -Adding entities like Users, Organizations, ... also is done via the gitea object. +Adding entities like Users, Organizations, ... also is done via the gitea object. ```python user = gitea.create_user("Test Testson", "test@test.test", "password") ``` All operations on entities in gitea are then accomplished via the according wrapper objects for those entities. -Each of those objects has a `.request` method that creates an entity according to your gitea instance. +Each of those objects has a `.request` method that creates an entity according to your gitea instance. ```python other_user = User.request(gitea, "OtherUserName") print(other_user.username) ``` -Note that the fields of the User, Organization,... classes are dynamically created at runtime, and thus not visible during divelopment. Refer to the Gitea-API documentation for the fields names. +Note that the fields of the User, Organization,... classes are dynamically created at runtime, and thus not visible +during divelopment. Refer to the Gitea-API documentation for the fields names. - -Fields that can not be altered via gitea-api, are read only. After altering a field, the `.commit` method of the according object must be called to synchronize the changed fields with your gitea instance. +Fields that can not be altered via gitea-api, are read only. After altering a field, the `.commit` method of the +according object must be called to synchronize the changed fields with your gitea instance. ```python org = Organization.request(gitea, test_org) @@ -54,30 +55,31 @@ org.commit() ``` An entity in gitea can be deleted by calling delete. + ```python org.delete() ``` All entity objects do have methods to execute some of the requests possible though the gitea-api: + ```python org = Organization.request(gitea, ORGNAME) teams = org.get_teams() for team in teams: - repos = team.get_repos() - for repo in repos: - print(repo.name) + repos = team.get_repos() + for repo in repos: + print(repo.name) ``` - ## Installation Use ``pip install py-gitea`` to install. ## Tests -Tests can be run with: +Tests can be run with: ```python3 -m pytest test_api.py``` -Make sure to have a gitea-instance running on `http://localhost:3000`, and an admin-user token at `.token`. +Make sure to have a gitea-instance running on `http://localhost:3000`, and an admin-user token at `.token`. The admin user must be named ``test``, with email ``secondarytest@test.org``. diff --git a/__init__.py b/__init__.py index fd1e2e3..7b32042 100644 --- a/__init__.py +++ b/__init__.py @@ -9,4 +9,4 @@ from .gitea import ( AlreadyExistsException, Issue, Milestone, -) \ No newline at end of file +) diff --git a/gitea/__init__.py b/gitea/__init__.py index 97bbedd..577ba42 100644 --- a/gitea/__init__.py +++ b/gitea/__init__.py @@ -15,21 +15,21 @@ from .apiobject import ( Milestone, Commit, Comment, - Content + Content, ) __all__ = [ - 'Gitea', - 'User', - 'Organization', - 'Team', - 'Repository', - 'Branch', - 'NotFoundException', - 'AlreadyExistsException', - 'Issue', - 'Milestone', - 'Commit', - 'Comment', - 'Content' + "Gitea", + "User", + "Organization", + "Team", + "Repository", + "Branch", + "NotFoundException", + "AlreadyExistsException", + "Issue", + "Milestone", + "Commit", + "Comment", + "Content", ] diff --git a/gitea/apiobject.py b/gitea/apiobject.py index 79f0002..3343919 100644 --- a/gitea/apiobject.py +++ b/gitea/apiobject.py @@ -21,18 +21,19 @@ class Organization(ApiObject): super().__init__(gitea) def __eq__(self, other): - if not isinstance(other, Organization): return False + if not isinstance(other, Organization): + return False return self.gitea == other.gitea and self.name == other.name def __hash__(self): return hash(self.gitea) ^ hash(self.name) @classmethod - def request(cls, gitea: 'Gitea', name: str) -> 'Organization': + def request(cls, gitea: "Gitea", name: str) -> "Organization": return cls._request(gitea, {"name": name}) @classmethod - def parse_response(cls, gitea, result) -> 'Organization': + def parse_response(cls, gitea, result) -> "Organization": api_object = super().parse_response(gitea, result) # add "name" field to make this behave similar to users for gitea < 1.18 # also necessary for repository-owner when org is repo owner @@ -40,27 +41,31 @@ class Organization(ApiObject): Organization._add_read_property("name", result["username"], api_object) return api_object - _patchable_fields = {"description", "full_name", "location", "visibility", "website"} + _patchable_fields = { + "description", + "full_name", + "location", + "visibility", + "website", + } def commit(self): values = self.get_dirty_fields() args = {"name": self.name} - self.gitea.requests_patch( - Organization.API_OBJECT.format(**args), data=values - ) + self.gitea.requests_patch(Organization.API_OBJECT.format(**args), data=values) self.dirty_fields = {} def create_repo( - self, - repoName: str, - description: str = "", - private: bool = False, - autoInit=True, - gitignores: str = None, - license: str = None, - readme: str = "Default", - issue_labels: str = None, - default_branch="master", + self, + repoName: str, + description: str = "", + private: bool = False, + autoInit=True, + gitignores: str = None, + license: str = None, + readme: str = "Default", + issue_labels: str = None, + default_branch="master", ): """Create an organization Repository @@ -83,7 +88,9 @@ class Organization(ApiObject): }, ) if "id" in result: - self.gitea.logger.info("Successfully created Repository %s " % result["name"]) + self.gitea.logger.info( + "Successfully created Repository %s " % result["name"] + ) else: self.gitea.logger.error(result["message"]) raise Exception("Repository not created... (gitea: %s)" % result["message"]) @@ -108,7 +115,8 @@ class Organization(ApiObject): ) teams = [Team.parse_response(self.gitea, result) for result in results] # organisation seems to be missing using this request, so we add org manually - for t in teams: setattr(t, "_organization", self) + for t in teams: + setattr(t, "_organization", self) return teams def get_team(self, name) -> "Team": @@ -139,7 +147,7 @@ class Organization(ApiObject): self.gitea.requests_delete(path) def delete(self): - """ Delete this Organization. Invalidates this Objects data. Also deletes all Repositories owned by the User""" + """Delete this Organization. Invalidates this Objects data. Also deletes all Repositories owned by the User""" for repo in self.get_repositories(): repo.delete() self.gitea.requests_delete(Organization.API_OBJECT.format(name=self.username)) @@ -167,7 +175,8 @@ class User(ApiObject): self._emails = [] def __eq__(self, other): - if not isinstance(other, User): return False + if not isinstance(other, User): + return False return self.gitea == other.gitea and self.id == other.id def __hash__(self): @@ -179,7 +188,7 @@ class User(ApiObject): return self._emails @classmethod - def request(cls, gitea: 'Gitea', name: str) -> "User": + def request(cls, gitea: "Gitea", name: str) -> "User": api_object = cls._request(gitea, {"name": name}) return api_object @@ -217,16 +226,16 @@ class User(ApiObject): self.dirty_fields = {} def create_repo( - self, - repoName: str, - description: str = "", - private: bool = False, - autoInit=True, - gitignores: str = None, - license: str = None, - readme: str = "Default", - issue_labels: str = None, - default_branch="master", + self, + repoName: str, + description: str = "", + private: bool = False, + autoInit=True, + gitignores: str = None, + license: str = None, + readme: str = "Default", + issue_labels: str = None, + default_branch="master", ): """Create a user Repository @@ -249,31 +258,33 @@ class User(ApiObject): }, ) if "id" in result: - self.gitea.logger.info("Successfully created Repository %s " % result["name"]) + self.gitea.logger.info( + "Successfully created Repository %s " % result["name"] + ) else: self.gitea.logger.error(result["message"]) raise Exception("Repository not created... (gitea: %s)" % result["message"]) return Repository.parse_response(self, result) def get_repositories(self) -> List["Repository"]: - """ Get all Repositories owned by this User.""" + """Get all Repositories owned by this User.""" url = f"/users/{self.username}/repos" results = self.gitea.requests_get_paginated(url) return [Repository.parse_response(self.gitea, result) for result in results] def get_orgs(self) -> List[Organization]: - """ Get all Organizations this user is a member of.""" + """Get all Organizations this user is a member of.""" url = f"/users/{self.username}/orgs" results = self.gitea.requests_get_paginated(url) return [Organization.parse_response(self.gitea, result) for result in results] - def get_teams(self) -> List['Team']: + def get_teams(self) -> List["Team"]: url = f"/user/teams" results = self.gitea.requests_get_paginated(url, sudo=self) return [Team.parse_response(self.gitea, result) for result in results] - def get_accessible_repos(self) -> List['Repository']: - """ Get all Repositories accessible by the logged in User.""" + def get_accessible_repos(self) -> List["Repository"]: + """Get all Repositories accessible by the logged in User.""" results = self.gitea.requests_get("/user/repos", sudo=self) return [Repository.parse_response(self, result) for result in results] @@ -286,7 +297,7 @@ class User(ApiObject): self._email = mail["email"] def delete(self): - """ Deletes this User. Also deletes all Repositories he owns.""" + """Deletes this User. Also deletes all Repositories he owns.""" self.gitea.requests_delete(User.ADMIN_DELETE_USER % self.username) self.deleted = True @@ -300,7 +311,6 @@ class User(ApiObject): class Branch(ReadonlyApiObject): - def __init__(self, gitea): super().__init__(gitea) @@ -318,13 +328,15 @@ class Branch(ReadonlyApiObject): } @classmethod - def request(cls, gitea: 'Gitea', owner: str, repo: str, ref: str): + def request(cls, gitea: "Gitea", owner: str, repo: str, ref: str): return cls._request(gitea, {"owner": owner, "repo": repo, "ref": ref}) class Repository(ApiObject): API_OBJECT = """/repos/{owner}/{name}""" # , - REPO_IS_COLLABORATOR = """/repos/%s/%s/collaborators/%s""" # , , + REPO_IS_COLLABORATOR = ( + """/repos/%s/%s/collaborators/%s""" # , , + ) REPO_SEARCH = """/repos/search/%s""" # REPO_BRANCHES = """/repos/%s/%s/branches""" # , REPO_ISSUES = """/repos/{owner}/{repo}/issues""" # @@ -339,7 +351,8 @@ class Repository(ApiObject): super().__init__(gitea) def __eq__(self, other): - if not isinstance(other, Repository): return False + if not isinstance(other, Repository): + return False return self.owner == other.owner and self.name == other.name def __hash__(self): @@ -348,12 +361,13 @@ class Repository(ApiObject): _fields_to_parsers = { # dont know how to tell apart user and org as owner except form email being empty. "owner": lambda gitea, r: Organization.parse_response(gitea, r) - if r["email"] == "" else User.parse_response(gitea, r), + if r["email"] == "" + else User.parse_response(gitea, r), "updated_at": lambda gitea, t: Util.convert_time(t), } @classmethod - def request(cls, gitea: 'Gitea', owner: str, name: str): + def request(cls, gitea: "Gitea", owner: str, name: str): return cls._request(gitea, {"owner": owner, "name": name}) _patchable_fields = { @@ -391,7 +405,7 @@ class Repository(ApiObject): self.gitea.requests_patch(self.API_OBJECT.format(**args), data=values) self.dirty_fields = {} - def get_branches(self) -> List['Branch']: + def get_branches(self) -> List["Branch"]: """Get all the Branches of this Repository.""" results = self.gitea.requests_get( Repository.REPO_BRANCHES % (self.owner.username, self.name) @@ -431,7 +445,8 @@ class Repository(ApiObject): issues = [] data = {"state": state} results = self.gitea.requests_get_paginated( - Repository.REPO_ISSUES.format(owner=self.owner.username, repo=self.name), params=data + Repository.REPO_ISSUES.format(owner=self.owner.username, repo=self.name), + params=data, ) for result in results: issue = Issue.parse_response(self.gitea, result) @@ -467,14 +482,20 @@ class Repository(ApiObject): "title": title, } result = self.gitea.requests_post( - Repository.REPO_ISSUES.format(owner=self.owner.username, repo=self.name), data=data + Repository.REPO_ISSUES.format(owner=self.owner.username, repo=self.name), + data=data, ) return Issue.parse_response(self.gitea, result) - def create_milestone(self, title: str, description: str, due_date: str = None, state: str = "open") -> "Milestone": - url = Repository.REPO_MILESTONES.format(owner=self.owner.username, repo=self.name) + def create_milestone( + self, title: str, description: str, due_date: str = None, state: str = "open" + ) -> "Milestone": + url = Repository.REPO_MILESTONES.format( + owner=self.owner.username, repo=self.name + ) data = {"title": title, "description": description, "state": state} - if due_date: data["due_date"] = due_date + if due_date: + data["due_date"] = due_date result = self.gitea.requests_post(url, data=data) return Milestone.parse_response(self.gitea, result) @@ -528,11 +549,17 @@ class Repository(ApiObject): url = f"/repos/{self.owner.username}/{self.name}/collaborators/{user_name}" self.gitea.requests_delete(url) - def transfer_ownership(self, new_owner: Union["User", "Organization"], new_teams: Set["Team"] = frozenset()): + def transfer_ownership( + self, + new_owner: Union["User", "Organization"], + new_teams: Set["Team"] = frozenset(), + ): url = Repository.REPO_TRANSFER.format(owner=self.owner.username, repo=self.name) data = {"new_owner": new_owner.username} if isinstance(new_owner, Organization): - new_team_ids = [team.id for team in new_teams if team in new_owner.get_teams()] + new_team_ids = [ + team.id for team in new_teams if team in new_owner.get_teams() + ] data["team_ids"] = new_team_ids self.gitea.requests_post(url, data=data) # TODO: make sure this instance is either updated or discarded @@ -541,17 +568,25 @@ class Repository(ApiObject): """https://try.gitea.io/api/swagger#/repository/repoGetContentsList""" url = f"/repos/{self.owner.username}/{self.name}/contents" data = {"ref": commit.sha} if commit else {} - result = [Content.parse_response(self.gitea, f) for f in self.gitea.requests_get(url, data)] + result = [ + Content.parse_response(self.gitea, f) + for f in self.gitea.requests_get(url, data) + ] return result - def get_file_content(self, content: "Content", commit: "Commit" = None) -> Union[str, List["Content"]]: + def get_file_content( + self, content: "Content", commit: "Commit" = None + ) -> Union[str, List["Content"]]: """https://try.gitea.io/api/swagger#/repository/repoGetContents""" url = f"/repos/{self.owner.username}/{self.name}/contents/{content.path}" data = {"ref": commit.sha} if commit else {} if content.type == Content.FILE: return self.gitea.requests_get(url, data)["content"] else: - return [Content.parse_response(self.gitea, f) for f in self.gitea.requests_get(url, data)] + return [ + Content.parse_response(self.gitea, f) + for f in self.gitea.requests_get(url, data) + ] def create_file(self, file_path: str, content: str, data: dict = None): """https://try.gitea.io/api/swagger#/repository/repoCreateFile""" @@ -561,7 +596,9 @@ class Repository(ApiObject): data.update({"content": content}) return self.gitea.requests_post(url, data) - def change_file(self, file_path: str, file_sha: str, content: str, data: dict = None): + def change_file( + self, file_path: str, file_sha: str, content: str, data: dict = None + ): """https://try.gitea.io/api/swagger#/repository/repoCreateFile""" if not data: data = {} @@ -583,7 +620,8 @@ class Milestone(ApiObject): super().__init__(gitea) def __eq__(self, other): - if not isinstance(other, Milestone): return False + if not isinstance(other, Milestone): + return False return self.gitea == other.gitea and self.id == other.id def __hash__(self): @@ -612,17 +650,17 @@ class Milestone(ApiObject): } @classmethod - def request(cls, gitea: 'Gitea', owner: str, repo: str, number: str): + def request(cls, gitea: "Gitea", owner: str, repo: str, number: str): return cls._request(gitea, {"owner": owner, "repo": repo, "number": number}) class Comment(ApiObject): - def __init__(self, gitea): super().__init__(gitea) def __eq__(self, other): - if not isinstance(other, Comment): return False + if not isinstance(other, Comment): + return False return self.repo == other.repo and self.id == other.id def __hash__(self): @@ -636,24 +674,26 @@ class Comment(ApiObject): class Commit(ReadonlyApiObject): - def __init__(self, gitea): super().__init__(gitea) _fields_to_parsers = { # NOTE: api may return None for commiters that are no gitea users - "author": lambda gitea, u: User.parse_response(gitea, u) if u else None + "author": lambda gitea, u: User.parse_response(gitea, u) + if u + else None } def __eq__(self, other): - if not isinstance(other, Commit): return False + if not isinstance(other, Commit): + return False return self.sha == other.sha def __hash__(self): return hash(self.sha) @classmethod - def parse_response(cls, gitea, result) -> 'Commit': + def parse_response(cls, gitea, result) -> "Commit": commit_cache = result["commit"] api_object = cls(gitea) cls._initialize(gitea, api_object, result) @@ -675,7 +715,8 @@ class Issue(ApiObject): super().__init__(gitea) def __eq__(self, other): - if not isinstance(other, Issue): return False + if not isinstance(other, Issue): + return False return self.repo == other.repo and self.id == other.id def __hash__(self): @@ -688,7 +729,7 @@ class Issue(ApiObject): "assignees": lambda gitea, us: [User.parse_response(gitea, u) for u in us], "state": lambda gitea, s: Issue.CLOSED if s == "closed" else Issue.OPENED, # Repository in this request is just a "RepositoryMeta" record, thus request whole object - "repository": lambda gitea, r: Repository.request(gitea, r["owner"], r["name"]) + "repository": lambda gitea, r: Repository.request(gitea, r["owner"], r["name"]), } _parsers_to_fields = { @@ -707,13 +748,19 @@ class Issue(ApiObject): def commit(self): values = self.get_dirty_fields() - args = {"owner": self.repository.owner.username, "repo": self.repository.name, "index": self.number} + args = { + "owner": self.repository.owner.username, + "repo": self.repository.name, + "index": self.number, + } self.gitea.requests_patch(Issue.API_OBJECT.format(**args), data=values) self.dirty_fields = {} @classmethod - def request(cls, gitea: 'Gitea', owner: str, repo: str, number: str): - api_object = cls._request(gitea, {"owner": owner, "repo": repo, "index": number}) + def request(cls, gitea: "Gitea", owner: str, repo: str, number: str): + api_object = cls._request( + gitea, {"owner": owner, "repo": repo, "index": number} + ) return api_object @classmethod @@ -774,7 +821,8 @@ class Team(ApiObject): super().__init__(gitea) def __eq__(self, other): - if not isinstance(other, Team): return False + if not isinstance(other, Team): + return False return self.organization == other.organization and self.id == other.id def __hash__(self): @@ -813,12 +861,12 @@ class Team(ApiObject): self.gitea.requests_put(Team.ADD_REPO % (self.id, org, repo.name)) def get_members(self): - """ Get all users assigned to the team. """ + """Get all users assigned to the team.""" results = self.gitea.requests_get(Team.GET_MEMBERS % self.id) return [User.parse_response(self.gitea, result) for result in results] def get_repos(self): - """ Get all repos of this Team.""" + """Get all repos of this Team.""" results = self.gitea.requests_get(Team.GET_REPOS % self.id) return [Repository.parse_response(self.gitea, result) for result in results] @@ -838,8 +886,11 @@ class Content(ReadonlyApiObject): super().__init__(gitea) def __eq__(self, other): - if not isinstance(other, Team): return False - return self.repo == self.repo and self.sha == other.sha and self.name == other.name + if not isinstance(other, Team): + return False + return ( + self.repo == self.repo and self.sha == other.sha and self.name == other.name + ) def __hash__(self): return hash(self.repo) ^ hash(self.sha) ^ hash(self.name) @@ -848,7 +899,7 @@ class Content(ReadonlyApiObject): class Util: @staticmethod def convert_time(time: str) -> datetime: - """ Parsing of strange Gitea time format ("%Y-%m-%dT%H:%M:%S:%z" but with ":" in time zone notation)""" + """Parsing of strange Gitea time format ("%Y-%m-%dT%H:%M:%S:%z" but with ":" in time zone notation)""" try: return datetime.strptime(time[:-3] + "00", "%Y-%m-%dT%H:%M:%S%z") except ValueError: diff --git a/gitea/baseapiobject.py b/gitea/baseapiobject.py index 14073ac..34b0789 100644 --- a/gitea/baseapiobject.py +++ b/gitea/baseapiobject.py @@ -1,8 +1,11 @@ -from .exceptions import ObjectIsInvalid, MissiongEqualyImplementation, RawRequestEndpointMissing +from .exceptions import ( + ObjectIsInvalid, + MissiongEqualyImplementation, + RawRequestEndpointMissing, +) class ReadonlyApiObject: - def __init__(self, gitea): self.gitea = gitea self.deleted = False # set if .delete was called, so that an exception is risen @@ -35,7 +38,7 @@ class ReadonlyApiObject: @classmethod def _get_gitea_api_object(cls, gitea, args): - """Retrieving an object always as GET_API_OBJECT """ + """Retrieving an object always as GET_API_OBJECT""" return gitea.requests_get(cls.API_OBJECT.format(**args)) @classmethod @@ -61,8 +64,7 @@ class ReadonlyApiObject: def _add_read_property(cls, name, value, api_object): if not hasattr(api_object, name): setattr(api_object, "_" + name, value) - prop = property( - (lambda n: lambda self: self._get_var(n))(name)) + prop = property((lambda n: lambda self: self._get_var(n))(name)) setattr(cls, name, prop) else: raise AttributeError(f"Attribute {name} already exists on api object.") @@ -107,7 +109,8 @@ class ApiObject(ReadonlyApiObject): setattr(api_object, "_" + name, value) prop = property( (lambda n: lambda self: self._get_var(n))(name), - (lambda n: lambda self, v: self.__set_var(n, v))(name)) + (lambda n: lambda self, v: self.__set_var(n, v))(name), + ) setattr(cls, name, prop) def __set_var(self, name, i): diff --git a/gitea/exceptions.py b/gitea/exceptions.py index abeab50..dfb90d1 100644 --- a/gitea/exceptions.py +++ b/gitea/exceptions.py @@ -17,6 +17,7 @@ class ConflictException(Exception): class RawRequestEndpointMissing(Exception): """This ApiObject can only be obtained through other api objects and does not have diret .request method.""" + pass @@ -25,4 +26,5 @@ class MissiongEqualyImplementation(Exception): Each Object obtained from the gitea api must be able to check itself for equality in relation to its fields obtained from gitea. Risen if an api object is lacking the proper implementation. """ + pass diff --git a/gitea/gitea.py b/gitea/gitea.py index 6071fd4..1eaa70e 100644 --- a/gitea/gitea.py +++ b/gitea/gitea.py @@ -11,7 +11,8 @@ from .exceptions import NotFoundException, ConflictException, AlreadyExistsExcep class Gitea: - """ Object to establish a session with Gitea. """ + """Object to establish a session with Gitea.""" + ADMIN_CREATE_USER = """/admin/users""" GET_USERS_ADMIN = """/admin/users""" ADMIN_REPO_CREATE = """/admin/users/%s/repos""" # @@ -21,14 +22,9 @@ class Gitea: CREATE_TEAM = """/orgs/%s/teams""" # def __init__( - self, - gitea_url: str, - token_text=None, - auth=None, - verify=True, - log_level="INFO" - ): - """ Initializing Gitea-instance + self, gitea_url: str, token_text=None, auth=None, verify=True, log_level="INFO" + ): + """Initializing Gitea-instance Args: gitea_url (str): The Gitea instance URL. @@ -60,7 +56,6 @@ class Gitea: if not verify: urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning) - def __get_url(self, endpoint): url = self.url + "/api/v1" + endpoint self.logger.debug("Url: %s" % url) @@ -68,7 +63,7 @@ class Gitea: @staticmethod def parse_result(result) -> Dict: - """ Parses the result-JSON to a dict. """ + """Parses the result-JSON to a dict.""" if result.text and len(result.text) > 3: return json.loads(result.text) return {} @@ -78,19 +73,25 @@ class Gitea: combined_params.update(params) if sudo: combined_params["sudo"] = sudo.username - request = self.requests.get(self.__get_url(endpoint), headers=self.headers, params=combined_params) + request = self.requests.get( + self.__get_url(endpoint), headers=self.headers, params=combined_params + ) if request.status_code not in [200, 201]: message = f"Received status code: {request.status_code} ({request.url})" if request.status_code in [404]: raise NotFoundException(message) if request.status_code in [403]: - raise Exception(f"Unauthorized: {request.url} - Check your permissions and try again! ({message})") + raise Exception( + f"Unauthorized: {request.url} - Check your permissions and try again! ({message})" + ) if request.status_code in [409]: raise ConflictException(message) raise Exception(message) return self.parse_result(request) - def requests_get_paginated(self, endpoint: str, params=frozendict(), sudo=None, page_key: str = "page"): + def requests_get_paginated( + self, endpoint: str, params=frozendict(), sudo=None, page_key: str = "page" + ): page = 1 combined_params = {} combined_params.update(params) @@ -106,7 +107,9 @@ class Gitea: def requests_put(self, endpoint: str, data: dict = None): if not data: data = {} - request = self.requests.put(self.__get_url(endpoint), headers=self.headers, data=json.dumps(data)) + request = self.requests.put( + self.__get_url(endpoint), headers=self.headers, data=json.dumps(data) + ) if request.status_code not in [200, 204]: message = f"Received status code: {request.status_code} ({request.url}) {request.text}" self.logger.error(message) @@ -120,21 +123,34 @@ class Gitea: raise Exception(message) def requests_post(self, endpoint: str, data: dict): - request = self.requests.post(self.__get_url(endpoint), headers=self.headers, data=json.dumps(data)) + request = self.requests.post( + self.__get_url(endpoint), headers=self.headers, data=json.dumps(data) + ) if request.status_code not in [200, 201, 202]: - if ("already exists" in request.text or "e-mail already in use" in request.text): + if ( + "already exists" in request.text + or "e-mail already in use" in request.text + ): self.logger.warning(request.text) raise AlreadyExistsException() - self.logger.error(f"Received status code: {request.status_code} ({request.url})") + self.logger.error( + f"Received status code: {request.status_code} ({request.url})" + ) self.logger.error(f"With info: {data} ({self.headers})") self.logger.error(f"Answer: {request.text}") - raise Exception(f"Received status code: {request.status_code} ({request.url}), {request.text}") + raise Exception( + f"Received status code: {request.status_code} ({request.url}), {request.text}" + ) return self.parse_result(request) def requests_patch(self, endpoint: str, data: dict): - request = self.requests.patch(self.__get_url(endpoint), headers=self.headers, data=json.dumps(data)) + request = self.requests.patch( + self.__get_url(endpoint), headers=self.headers, data=json.dumps(data) + ) if request.status_code not in [200, 201]: - error_message = f"Received status code: {request.status_code} ({request.url}) {data}" + error_message = ( + f"Received status code: {request.status_code} ({request.url}) {data}" + ) self.logger.error(error_message) raise Exception(error_message) return self.parse_result(request) @@ -175,17 +191,17 @@ class Gitea: return None def create_user( - self, - user_name: str, - email: str, - password: str, - full_name: str = None, - login_name: str = None, - change_pw=True, - send_notify=True, - source_id=0, + self, + user_name: str, + email: str, + password: str, + full_name: str = None, + login_name: str = None, + change_pw=True, + send_notify=True, + source_id=0, ): - """ Create User. + """Create User. Throws: AlreadyExistsException, if the User exists already Exception, if something else went wrong. @@ -222,19 +238,19 @@ class Gitea: return user def create_repo( - self, - repoOwner: Union[User, Organization], - repoName: str, - description: str = "", - private: bool = False, - autoInit=True, - gitignores: str = None, - license: str = None, - readme: str = "Default", - issue_labels: str = None, - default_branch="master", + self, + repoOwner: Union[User, Organization], + repoName: str, + description: str = "", + private: bool = False, + autoInit=True, + gitignores: str = None, + license: str = None, + readme: str = "Default", + issue_labels: str = None, + default_branch="master", ): - """ Create a Repository as the administrator + """Create a Repository as the administrator Throws: AlreadyExistsException: If the Repository exists already. @@ -269,13 +285,13 @@ class Gitea: return Repository.parse_response(self, result) def create_org( - self, - owner: User, - orgName: str, - description: str, - location="", - website="", - full_name="", + self, + owner: User, + orgName: str, + description: str, + location="", + website="", + full_name="", ): assert isinstance(owner, User) result = self.requests_post( @@ -303,24 +319,24 @@ class Gitea: return Organization.parse_response(self, result) def create_team( - self, - org: Organization, - name: str, - description: str = "", - permission: str = "read", - can_create_org_repo: bool = False, - includes_all_repositories: bool = False, - units=( - "repo.code", - "repo.issues", - "repo.ext_issues", - "repo.wiki", - "repo.pulls", - "repo.releases", - "repo.ext_wiki", - ), + self, + org: Organization, + name: str, + description: str = "", + permission: str = "read", + can_create_org_repo: bool = False, + includes_all_repositories: bool = False, + units=( + "repo.code", + "repo.issues", + "repo.ext_issues", + "repo.wiki", + "repo.pulls", + "repo.releases", + "repo.ext_wiki", + ), ): - """ Creates a Team. + """Creates a Team. Args: org (Organization): Organization the Team will be part of. diff --git a/setup.py b/setup.py index 27c94d7..0632694 100644 --- a/setup.py +++ b/setup.py @@ -1,35 +1,31 @@ from setuptools import setup, find_packages -with open('README.md') as readme_file: +with open("README.md") as readme_file: README = readme_file.read() setup_args = dict( - name='py-gitea', - version='0.2.6', - description='A python wrapper for the Gitea API', + name="py-gitea", + version="0.2.6", + description="A python wrapper for the Gitea API", long_description_content_type="text/markdown", long_description=README, - license='MIT', + license="MIT", packages=find_packages(), - author='Vincent Langenfeld ', - author_email='langenfv@tf.uni-freiburg.de', - keywords=['Gitea','api','wrapper'], - url='https://github.com/Langenfeld/py-gitea', - download_url='https://pypi.org/project/py-gitea/' + author="Vincent Langenfeld ", + author_email="langenfv@tf.uni-freiburg.de", + keywords=["Gitea", "api", "wrapper"], + url="https://github.com/Langenfeld/py-gitea", + download_url="https://pypi.org/project/py-gitea/", ) install_requires = [ - 'requests', - 'frozendict', + "requests", + "frozendict", ] -extras_require = { - 'test': ['pytest'] -} +extras_require = {"test": ["pytest"]} -if __name__ == '__main__': +if __name__ == "__main__": setup( - **setup_args, - install_requires=install_requires, - extras_require=extras_require + **setup_args, install_requires=install_requires, extras_require=extras_require ) diff --git a/tests/conftest.py b/tests/conftest.py index 607c173..add6ef8 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -13,6 +13,7 @@ import pytest from gitea import Gitea + @pytest.fixture def instance(scope="module"): try: diff --git a/tests/test_api.py b/tests/test_api.py index 46aa82e..89852d8 100644 --- a/tests/test_api.py +++ b/tests/test_api.py @@ -6,6 +6,7 @@ import uuid from gitea import Gitea, User, Organization, Team, Repository, Issue, Milestone from gitea import NotFoundException, AlreadyExistsException + # put a ".token" file into your directory containg only the token for gitea @pytest.fixture def instance(scope="module"): @@ -22,10 +23,13 @@ def instance(scope="module"): - Token at .token \ ?" + # make up some fresh names for the tests run test_org = "org_" + uuid.uuid4().hex[:8] test_user = "user_" + uuid.uuid4().hex[:8] -test_team = "team_" + uuid.uuid4().hex[:8] # team names seem to have a rather low max lenght +test_team = ( + "team_" + uuid.uuid4().hex[:8] +) # team names seem to have a rather low max lenght test_repo = "repo_" + uuid.uuid4().hex[:8] @@ -65,6 +69,7 @@ def test_create_user(instance): assert type(user.id) is int assert user.is_admin is False + def test_change_user(instance): user = instance.get_user_by_name(test_user) location = "a house" @@ -72,7 +77,7 @@ def test_change_user(instance): new_fullname = "Other Test Full Name" user.full_name = new_fullname user.commit(user.username, 0) - del(user) + del user user = instance.get_user_by_name(test_user) assert user.full_name == new_fullname assert user.location == location @@ -153,6 +158,7 @@ def test_list_branches(instance): master = [b for b in branches if b.name == "master"] assert len(master) > 0 + def test_list_files_and_content(instance): org = Organization.request(instance, test_org) repo = org.get_repository(test_repo) @@ -165,13 +171,13 @@ def test_list_files_and_content(instance): assert len(readme_content) > 0 assert "descr" in str(base64.b64decode(readme_content)) + def test_create_file(instance): TESTFILE_CONENTE = "TestStringFileContent" - TESTFILE_CONENTE_B64 = base64.b64encode(bytes(TESTFILE_CONENTE, 'utf-8')) + TESTFILE_CONENTE_B64 = base64.b64encode(bytes(TESTFILE_CONENTE, "utf-8")) org = Organization.request(instance, test_org) repo = org.get_repository(test_repo) - repo.create_file("testfile.md", - content = TESTFILE_CONENTE_B64.decode("ascii")) + repo.create_file("testfile.md", content=TESTFILE_CONENTE_B64.decode("ascii")) # test if putting was successful content = repo.get_git_content() readmes = [c for c in content if c.name == "testfile.md"] @@ -180,17 +186,19 @@ def test_create_file(instance): assert len(readme_content) > 0 assert TESTFILE_CONENTE in str(base64.b64decode(readme_content)) + def test_change_file(instance): TESTFILE_CONENTE = "TestStringFileContent with changed content now" - TESTFILE_CONENTE_B64 = base64.b64encode(bytes(TESTFILE_CONENTE, 'utf-8')) + TESTFILE_CONENTE_B64 = base64.b64encode(bytes(TESTFILE_CONENTE, "utf-8")) org = Organization.request(instance, test_org) repo = org.get_repository(test_repo) - #figure out the sha of the file to change + # figure out the sha of the file to change content = repo.get_git_content() readmes = [c for c in content if c.name == "testfile.md"] # change - repo.change_file("testfile.md", readmes[0].sha, - content = TESTFILE_CONENTE_B64.decode("ascii")) + repo.change_file( + "testfile.md", readmes[0].sha, content=TESTFILE_CONENTE_B64.decode("ascii") + ) # test if putting was successful content = repo.get_git_content() readmes = [c for c in content if c.name == "testfile.md"] @@ -199,6 +207,7 @@ def test_change_file(instance): assert len(readme_content) > 0 assert TESTFILE_CONENTE in str(base64.b64decode(readme_content)) + def test_create_branch(instance): org = Organization.request(instance, test_org) repo = org.get_repository(test_repo) @@ -207,6 +216,7 @@ def test_create_branch(instance): assert len(master) > 0 repo.add_branch(master[0], "test_branch") + def test_create_team(instance): org = Organization.request(instance, test_org) team = instance.create_team(org, test_team, "descr") @@ -214,6 +224,7 @@ def test_create_team(instance): assert team.description == "descr" assert team.organization == org + def test_patch_team(instance): fields = { "can_create_org_repo": True, @@ -238,12 +249,16 @@ def test_request_team(instance): team2 = Team.request(instance, team.id) assert team.name == team2.name + def test_create_milestone(instance): - org = Organization.request(instance, test_org) - repo = org.get_repository(test_repo) - ms = repo.create_milestone("I love this Milestone", "Find an otter to adopt this milestone") - assert isinstance(ms, Milestone) - assert ms.title == "I love this Milestone" + org = Organization.request(instance, test_org) + repo = org.get_repository(test_repo) + ms = repo.create_milestone( + "I love this Milestone", "Find an otter to adopt this milestone" + ) + assert isinstance(ms, Milestone) + assert ms.title == "I love this Milestone" + def test_user_teams(instance): org = Organization.request(instance, test_org) @@ -253,11 +268,13 @@ def test_user_teams(instance): teams = user.get_teams() assert team in teams + def test_get_accessible_repositories(instance): user = instance.get_user_by_name(test_user) repos = user.get_accessible_repos() assert len(repos) > 0 + def test_create_issue(instance): org = Organization.request(instance, test_org) repo = Repository.request(instance, org.username, test_repo) @@ -266,12 +283,13 @@ def test_create_issue(instance): assert issue.title == "TestIssue" assert issue.body == "Body text with this issue" + def test_hashing(instance): - #just call the hash function of each object to see if something bad happens + # just call the hash function of each object to see if something bad happens org = Organization.request(instance, test_org) team = org.get_team(test_team) user = instance.get_user_by_name(test_user) - #TODO test for milestones (Todo: add milestone adding) + # TODO test for milestones (Todo: add milestone adding) repo = org.get_repositories()[0] milestone = repo.create_milestone("mystone", "this is only a teststone") issue = repo.get_issues()[0] @@ -284,7 +302,9 @@ def test_change_issue(instance): org = Organization.request(instance, test_org) repo = org.get_repositories()[0] ms_title = "othermilestone" - issue = Issue.create_issue(instance, repo, "IssueTestissue with Testinput", "asdf2332") + issue = Issue.create_issue( + instance, repo, "IssueTestissue with Testinput", "asdf2332" + ) new_body = "some new description with some more of that char stuff :)" issue.body = new_body issue.commit() @@ -300,8 +320,17 @@ def test_change_issue(instance): assert issue3.milestone is not None assert issue3.milestone.description == "this is only a teststone2" issues = repo.get_issues() - assert len([issue for issue in issues - if issue.milestone is not None and issue.milestone.title == ms_title]) > 0 + assert ( + len( + [ + issue + for issue in issues + if issue.milestone is not None and issue.milestone.title == ms_title + ] + ) + > 0 + ) + def test_team_get_org(instance): org = Organization.request(instance, test_org) @@ -309,6 +338,7 @@ def test_team_get_org(instance): teams = user.get_teams() assert org.username == teams[0].organization.name + def test_delete_repo_userowned(instance): user = User.request(instance, test_user) repo = Repository.request(instance, user.username, test_repo) @@ -316,6 +346,7 @@ def test_delete_repo_userowned(instance): with pytest.raises(NotFoundException) as e: Repository.request(instance, test_user, test_repo) + def test_secundary_email(instance): SECONDARYMAIL = "secondarytest@test.org" # set up with real email sec_user = instance.get_user_by_email(SECONDARYMAIL) @@ -334,18 +365,21 @@ def test_delete_repo_orgowned(instance): def test_change_repo_ownership_org(instance): old_org = Organization.request(instance, test_org) user = User.request(instance, test_user) - new_org = instance.create_org(user,test_org+"_repomove", "Org for testing moving repositories") + new_org = instance.create_org( + user, test_org + "_repomove", "Org for testing moving repositories" + ) new_team = instance.create_team(new_org, test_team + "_repomove", "descr") - repo_name = test_repo+"_repomove" - repo = instance.create_repo(old_org, repo_name , "descr") + repo_name = test_repo + "_repomove" + repo = instance.create_repo(old_org, repo_name, "descr") repo.transfer_ownership(new_org, set([new_team])) assert repo_name not in [repo.name for repo in old_org.get_repositories()] assert repo_name in [repo.name for repo in new_org.get_repositories()] + def test_change_repo_ownership_user(instance): old_org = Organization.request(instance, test_org) user = User.request(instance, test_user) - repo_name = test_repo+"_repomove" + repo_name = test_repo + "_repomove" repo = instance.create_repo(old_org, repo_name, "descr") repo.transfer_ownership(user) assert repo_name not in [repo.name for repo in old_org.get_repositories()] @@ -363,6 +397,7 @@ def test_delete_team(instance): with pytest.raises(NotFoundException) as e: team = org.get_team(test_team) + def test_delete_teams(instance): org = Organization.request(instance, test_org) repos = org.get_repositories() @@ -371,6 +406,7 @@ def test_delete_teams(instance): repos = org.get_repositories() assert len(repos) == 0 + def test_delete_org(instance): org = Organization.request(instance, test_org) org.delete() diff --git a/tests/test_api_longtests.py b/tests/test_api_longtests.py index d8a0c4b..d1eccfb 100644 --- a/tests/test_api_longtests.py +++ b/tests/test_api_longtests.py @@ -4,6 +4,7 @@ import uuid from gitea import Gitea, User, Organization, Team, Repository, Issue from gitea import NotFoundException, AlreadyExistsException + # put a ".token" file into your directory containg only the token for gitea @pytest.fixture def instance(scope="module"): @@ -20,15 +21,20 @@ def instance(scope="module"): - Token at .token \ ?" + # make up some fresh names for the tests run test_org = "org_" + uuid.uuid4().hex[:8] test_user = "user_" + uuid.uuid4().hex[:8] -test_team = "team_" + uuid.uuid4().hex[:8] # team names seem to have a rather low max lenght +test_team = ( + "team_" + uuid.uuid4().hex[:8] +) # team names seem to have a rather low max lenght test_repo = "repo_" + uuid.uuid4().hex[:8] def test_list_repos(instance): - user = instance.create_user(test_user, test_user + "@example.org", "abcdefg1.23AB", send_notify=False) + user = instance.create_user( + test_user, test_user + "@example.org", "abcdefg1.23AB", send_notify=False + ) org = instance.create_org(user, test_org, "some Description for longtests") repos = org.get_repositories() assert len(repos) == 0 @@ -41,8 +47,15 @@ def test_list_repos(instance): def test_list_issue(instance): org = Organization.request(instance, test_org) - repo = instance.create_repo(org, test_repo, "Testing a huge number of Issues and how they are listed") + repo = instance.create_repo( + org, test_repo, "Testing a huge number of Issues and how they are listed" + ) for x in range(0, 100): - Issue.create_issue(instance, repo, "TestIssue" + str(x), "We will be too many to be listed on one page") + Issue.create_issue( + instance, + repo, + "TestIssue" + str(x), + "We will be too many to be listed on one page", + ) issues = repo.get_issues() assert len(issues) > 98