Consolidate Persimmons shared backend and tests
This commit is contained in:
1 parent
4019bb5b01
commit
cbe019f354
1 file changed
+69
@@ -0,0 +1,69 @@
|
||||
"""Contract fixture for the dedicated Gitea API, never a fallback auth store."""
|
||||
from copy import deepcopy
|
||||
from persimmons import gitea
|
||||
|
||||
|
||||
class GiteaFixture:
|
||||
def __init__(self):
|
||||
self.users = {}
|
||||
self.repos = {}
|
||||
self.calls = []
|
||||
self.variables = {}
|
||||
self.secrets = {}
|
||||
|
||||
def __call__(self, path, method='GET', body=None, login=None):
|
||||
self.calls.append((path, method, deepcopy(body)))
|
||||
if path == '/admin/users' and method == 'POST':
|
||||
name = body['username']
|
||||
if name in self.users:
|
||||
raise gitea.Failure(422)
|
||||
user = {'id': len(self.users) + 2, 'login': name, 'is_admin': False, 'active': True}
|
||||
self.users[name] = (user, body['password'])
|
||||
return deepcopy(user)
|
||||
if path == '/user':
|
||||
if not login or login[0] not in self.users or login[1] != self.users[login[0]][1]:
|
||||
raise gitea.Failure(401)
|
||||
return deepcopy(self.users[login[0]][0])
|
||||
if path.startswith('/users/'):
|
||||
if path[7:] not in self.users:
|
||||
raise gitea.Failure(404)
|
||||
return deepcopy(self.users[path[7:]][0])
|
||||
if path.startswith('/admin/users/') and path.endswith('/repos'):
|
||||
name = path.split('/')[3]
|
||||
key = (name, body['name'].lower())
|
||||
if key in self.repos:
|
||||
raise gitea.Failure(409)
|
||||
repo = {'id': len(self.repos) + 1, 'name': body['name'], 'owner': self.users[name][0],
|
||||
'private': body['private'], 'empty': not body['auto_init']}
|
||||
self.repos[key] = deepcopy(repo)
|
||||
return deepcopy(repo)
|
||||
if '/actions/' in path:
|
||||
base, action = path.split('/actions/')
|
||||
self(base)
|
||||
kind, _, name = action.partition('/')
|
||||
kind = kind.split('?')[0]
|
||||
store = self.variables if kind == 'variables' else self.secrets
|
||||
key = (base, name)
|
||||
if method == 'GET':
|
||||
if kind == 'secrets':
|
||||
raise AssertionError('Secret read forbidden')
|
||||
if not name:
|
||||
return [{'name': n, 'data': v} for (b, n), v in store.items() if b == base]
|
||||
if key not in store:
|
||||
raise gitea.Failure(404)
|
||||
return {'name': name, 'data': store[key]}
|
||||
if method == 'POST' and key in store:
|
||||
raise gitea.Failure(409)
|
||||
if method == 'DELETE':
|
||||
if key not in store:
|
||||
raise gitea.Failure(404)
|
||||
del store[key]
|
||||
else:
|
||||
store[key] = body['value' if kind == 'variables' else 'data']
|
||||
return None
|
||||
if path.startswith('/repos/'):
|
||||
_, _, owner, name = path.split('/')
|
||||
if (owner, name.lower()) not in self.repos:
|
||||
raise gitea.Failure(404)
|
||||
return deepcopy(self.repos[owner, name.lower()])
|
||||
raise AssertionError('Unexpected Gitea request: ' + path)
|
||||
Reference in new issue
Block a user