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, '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}')