00fccb72e5
Patching because I have other things to do but basically this diff at line 20 shows that even though self.secret was assigned toa string value, it was still being given the type of tuple with the first/only item being the string value I have no idea how this was happeneing but its fixed as of now * also im ignoring the cargo_bin flag in the environment for now as its just causing issues while debugging
127 lines
4.3 KiB
Python
127 lines
4.3 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.jwt = None # never gets assigned until /login is hit
|
|
self.basic_creds = self.__create_admin()
|
|
self.id = self.basic_creds['id']
|
|
self.secret = self.basic_creds['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'
|
|
|
|
'''
|
|
if type(auth) == str:
|
|
opts['id'] = self.id
|
|
if auth == 'basic':
|
|
opts['secret'] = self.secret
|
|
else:
|
|
opts['jwt'] = auth
|
|
return opts
|
|
else:
|
|
# if its not a string we don't add anything in
|
|
return opts
|
|
|
|
def request(self, method: str, path: str, auth: str, opts: dict, expectation: int):
|
|
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()
|
|
self.requests[r_id] = req
|
|
|
|
resp = req.make(expectation)
|
|
self.responses[r_id] = resp
|
|
|
|
return r_id
|
|
|
|
|
|
def run(worker: Worker):
|
|
VOICE_CHAN = 1
|
|
TEXT_CHAN = 2
|
|
# preliminary test
|
|
req_login = worker.request('post', '/login', 'basic',{}, 200)
|
|
jwt = worker.responses[req_login].json()
|
|
|
|
new_channel_name = time.time()
|
|
|
|
channel_tests = [
|
|
# init field provides args for Response object generation
|
|
# hop field gives us the ideal statuscode we want when we actually send the request
|
|
# NOTE: Grouping by status code
|
|
|
|
# sanity check
|
|
{'init': ['get', '/channels/list', {}], 'auth': None, 'hope': 400},
|
|
|
|
{'init': ['post', '/login', {}], 'auth': 'basic', 'hope': 200},
|
|
{'init': ['post', '/channels/list', {}], 'auth': jwt, 'hope': 200},
|
|
|
|
# somehow save this garbino
|
|
{'init': ['post', '/channels/create', {'name': str(new_channel_name), 'kind': 2, '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), 'kind': 2,}], '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:
|
|
method, path, opts = test['init']
|
|
auth = test['auth']
|
|
hope = test['hope']
|
|
|
|
worker.request(method, path, auth, opts, hope)
|
|
|
|
|
|
if __name__ == '__main__':
|
|
worker = Worker('http://localhost:8888', create_admin=True)
|
|
run(worker)
|