79d77e0007
Body parameters were being polluted so new logic is required to make the worker more stable and ignore its state properly
255 lines
7.8 KiB
Python
255 lines
7.8 KiB
Python
import time
|
|
import subprocess, os, sys
|
|
import json, requests
|
|
|
|
class RequestError(Exception):
|
|
pass
|
|
|
|
class Response:
|
|
def __init__(self, body, code):
|
|
self.body = body
|
|
self.code = code
|
|
|
|
def __str__(self):
|
|
return f'{self.code} => {self.body}'
|
|
|
|
class Request:
|
|
def __init__(self, domain: str, params: dict):
|
|
assert(path[len(path) - 1] != '/')
|
|
|
|
self.domain = domain
|
|
self.params = params
|
|
|
|
def _make_request(self, method: str, path: str, hope: int):
|
|
# Lower driver for actuall making the request we are looking for
|
|
assert(path.startswith('/'))
|
|
method = method.lower()
|
|
|
|
url = self.domain + path
|
|
if method == 'get':
|
|
resp = requests.get(url, data=self.params)
|
|
return Response(resp.body, resp.status_code, hope)
|
|
elif method == 'post':
|
|
resp = requests.post(url, data=self.params)
|
|
return Response(resp.body, resp.status_code, hope)
|
|
elif: method == 'delete':
|
|
resp = requests.delete(url, data=self.params)
|
|
return Response(resp.body, resp.status_code, hope)
|
|
|
|
else:
|
|
raise RequestError('Invalid method passed')
|
|
|
|
|
|
|
|
def get(self, path: str, hope: int):
|
|
return self._make_request('get', path, hope)
|
|
|
|
def post(self, path: str, hope: int):
|
|
return self._make_request('post', path, hope)
|
|
|
|
def delete(self, path: str, hope: int):
|
|
return self._make_request('delete', path, hope)
|
|
|
|
|
|
|
|
class Test:
|
|
def __init__(self, base='http://localhost:8888', create_admin=False, admin=None, init_verbose=False):
|
|
'''
|
|
@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.test_count = 0
|
|
self.out = sys.stdout # changed before worker runtime
|
|
if create_admin:
|
|
self.body = Test.__create_admin()
|
|
self.id = self.body['id']
|
|
self.secret = self.body['secret']
|
|
|
|
if init_verbose:
|
|
print(f'TESTBOT Using => {self.body}')
|
|
elif admin is not None:
|
|
self.body = body
|
|
self.id = self.body['id']
|
|
self.secret = self.body['secret']
|
|
else:
|
|
# for now we use this because my dev server has this fake ass acc on it
|
|
self.secret = 'mH7DfAfSLHhVbo9OW4dmqTbzzLnFaZQLxoZFIlKvKHm4NfpLyZsZrYdf12aACZXyEQxELOlBTe2rc36vTPVd8A==',
|
|
self.id = 2
|
|
# Required params
|
|
self.body = {
|
|
'name': 'owner sama uwu',
|
|
'joindate': 69,
|
|
'status': 0,
|
|
'permissions': 18446744073709551615
|
|
}
|
|
self.default_body = {}
|
|
# deep copy of body to prevent side-channel writes
|
|
for k in self.body: self.default_body[k] = self.body[k]
|
|
|
|
self.base = base
|
|
|
|
def stdout_to_stderr(self):
|
|
self.out = sys.stderr
|
|
|
|
@staticmethod
|
|
def __create_admin():
|
|
# /home/$user/.cargo/bin/cargo <- normally
|
|
# prolly some awful shit on pipes
|
|
CARGO_BIN = os.getenv('CARGO_BIN')
|
|
|
|
proc = subprocess.run(f'{CARGO_BIN} 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 log(self, params: dict, url: str, method: str, request: requests.Response, body=True):
|
|
f = sys.stdout
|
|
print(f'TESTBOT {method} {url} {request.status_code}', file=self.out)
|
|
print(f'TESTBOT\t\t[Parameters]: {params}', file=self.out)
|
|
if body:
|
|
print(f'TESTBOT\t\t[Body]: {request.text}\n', file=self.out)
|
|
|
|
def _build_req_body(self, **options):
|
|
_map = self.default_body
|
|
for key in options:
|
|
_map[key] = options[key]
|
|
|
|
return json.dumps(_map)
|
|
|
|
|
|
|
|
def post(self, url, jwt=False, **opts):
|
|
'''
|
|
@returns text of response
|
|
'''
|
|
options = opts
|
|
if jwt is False:
|
|
options['id'] = self.id
|
|
if isinstance(self.secret, tuple):
|
|
options['secret'] = self.secret[0]
|
|
else:
|
|
options['secret'] = self.secret
|
|
else:
|
|
options['jwt'] = jwt
|
|
|
|
body = self._build_req_body(**options)
|
|
r = requests.post(self.base + url, data=body)
|
|
self.log(opts, self.base + url, 'POST', r)
|
|
|
|
return r.text
|
|
|
|
def get(self, url, jwt=False, **opts):
|
|
'''
|
|
@returns text of response
|
|
'''
|
|
options = opts
|
|
if jwt is False:
|
|
options['id'] = self.id
|
|
if isinstance(self.secret, tuple):
|
|
options['secret'] = self.secret[0]
|
|
else:
|
|
options['secret'] = self.secret
|
|
else:
|
|
options['jwt'] = jwt
|
|
|
|
body = self._build_req_body(**options)
|
|
r = requests.get(self.base + url, data=body)
|
|
self.log(opts, self.base + url, 'GET', r)
|
|
return r.text
|
|
|
|
def delete(self, url, jwt=False, **opts):
|
|
'''
|
|
@returns text of response
|
|
'''
|
|
options = opts
|
|
if jwt is False:
|
|
options['id'] = self.id
|
|
if isinstance(self.secret, tuple):
|
|
options['secret'] = self.secret[0]
|
|
else:
|
|
options['secret'] = self.secret
|
|
else:
|
|
options['jwt'] = jwt
|
|
|
|
body = self._build_req_body(**options)
|
|
r = requests.delete(self.base + url, data=body)
|
|
self.log(opts, self.base + url, 'DELETE', r)
|
|
return r.text
|
|
|
|
def creds(self):
|
|
return self.body
|
|
|
|
def spam_messages(channel, jwt, worker):
|
|
for _ in range(15):
|
|
worker.post('/message/send', jwt=jwt, channel=channel, content='dummy post')
|
|
|
|
def run(worker):
|
|
VOICE_CHAN = 1
|
|
TEXT_CHAN = 2
|
|
worker.stdout_to_stderr()
|
|
|
|
# the first two are sanity checks and should not faill
|
|
worker.get('/channels/list') # Should 400 or something since we're not sending the right keys
|
|
jwt_response = worker.post('/login', jwt=False)
|
|
jwt = json.loads(jwt_response)['jwt']
|
|
worker.get('/channels/list', jwt=jwt) # maybe now we'll 200?
|
|
|
|
# now for some things that may/may not fail or something idk
|
|
cname = time.time()
|
|
# pass 200
|
|
newchannel = worker.post('/channels/create', jwt=jwt, name=f'{cname}', kind=2, description='some description')
|
|
newchannel = json.loads(newchannel)
|
|
|
|
# fail 400 or something
|
|
worker.post('/channels/create', jwt=jwt, name=f'{cname}', kind=VOICE_CHAN, description='some description')
|
|
|
|
# ez pass
|
|
worker.get('/channels/list', jwt=jwt)
|
|
|
|
# pass 200
|
|
worker.post('/message/send', jwt=jwt, channel=newchannel['id'], content='some bullshit message')
|
|
|
|
# pass 200
|
|
worker.delete('/channels/delete', jwt=jwt, channel_id=newchannel['id'])
|
|
|
|
# fail 400
|
|
worker.post('/channels/create', jwt=False, name=f'succ', kind=2)
|
|
|
|
# pass 200
|
|
worker.get('/meta', jwt=jwt)
|
|
|
|
|
|
# Getting messages now
|
|
# pass 200
|
|
send_chan = worker.post('/channels/create', jwt=jwt, name=f'{cname}', kind=2, description='some description')
|
|
send_chan = json.loads(send_chan)
|
|
# pass 200(all)
|
|
spam_messages(send_chan['id'], jwt, worker)
|
|
now = time.time()
|
|
messages = json.loads(worker.get('/message/get_range', jwt=jwt, **{
|
|
'start-time': int(now - 15),
|
|
'end-time': int(now + 1),
|
|
'channel': send_chan['id']
|
|
}))
|
|
|
|
# 200 pass
|
|
from_id = worker.get('/message/from_id', jwt=jwt, **{
|
|
'channel': send_chan['id'],
|
|
'start': messages['messages'][0]['id']
|
|
})
|
|
print('done')
|
|
|
|
|
|
|
|
if __name__ == '__main__':
|
|
worker = Test(create_admin=False, init_verbose=True)
|
|
run(worker)
|