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

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
        if create_admin:
            self.basic_creds = self.__create_admin()
            # two most important details for basically everything
            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

            # This is to never be overwritten
            self.basic_creds = {
                'name': 'owner sama uwu',
                'joindate': 69,
                'status': 0,
                'permissions': 18446744073709551615
            }

    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_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 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 = Request(method, url, opts)
        r_id = time.time()
        self.requests[r_id] = req

        resp = req.make(expectation)
        self.responses[r_id] = resp

        return req.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},

        {'init': ['post'], 'auth': jwt, 'hope': 200}
    ]

    for test in channel_tests:
        method = test['init'][0]
        path = test['init'][1]
        opts = test['init'][2]

        worker.request(method, path, test['auth'], opts, test['hope'])


#    # 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 = Worker('http://localhost:8888', create_admin=False)
    run(worker)