c25e8047cf
For now this sub api is at an mvp stage so it should be fine to move onto new features after this point for now.
178 lines
6.3 KiB
Python
178 lines
6.3 KiB
Python
from time import time
|
|
from request import Request
|
|
from config import create_admin, Admin
|
|
from config import Server
|
|
from json import dumps as to_json
|
|
|
|
RESPONSES = []
|
|
VOICE_CHAN = 1
|
|
TEXT_CHAN = 2
|
|
|
|
def login(admin: Admin) -> (Request, str):
|
|
print(f'Provided admin account {admin}')
|
|
req = Request(
|
|
admin.server.url + '/login',
|
|
'post',
|
|
{'id':admin.id, 'secret': admin.secret},
|
|
{},
|
|
200
|
|
)
|
|
response = req.fire()
|
|
try:
|
|
jwt = response.json().get('jwt')
|
|
return (req, jwt)
|
|
except:
|
|
return (req, None)
|
|
|
|
def make_channel(url: str, id: int, jwt: str) -> (Request, int):
|
|
# making a text channel
|
|
channel_name = str(time())
|
|
qs = {'id': id, 'jwt': jwt, 'kind': 2, 'name': channel_name}
|
|
req = Request(url + '/channels/create', 'post', qs, {}, 200)
|
|
resp = req.fire()
|
|
try:
|
|
channel = resp.json().get('channel')
|
|
chan_id = 0 if channel is None else channel.get('id')
|
|
return (req, chan_id)
|
|
except Exception as e:
|
|
print(e)
|
|
print(f'Actual value: {resp.text}: {resp.status_code}')
|
|
return (req, 0)
|
|
|
|
def bs_admin(server: Server) -> Admin:
|
|
user_ = {
|
|
'id': 123,
|
|
'jwt': None,
|
|
'secret': 'no thanks'
|
|
}
|
|
server_ = {
|
|
'url': server.url,
|
|
'wsurl': server.wsurl,
|
|
'name': server.serv_name
|
|
}
|
|
return Admin(user_, server_)
|
|
|
|
|
|
def req(admin: Admin, method: str, path: str, qs: dict, hope: int, headers: dict={}, body=None, verbose=False) -> Request:
|
|
url = admin.server.url + path
|
|
# Copy over the additional params into the query string
|
|
params = {'id': admin.id, 'jwt': admin.jwt}
|
|
for key in qs:
|
|
params[key] = qs[key]
|
|
|
|
return Request(url, method, params, headers, hope, body, verbose)
|
|
|
|
def make_and_receive_invite(admin: Admin) -> (Request, Request):
|
|
make_inv_req = req(admin, 'post', '/invite/create', {}, 200, verbose=True)
|
|
|
|
# Fire the first request and do some safety checks in case the request fails
|
|
make_inv_req.fire()
|
|
if make_inv_req.response is None:
|
|
return (make_inv_req, user_inv_req)
|
|
elif make_inv_req.response.status_code >= 400:
|
|
print('Params used ', make_inv_req)
|
|
return (make_inv_req, None)
|
|
|
|
# Setup to fire the second request, .fire() is safe to blindly use at this point
|
|
new_invite_body = make_inv_req.response.json()['invite']
|
|
code = new_invite_body['id']
|
|
user_inv_req = req(admin , 'get' , '/join', {'code': code}, 200, verbose=True)
|
|
user_inv_req.fire()
|
|
|
|
return (make_inv_req, user_inv_req)
|
|
|
|
|
|
if __name__ == '__main__':
|
|
admin = create_admin()
|
|
if admin is None:
|
|
print('Unable to parse/create admin account')
|
|
exit(1)
|
|
fake_user = bs_admin(admin.server)
|
|
|
|
requests = []
|
|
|
|
# First a quick sanity check for login
|
|
# add this after we fire the generic tests
|
|
login_req, jwt = login(admin)
|
|
if jwt is None:
|
|
print('Unable to /login - stopping now to avoid pointless failure')
|
|
req.show_response()
|
|
exit(1)
|
|
|
|
admin.jwt = jwt
|
|
|
|
# add this after we fire the generic tests
|
|
mk_chan_sanity, chan_id = make_channel(admin.server.url, admin.id, admin.jwt)
|
|
mk_chan_to_delete, del_chan_id = make_channel(admin.server.url, admin.id, admin.jwt)
|
|
|
|
# Invite test
|
|
|
|
# Container for most/all the generic requests we want to fire off
|
|
now = time()
|
|
tmp_neighbor = {
|
|
'name': str(now),
|
|
'wsurl': 'wsurl',
|
|
'url': str(now),
|
|
'description': 'asdf',
|
|
'tags': ['red','blue']
|
|
}
|
|
# This neighbor is used to "update" the tmp one above in /neighbor tests
|
|
updated_neighbor = tmp_neighbor.copy()
|
|
updated_neighbor['name'] = 'new'
|
|
print('Updated neighbor as json ', to_json(updated_neighbor))
|
|
|
|
requests.extend([
|
|
req(fake_user, 'get', '/channels/list', {}, 401),
|
|
req(admin, 'post', '/channels/list', {'kind': TEXT_CHAN}, 404),
|
|
req(admin, 'get' , '/channels/list', {'kind': TEXT_CHAN}, 200),
|
|
req(admin , 'delete', '/channels/delete', {'channel_id':del_chan_id}, 200),
|
|
req(admin, 'post', '/message/send', {'channel_id': chan_id},200,{'content-type':'text/plain'}, 'asdf'),
|
|
req(admin, 'post', '/message/send', {'channel_id': 123}, 400, {'content-type': 'text/plain'}, 'asdf'),
|
|
req(admin , 'post', '/message/send', {'channel_id': chan_id}, 200, {'content-type': 'image/png'}, 'asdf'),
|
|
req(admin , 'post', '/message/send', {'channel_id': 123}, 400, {'content-type': 'image/png'}, 'asdf'),
|
|
req(admin, 'get', '/message/recent', {'channel_id': chan_id, 'limit': 20}, 200, verbose=True),
|
|
req(admin, 'get', '/message/recent', {'channel_id': 123, 'limit': 20}, 404),
|
|
req(admin, 'get', '/members/me', {}, 200),
|
|
req(admin, 'get', '/members/get_online', {}, 200),
|
|
req(admin, 'post', '/members/me/nickname', {'nick': f'randy-{now}'}, 200),
|
|
req(admin , 'get', '/invite/join', {'code': 123}, 404),
|
|
req(admin , "get", "/meta", {}, 200),
|
|
req(admin, 'get', '/neighbor/list', {}, 200),
|
|
req(admin,'post', '/neighbor/add', {}, 200, body=to_json(tmp_neighbor)),
|
|
req(admin, 'get', '/neighbor/list', {}, 200, verbose=True),
|
|
req(admin, 'put', '/neighbor/update', {'url':str(now)}, 200, body=to_json(updated_neighbor)),
|
|
req(admin, 'put', '/neighbor/update', {'url':'fake'}, 404, body=to_json(updated_neighbor)),
|
|
req(admin, 'get', '/neighbor/list', {}, 200),
|
|
req(admin, 'delete', '/neighbor/delete', {'url':'fake'}, 200),
|
|
req(admin, 'delete', '/neighbor/delete', {'url': str(now)}, 200),
|
|
req(admin, 'get', '/neighbor/list', {}, 200, verbose=True),
|
|
])
|
|
|
|
# add this after fire the generic tests
|
|
inv_req_mk, inv_req_use = make_and_receive_invite(admin)
|
|
|
|
# Fire off
|
|
for req in requests:
|
|
req.fire()
|
|
|
|
# Prepend the sanity checks we did early on
|
|
requests.insert(0, login_req)
|
|
requests.insert(0, mk_chan_sanity)
|
|
requests.insert(0, mk_chan_to_delete)
|
|
# Append the invite endpoint tests(only because they're meant to come near the end)
|
|
requests.append(inv_req_mk)
|
|
if inv_req_use is not None:
|
|
requests.append(inv_req_use)
|
|
|
|
pass_count = 0
|
|
for r in requests:
|
|
r.show_response()
|
|
if r.passing:
|
|
pass_count += 1
|
|
|
|
|
|
req_count = len(requests)
|
|
print(f'Passing {pass_count}/{req_count}')
|
|
|
|
|