added ability to request paginated gitea api endpoints; fixed listing paginated organisation repository listing

pull/11/head
Langenfeld 2021-10-27 17:47:37 +02:00
rodzic 79276e94d1
commit bd2371b1b6
2 zmienionych plików z 32 dodań i 3 usunięć

Wyświetl plik

@ -1,6 +1,5 @@
import json
import logging
import uuid
from datetime import datetime
from typing import List, Tuple, Dict, Sequence, Optional
@ -50,7 +49,7 @@ class Organization(GiteaApiObject):
self.dirty_fields = {}
def get_repositories(self) -> List["Repository"]:
results = self.gitea.requests_get(
results = self.gitea.requests_get_paginated(
Organization.ORG_REPOS_REQUEST % self.username
)
return [Repository.parse_response(self.gitea, result) for result in results]
@ -403,7 +402,7 @@ class Repository(GiteaApiObject):
def delete(self):
self.gitea.requests_delete(
Repository.REPO_DELETE % (self.owner.username, self.name)
Repository.REPO_DELETE % (self.owner.username, self.name)
)
self.deleted = True
@ -658,6 +657,19 @@ class Gitea:
return json.loads(result.text)
return {}
def requests_get_paginated(self, endpoint, params ={}, requests=None, sudo=None, page_key: str = "page"):
page = 1
combined_params = {}
combined_params.update(params)
aggregated_result = []
while True:
combined_params[page_key] = page
result = self.requests_get(endpoint, combined_params, requests, sudo)
aggregated_result.extend(result)
page += 1
if len(result) == 0:
return aggregated_result
def requests_get(self, endpoint, params={}, requests=None, sudo=None):
combined_params = {}
combined_params.update(params)

Wyświetl plik

@ -125,6 +125,16 @@ def test_create_repo_orgowned(instance):
assert repo.name == test_repo
assert not repo.private
def test_list_repos(instance):
org = Organization.request(instance, test_org)
repos = org.get_repositories()
assert len(repos) > 0
# test a number of repository listings larger than the pagination number (default 30)
for i in range(1, 34):
instance.create_repo(org, test_repo + "_" + str(i), str(i))
repos = org.get_repositories()
assert len(repos) >= 33
def test_list_branches(instance):
org = Organization.request(instance, test_org)
repo = org.get_repository(test_repo)
@ -204,6 +214,13 @@ 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()
for repo in repos:
repo.delete()
repos = org.get_repositories()
assert len(repos) == 0
def test_delete_org(instance):
org = Organization.request(instance, test_org)