7c3537e4f6
✨ Tests now use the new flags required for /channels/create ❗ Doubled size for channel descriptions
268 lines
9.2 KiB
Python
268 lines
9.2 KiB
Python
import time
|
|
import subprocess, os, sys
|
|
import json, requests
|
|
from web import http
|
|
|
|
|
|
class Worker:
|
|
def __init__(self, domain: str, create_admin: bool):
|
|
'''
|
|
@opt:base = base url string
|
|
@opt:create_admin = creates admin account directly off cargo -c <NAME>
|
|
@opt:admin = optionally pass in dictionary with some admin credentials
|
|
potentially from another instance to run multiple tests at once
|
|
'''
|
|
|
|
self.domain = domain
|
|
self.requests = {}
|
|
self.responses = {}
|
|
self.body_logs = {}
|
|
|
|
self.jwt = None # never gets assigned until /login is hit
|
|
self.basic_creds = self.__create_admin()
|
|
self.id = self.basic_creds['user']['id']
|
|
self.secret = self.basic_creds['user']['secret']
|
|
|
|
def __create_admin(self):
|
|
# /home/$user/.cargo/bin/cargo <- normally
|
|
# NOTE: the above path is prolly some awful shit on ci pipeline
|
|
CARGO_BIN = os.getenv('CARGO_BIN')
|
|
|
|
proc = subprocess.run(f'cargo run --release -- -c dev-test'.split(), text=True, capture_output=True)
|
|
try:
|
|
return json.loads(proc.stdout)
|
|
except:
|
|
import sys
|
|
print('TESTBOT UNABLE TO LOAD JSON DATA FROM -c flag', file=sys.stderr)
|
|
exit(1)
|
|
|
|
def auth(self, auth_opt: str):
|
|
if auth_opt == 'basic':
|
|
return self.basic_creds
|
|
else:
|
|
return {'id': self.id, 'jwt': auth_opt}
|
|
|
|
|
|
def _append_auth(self, opts: dict, auth: str):
|
|
'''
|
|
Default auth fallback type is jwt because that's they only other type of auth
|
|
FC cares about really
|
|
|
|
@param opts: Dictionary of parameters to pass to the endpoint
|
|
@auth: Denotes if we use basic auth or jwt if its not 'basic'
|
|
|
|
'''
|
|
# id is basically always required by the api so we mindlessly add it here
|
|
opts['id'] = self.id
|
|
|
|
if auth == 'basic':
|
|
opts['secret'] = self.secret
|
|
else:
|
|
opts['jwt'] = auth
|
|
return opts
|
|
|
|
def run_test(self, test):
|
|
# {
|
|
# 'init': [method, uri, query_string_dict],
|
|
# 'auth':jwt|none,
|
|
# 'hope': <status-code>
|
|
# 'body': <bool> # show response body or not (optional
|
|
# }
|
|
|
|
# not 'try'ing these as misconfigurations are to be cleaned up before testing
|
|
method, path, opts = test['init']
|
|
auth = test['auth']
|
|
hope = test['hope']
|
|
|
|
if 'body' in test:
|
|
self.request(method, path, auth, opts, hope, show_body=test['body'])
|
|
else:
|
|
self.request(method, path, auth, opts, hope)
|
|
|
|
def logs(self):
|
|
ids = sorted(self.requests.keys()) # shared keys in requests/responses
|
|
for key in ids:
|
|
self.responses[key].log()
|
|
|
|
# if body is request to be shown then dump it tabbed out by 1 tab
|
|
if self.body_logs[key] is True:
|
|
print(f'\tBody: {self.responses[key].body}')
|
|
|
|
def test_stats(self):
|
|
passc = 0
|
|
failc = 0
|
|
total = len(self.responses)
|
|
|
|
for key in self.responses:
|
|
hope = self.responses[key].expected
|
|
real = self.responses[key].code
|
|
if hope == real:
|
|
passc += 1
|
|
else:
|
|
failc += 1
|
|
|
|
print('=======')
|
|
print(f'\033[1;32mPassing\033[0m {passc}/{total}')
|
|
print(f'\033[1;31mFailing\033[0m {failc}/{total}')
|
|
print('=======')
|
|
|
|
|
|
|
|
def request(self, method: str, path: str, auth: str, opts: dict, expectation: int, show_body=False):
|
|
assert(path[0] == '/')
|
|
|
|
# First make sure we add in the correct auth params that are requested
|
|
opts = self._append_auth(opts, auth)
|
|
|
|
# Build the request and store it in our structure
|
|
url = self.domain + path
|
|
req = http.Request(method, url, opts)
|
|
r_id = time.time()
|
|
|
|
resp = req.make(expectation)
|
|
|
|
# update log trackers
|
|
self.requests[r_id] = req
|
|
self.responses[r_id] = resp
|
|
self.body_logs[r_id] = show_body
|
|
|
|
return r_id
|
|
|
|
|
|
def run(worker: Worker):
|
|
VOICE_CHAN = 1
|
|
TEXT_CHAN = 2
|
|
|
|
# Basically every test requires a jwt to be passed in so we grab that here
|
|
# Should this fail so should nearly every other test from this point
|
|
req_login = worker.request('post', '/login', 'basic',{}, 200)
|
|
jwt = worker.responses[req_login].json()['jwt']
|
|
|
|
new_channel_name = time.time()
|
|
|
|
channel_tests = [
|
|
# sanity check
|
|
{'init': ['get', '/channels/list', {}], 'auth': None, 'hope': 401},
|
|
{'init': ['post', '/channels/list', {}], 'auth': jwt, 'hope': 404},
|
|
|
|
{'init': ['post', '/channels/create', {'name': str(new_channel_name), 'type': TEXT_CHAN, 'description': 'asdf'}], 'auth': jwt, 'hope': 200},
|
|
# Just a regular test no saving for this one
|
|
{'init': ['post', '/channels/create', {'name': str(new_channel_name+1), 'type': TEXT_CHAN,}], 'auth': jwt, 'hope': 200},
|
|
|
|
{'init': ['post', '/channels/create', {}], 'auth': jwt, 'hope': 400},
|
|
{'init': ['post', '/channels/create', {'name': 123, 'kind': 'adsf'}], 'auth': jwt, 'hope': 400},
|
|
|
|
# save this and compare its results to the previous
|
|
{'init': ['get', '/channels/list', {}], 'auth': jwt, 'hope': 200},
|
|
{'init': ['get', '/channels/list', {'random-param': 123}], 'auth': jwt, 'hope': 200},
|
|
]
|
|
|
|
for test in channel_tests:
|
|
worker.run_test(test)
|
|
|
|
msg_chan_name = time.time()
|
|
_id = worker.request('post', '/channels/create', jwt, {
|
|
'name': str(msg_chan_name),
|
|
'type': TEXT_CHAN,
|
|
'description': 'asdf'
|
|
}, 200)
|
|
|
|
chan_d = worker.responses[_id].json()
|
|
|
|
message_tests = [
|
|
# bs message spam
|
|
{'init': ['post', '/message/send', {'channel_id': chan_d['id'], 'content': 'bs content'}], 'auth': jwt, 'hope': 200, 'body': True},
|
|
{'init': ['post', '/message/send', {'channel_id': chan_d['id'], 'content': 'bs content'}], 'auth': jwt, 'hope': 200},
|
|
{'init': ['post', '/message/send', {'channel_id': chan_d['id'], 'content': 'bs content'}], 'auth': jwt, 'hope': 200},
|
|
{'init': ['post', '/message/send', {'channel_id': chan_d['id'], 'content': 'bs content'}], 'auth': jwt, 'hope': 200},
|
|
|
|
# can we get them back tho?
|
|
{
|
|
'init': [
|
|
'get', '/message/get_range', {'channel_id': chan_d['id'], 'start_time': int(msg_chan_name-10), 'end_time': int(msg_chan_name + 10)}
|
|
],
|
|
'auth': jwt, 'hope': 200
|
|
},
|
|
{
|
|
'init': [
|
|
'get', '/message/get_range', {'channel_id': chan_d['id'], 'end_time': int(msg_chan_name)}
|
|
],
|
|
'auth': jwt, 'hope': 400
|
|
},
|
|
{
|
|
'init': [
|
|
'get', '/message/get_range', {'channel_id': chan_d['id'], 'start_time': int(msg_chan_name), 'end_time': int(msg_chan_name)}
|
|
],
|
|
'auth': jwt, 'hope': 400
|
|
},
|
|
{
|
|
'init': [
|
|
'get', '/message/get_range', {'channel_id': chan_d['id'], 'end_time': int(msg_chan_name), 'start_time': int(msg_chan_name)}
|
|
],
|
|
'auth': jwt, 'hope': 400
|
|
},
|
|
# two tests that follow the rules
|
|
{
|
|
'init': [
|
|
'get', '/message/from_id', {'start': 1, 'channel_id': 3}
|
|
],
|
|
'auth': jwt, 'hope': 200, 'body': True
|
|
},
|
|
{
|
|
'init': [
|
|
'get', '/message/from_id', {'start':1, 'channel_id':3, 'limit':2}
|
|
],
|
|
'auth': jwt, 'hope': 200, 'body': True
|
|
},
|
|
# tests that don't follow the api's rules
|
|
{
|
|
# channel doesn't exist so a 404 seems to be inorder
|
|
'init': [
|
|
'get', '/message/from_id', {'start': 1, 'channel_id':9}
|
|
],
|
|
'auth': jwt, 'hope': 404
|
|
},
|
|
{
|
|
'init': [
|
|
# good channel but id is tooo high
|
|
'get', '/message/from_id', {'start': 5, 'channel_id':3}
|
|
],
|
|
'auth': jwt, 'hope': 404
|
|
},
|
|
]
|
|
|
|
for test in message_tests:
|
|
worker.run_test(test)
|
|
|
|
|
|
|
|
member_tests = [
|
|
{ 'init': ['get', '/members/me', {}], 'auth': jwt, 'hope': 200, 'body': True},
|
|
{ 'init': ['get', '/members/get_online', {}], 'auth': jwt, 'hope': 200, 'body': True},
|
|
{ 'init': ['post', '/members/me/nickname', {'nick': f'New name: {time.ctime()}'}], 'auth': jwt, 'hope': 200 },
|
|
{ 'init': ['get', '/members/me', {}], 'auth': jwt, 'hope': 200, 'body': True},
|
|
]
|
|
|
|
for test in member_tests:
|
|
worker.run_test(test)
|
|
|
|
invite_tests = [
|
|
{'init': ['post', '/invite/create', {}], 'auth': jwt, 'hope': 200, 'body': True},
|
|
{'init': ['get', '/invite/create', {}], 'auth': jwt, 'hope': 404, 'body': True},
|
|
]
|
|
for test in invite_tests:
|
|
worker.run_test(test)
|
|
|
|
# ad-hoc test for joining now
|
|
invite_req_id = worker.request('post', '/invite/create', jwt, {}, 200, True)
|
|
code = worker.responses[invite_req_id].json()['id']
|
|
worker.request('get', '/join', None, {'code': code}, 200, True)
|
|
|
|
|
|
worker.logs()
|
|
worker.test_stats()
|
|
|
|
if __name__ == '__main__':
|
|
worker = Worker('http://localhost:4536', create_admin=True)
|
|
run(worker)
|