mirror
/
py-gitea
kopia lustrzana https://github.com/Langenfeld/py-gitea
1
0
Forkuj 0
Kod Zgłoszenia Packages Projekty Wydania Wiki Aktywność

improving editing api objects

pull/12/head
Langenfeld 2021-11-16 14:21:48 +01:00
rodzic d1430acf8a
commit 94261487cf
4 zmienionych plików z 100 dodań i 44 usunięć
Pokaż statystyki Ściągnij plik aktualizacji Ściągnij plik porównania Expand all files Collapse all files

74
gitea/apiobject.py
Wyświetl plik

@ -31,13 +31,13 @@ class Organization(ApiObject):
return cls._request(gitea, {"name": name})
@classmethod
def parse_response(cls, gitea, result):
def parse_response(cls, gitea, result) -> 'Organization':
api_object = super().parse_response(gitea, result)
# add "name" field to make this behave similar to users
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()
@ -141,7 +141,7 @@ class User(ApiObject):
api_object = cls._request(gitea, {"name": name})
return api_object
patchable_fields = {
_patchable_fields = {
"active",
"admin",
"allow_create_organization",
@ -231,7 +231,7 @@ class Branch(ReadonlyApiObject):
def __hash__(self):
return hash(self.commit["id"]) ^ hash(self.name)
fields_to_parsers = {
_fields_to_parsers = {
# This is not a commit object
#"commit": lambda gitea, c: Commit.parse_response(gitea, c)
}
@ -246,7 +246,7 @@ class Repository(ApiObject):
REPO_IS_COLLABORATOR = """/repos/%s/%s/collaborators/%s""" # <owner>, <reponame>, <username>
REPO_SEARCH = """/repos/search/%s""" # <reponame>
REPO_BRANCHES = """/repos/%s/%s/branches""" # <owner>, <reponame>
REPO_ISSUES = """/repos/%s/%s/issues""" # <owner, reponame>
REPO_ISSUES = """/repos/{owner}/{repo}/issues""" # <owner, reponame>
REPO_DELETE = """/repos/%s/%s""" # <owner>, <reponame>
REPO_TIMES = """/repos/%s/%s/times""" # <owner>, <reponame>
REPO_USER_TIME = """/repos/%s/%s/times/%s""" # <owner>, <reponame>, <username>
@ -266,11 +266,10 @@ class Repository(ApiObject):
def __hash__(self):
return hash(self.owner) ^ hash(self.name)
fields_to_parsers = {
_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),
}
@ -278,7 +277,7 @@ class Repository(ApiObject):
def request(cls, gitea, owner, name):
return cls._request(gitea, {"owner": owner, "name": name})
patchable_fields = {
_patchable_fields = {
"allow_merge_commits",
"allow_rebase",
"allow_rebase_explicit",
@ -333,16 +332,15 @@ class Repository(ApiObject):
"""Get issues of state Issue.open or Issue.closed of a repository."""
assert state in [Issue.OPENED, Issue.CLOSED]
issues = []
# "page": -1 is returning _all_ issues instead of pages. Hopefully this is intended behaviour.
data = {"page": -1, "state": state}
results = self.gitea.requests_get(
Repository.REPO_ISSUES % (self.owner.username, self.name), params=data
data = {"state": state}
results = self.gitea.requests_get_paginated(
Repository.REPO_ISSUES.format(owner=self.owner.username, repo=self.name), params=data
)
for result in results:
issue = Issue.parse_response(self.gitea, result)
# adding data not contained in the issue answer
setattr(issue, "repo", self.name)
setattr(issue, "owner", self.owner)
Issue._add_read_property("repo", self, issue)
Issue._add_read_property("owner", self.owner, issue)
issues.append(issue)
return issues
@ -372,7 +370,7 @@ class Repository(ApiObject):
"title": title,
}
result = self.gitea.requests_post(
Repository.REPO_ISSUES % (self.owner.username, self.name), data=data
Repository.REPO_ISSUES.format(owner=self.owner.username, repo=self.name), data=data
)
return Issue.parse_response(self.gitea, result)
@ -478,12 +476,12 @@ class Milestone(ApiObject):
def __hash__(self):
return hash(self.gitea) ^ hash(self.id)
fields_to_parsers = {
_fields_to_parsers = {
"closed_at": lambda gitea, t: Util.convert_time(t),
"due_on": lambda gitea, t: Util.convert_time(t),
}
patchable_fields = {
_patchable_fields = {
"allow_merge_commits",
"allow_rebase",
"allow_rebase_explicit",
@ -517,7 +515,7 @@ class Comment(ApiObject):
def __hash__(self):
return hash(self.repo) ^ hash(self.id)
fields_to_parsers = {
_fields_to_parsers = {
"user": lambda gitea, r: User.parse_response(gitea, r),
"created_at": lambda gitea, t: Util.convert_time(t),
"updated_at": lambda gitea, t: Util.convert_time(t),
@ -528,7 +526,7 @@ class Commit(ReadonlyApiObject):
def __init__(self, gitea):
super(Commit, self).__init__(gitea)
fields_to_parsers = {
_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
}
@ -551,7 +549,7 @@ class Commit(ReadonlyApiObject):
class Issue(ApiObject):
API_OBJECT = """/repos/{owner}/{repo}/issues/{number}""" # <owner, repo, index>
API_OBJECT = """/repos/{owner}/{repo}/issues/{index}""" # <owner, repo, index>
GET_TIME = """/repos/%s/%s/issues/%s/times""" # <owner, repo, index>
GET_COMMENTS = """/repos/%s/%s/issues/comments"""
CREATE_ISSUE = """/repos/{owner}/{repo}/issues"""
@ -569,15 +567,21 @@ class Issue(ApiObject):
def __hash__(self):
return hash(self.repo) ^ hash(self.id)
fields_to_parsers = {
_fields_to_parsers = {
"milestone": lambda gitea, m: Milestone.parse_response(gitea, m),
"user": lambda gitea, u: User.parse_response(gitea, u),
"assignee": lambda gitea, u: User.parse_response(gitea, u),
"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"])
}
patchable_fields = {
_parsers_to_fields = {
"milestone": lambda m: m.id,
}
_patchable_fields = {
"assignee",
"assignees",
"body",
@ -587,9 +591,15 @@ class Issue(ApiObject):
"title",
}
def commit(self):
values = self.get_dirty_fields()
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, owner, repo, number):
api_object = cls._request(gitea, {"owner": owner, "repo": repo, "number": number})
api_object = cls._request(gitea, {"owner": owner, "repo": repo, "index": number})
return api_object
@classmethod
@ -601,7 +611,7 @@ class Issue(ApiObject):
def get_time_sum(self, user: User) -> int:
results = self.gitea.requests_get(
Issue.GET_TIME % (self.owner.username, self.repo, self.number)
Issue.GET_TIME % (self.owner.username, self.repo.name, self.number)
)
return sum(
result["time"]
@ -611,22 +621,22 @@ class Issue(ApiObject):
def get_times(self) -> Optional[Dict]:
return self.gitea.requests_get(
Issue.GET_TIME % (self.owner.username, self.repo, self.number)
Issue.GET_TIME % (self.owner.username, self.repository.name, self.number)
)
def delete_time(self, time_id: str):
path = f"/repos/{self.owner.username}/{self.repo}/issues/{self.number}/times/{time_id}"
path = f"/repos/{self.owner.username}/{self.repository.name}/issues/{self.number}/times/{time_id}"
self.gitea.requests_delete(path)
def add_time(self, time: int, created: str = None, user_name: str = None):
path = f"/repos/{self.owner.username}/{self.repo}/issues/{self.number}/times"
def add_time(self, time: int, created: str = None, user_name: User = None):
path = f"/repos/{self.owner.username}/{self.repository.name}/issues/{self.number}/times"
self.gitea.requests_post(
path, data={"created": created, "time": int(time), "user_name": user_name}
)
def get_comments(self) -> List[ApiObject]:
results = self.gitea.requests_get(
Issue.GET_COMMENTS % (self.owner.username, self.repo)
Issue.GET_COMMENTS % (self.owner.username, self.repo.name)
)
allProjectComments = [
Comment.parse_response(self.gitea, result) for result in results
@ -657,7 +667,7 @@ class Team(ApiObject):
def __hash__(self):
return hash(self.organization) ^ hash(self.id)
fields_to_parsers = {
_fields_to_parsers = {
"organization": lambda gitea, o: Organization.parse_response(gitea, o)
}
@ -665,7 +675,7 @@ class Team(ApiObject):
def request(cls, gitea, organization, team):
return cls._request(gitea, {"id": id})
patchable_fields = {"description", "name", "permission", "units"}
_patchable_fields = {"description", "name", "permission", "units"}
def add_user(self, user: User):
self.gitea.requests_put(Team.ADD_USER % (self.id, user.login))

29
gitea/baseapiobject.py
Wyświetl plik

@ -17,7 +17,7 @@ class ReadonlyApiObject:
"""Hash only fields that are part of the gitea-data identity"""
raise MissiongEqualyImplementation()
fields_to_parsers = {}
_fields_to_parsers = {}
@classmethod
def request(cls, gitea, id):
@ -51,12 +51,12 @@ class ReadonlyApiObject:
@classmethod
def _initialize(cls, gitea, api_object, result):
for name, value in result.items():
if name in cls.fields_to_parsers and value is not None:
parse_func = cls.fields_to_parsers[name]
if name in cls._fields_to_parsers and value is not None:
parse_func = cls._fields_to_parsers[name]
value = parse_func(gitea, value)
cls._add_read_property(name, value, api_object)
# add all patchable fields missing in the request to be writable
for name in cls.fields_to_parsers.keys():
for name in cls._fields_to_parsers.keys():
if not hasattr(api_object,name):
cls._add_read_property(name, None, api_object)
@ -78,26 +78,35 @@ class ReadonlyApiObject:
class ApiObject(ReadonlyApiObject):
patchable_fields = set()
_patchable_fields = set()
def __init__(self, gitea):
super().__init__(gitea)
self.dirty_fields = set()
self._dirty_fields = set()
def commit(self):
raise NotImplemented()
_parsers_to_fields = {}
def get_dirty_fields(self):
return {name: getattr(self, name) for name in self.dirty_fields}
dirty_fields_values = {}
for field in self._dirty_fields:
value = getattr(self, field)
if field in self._parsers_to_fields:
dirty_fields_values[field] = self._parsers_to_fields[field](value)
else:
dirty_fields_values[field] = value
return dirty_fields_values
@classmethod
def _initialize(cls, gitea, api_object, result):
super()._initialize(gitea,api_object,result)
for name, value in result.items():
if name in cls.patchable_fields:
if name in cls._patchable_fields:
cls._add_write_property(name,value,api_object)
# add all patchable fields missing in the request to be writable
for name in cls.patchable_fields:
for name in cls._patchable_fields:
if not hasattr(api_object,name):
cls._add_write_property(name, None, api_object)
@ -112,5 +121,5 @@ class ApiObject(ReadonlyApiObject):
def __set_var(self, name, i):
if self.deleted:
raise ObjectIsInvalid()
self.dirty_fields.add(name)
self._dirty_fields.add(name)
setattr(self, "_" + name, i)

30
tests/test_api.py
Wyświetl plik

@ -1,7 +1,7 @@
import pytest
import uuid
from gitea import Gitea, User, Organization, Team, Repository, Issue
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
@ -172,6 +172,13 @@ def test_create_team(instance):
assert team.description == "descr"
assert team.organization == org
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"
def test_user_teams(instance):
org = Organization.request(instance, test_org)
team = org.get_team(test_team)
@ -206,6 +213,27 @@ def test_hashing(instance):
commit = repo.get_commits()[0]
assert len(set([org, team, user, repo, issue, branch, commit, milestone]))
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")
new_body = "some new description with some more of that char stuff :)"
issue.body = new_body
issue.commit()
number = issue.number
del issue
issue2 = Issue.request(instance, org.username, repo.name, number)
assert issue2.body == new_body
milestone = repo.create_milestone(ms_title, "this is only a teststone2")
issue2.milestone = milestone
issue2.commit()
del issue2
issues = repo.get_issues()
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)
user = instance.get_user_by_name(test_user)

11
tests/test_api_longtests.py
Wyświetl plik

@ -36,4 +36,13 @@ def test_list_repos(instance):
for i in range(1, 34):
instance.create_repo(org, test_repo + "_" + str(i), str(i))
repos = org.get_repositories()
assert len(repos) >= 33
assert len(repos) >= 33
def test_list_issue(instance):
org = Organization.request(instance, test_org)
repo = Repository.request(instance, org.username, test_repo)
for x in range(0,100):
Issue.create_issue(instance, repo, "TestIssue" + str(x), "We will be to many to be listed")
issues = repo.get_issues()
assert len(issues) > 98