Removal/skeltonizing of a shitload of old code
This commit is contained in:
parent
fc74a3dbc7
commit
2f15e2ef62
@ -4,8 +4,8 @@ import json, requests
|
|||||||
from web import http
|
from web import http
|
||||||
|
|
||||||
|
|
||||||
class Test:
|
class Worker:
|
||||||
def __init__(self, base='http://localhost:8888', create_admin=False, admin=None, init_verbose=False):
|
def __init__(self, domain: str, create_admin: bool):
|
||||||
'''
|
'''
|
||||||
@opt:base = base url string
|
@opt:base = base url string
|
||||||
@opt:create_admin = creates admin account directly off cargo -c <NAME>
|
@opt:create_admin = creates admin account directly off cargo -c <NAME>
|
||||||
@ -13,43 +13,28 @@ class Test:
|
|||||||
potentially from another instance to run multiple tests at once
|
potentially from another instance to run multiple tests at once
|
||||||
'''
|
'''
|
||||||
|
|
||||||
self.test_count = 0
|
self.domain = domain
|
||||||
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:
|
if create_admin:
|
||||||
print(f'TESTBOT Using => {self.body}')
|
self.creds = self.__create_admin()
|
||||||
elif admin is not None:
|
# two most important details for basically everything
|
||||||
self.body = body
|
|
||||||
self.id = self.body['id']
|
self.id = self.body['id']
|
||||||
self.secret = self.body['secret']
|
self.secret = self.body['secret']
|
||||||
else:
|
else:
|
||||||
# for now we use this because my dev server has this fake ass acc on it
|
# for now we use this because my dev server has this fake ass acc on it
|
||||||
self.secret = 'mH7DfAfSLHhVbo9OW4dmqTbzzLnFaZQLxoZFIlKvKHm4NfpLyZsZrYdf12aACZXyEQxELOlBTe2rc36vTPVd8A==',
|
self.secret = 'mH7DfAfSLHhVbo9OW4dmqTbzzLnFaZQLxoZFIlKvKHm4NfpLyZsZrYdf12aACZXyEQxELOlBTe2rc36vTPVd8A==',
|
||||||
self.id = 2
|
self.id = 2
|
||||||
# Required params
|
# This is to never be overwritten
|
||||||
self.body = {
|
self.creds = {
|
||||||
'name': 'owner sama uwu',
|
'name': 'owner sama uwu',
|
||||||
'joindate': 69,
|
'joindate': 69,
|
||||||
'status': 0,
|
'status': 0,
|
||||||
'permissions': 18446744073709551615
|
'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 __create_admin(self):
|
||||||
|
|
||||||
def stdout_to_stderr(self):
|
|
||||||
self.out = sys.stderr
|
|
||||||
|
|
||||||
@staticmethod
|
|
||||||
def __create_admin():
|
|
||||||
# /home/$user/.cargo/bin/cargo <- normally
|
# /home/$user/.cargo/bin/cargo <- normally
|
||||||
# prolly some awful shit on pipes
|
# NOTE: the above path is prolly some awful shit on ci pipeline
|
||||||
CARGO_BIN = os.getenv('CARGO_BIN')
|
CARGO_BIN = os.getenv('CARGO_BIN')
|
||||||
|
|
||||||
proc = subprocess.run(f'{CARGO_BIN} run --release -- -c dev-test'.split(), text=True, capture_output=True)
|
proc = subprocess.run(f'{CARGO_BIN} run --release -- -c dev-test'.split(), text=True, capture_output=True)
|
||||||
@ -63,81 +48,24 @@ class Test:
|
|||||||
|
|
||||||
|
|
||||||
def log(self, params: dict, url: str, method: str, request: requests.Response, body=True):
|
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 {method} {url} {request.status_code}', file=self.out)
|
||||||
print(f'TESTBOT\t\t[Parameters]: {params}', file=self.out)
|
print(f'TESTBOT\t\t[Parameters]: {params}', file=self.out)
|
||||||
if body:
|
if body:
|
||||||
print(f'TESTBOT\t\t[Body]: {request.text}\n', file=self.out)
|
print(f'TESTBOT\t\t[Body]: {request.text}\n', file=self.out)
|
||||||
|
|
||||||
def _build_req_body(self, **options):
|
def _build_req_body(self, **options):
|
||||||
_map = self.default_body
|
pass
|
||||||
for key in options:
|
|
||||||
_map[key] = options[key]
|
|
||||||
|
|
||||||
return json.dumps(_map)
|
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
def post(self, url, jwt=False, **opts):
|
def post(self, url, jwt=False, **opts):
|
||||||
'''
|
pass
|
||||||
@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):
|
def get(self, url, jwt=False, **opts):
|
||||||
'''
|
pass
|
||||||
@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)
|
def delete(self):
|
||||||
r = requests.get(self.base + url, data=body)
|
pass
|
||||||
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):
|
def spam_messages(channel, jwt, worker):
|
||||||
for _ in range(15):
|
for _ in range(15):
|
||||||
@ -202,5 +130,5 @@ def run(worker):
|
|||||||
|
|
||||||
|
|
||||||
if __name__ == '__main__':
|
if __name__ == '__main__':
|
||||||
worker = Test(create_admin=False, init_verbose=True)
|
worker = Worker('http://localhost:8888', create_admin=False, init_verbose=True)
|
||||||
run(worker)
|
run(worker)
|
||||||
|
Loading…
Reference in New Issue
Block a user