freechat/server-api/client-tests/client.py

import time
import subprocess, os, sys
import json, requests

class Response:
    def __init__(self, body, code):
        self.body = body
        self.code = code

    def __str__(self):
        return f'{self.code} => {self.body}'

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
            self.body = {
                'name': 'owner sama uwu',
                'joindate': 69,
                'status': 0,
                'permissions': 18446744073709551615
            }

        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.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 auth_tests(worker):
    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=1, description='some description')
    newchannel = json.loads(newchannel)

    # fail 400 or something
    worker.post('/channels/create', jwt=jwt, name=f'{cname}', kind=1, description='some description')

    # ez pass 
    worker.get('/channels/list', jwt=jwt)
    # pass 200
    worker.delete('/channels/delete', jwt=jwt, channel_id=newchannel['id'])

# these might be fucked now
def base_working_tests(worker):
    # First the invites api gets some basic tests

    # Channels things
    VOICE_CHANNEL = 1
    TEXT_CHANNEL = 2

    new_channel_response = worker.post('/channels/create', name=f'{time.time()}', kind=TEXT_CHANNEL)
    new_channel = json.loads(new_channel_response)

    channel_list = json.loads(worker.get('/channels/list'))
    worker.delete('/channels/delete', channel_id=new_channel['id'])
    worker.get('/channels/list')

    # Messaging
    msg_chan_id = time.time()
    msg_chan_raw = worker.post('/channels/create', name=f'{msg_chan_id}', kind=TEXT_CHANNEL)
    msg_chan = json.loads(msg_chan_raw)

    worker.post('/message/send', channel=msg_chan['id'], content="some random content")
    worker.delete('/channels/delete', channel_id=msg_chan['id']) # finally clean up the channel we created


if __name__ == '__main__':
    worker = Test(create_admin=False, init_verbose=True)
    #base_working_tests(worker)
    auth_tests(worker)