runner.py 文件源码

def run_test(config, args):
    """
    Run the 'test' subcommand

    :param config: configuration
    :type config: :py:class:`~.Config`
    :param args: command line arguments
    :type args: :py:class:`argparse.Namespace`
    """
    base_url = get_base_url(config, args)
    logger.debug('API base url: %s', base_url)
    endpoints = config.get('endpoints')
    if args.endpoint_name is not None:
        endpoints = {
            args.endpoint_name: endpoints[args.endpoint_name]
        }
    dt = datetime.now().strftime('%Y-%m-%dT%H:%M:%S.%f')
    data = {
        'message': 'testing via webhook2lambda2sqs CLI',
        'version': VERSION,
        'host': node(),
        'datetime': dt
    }
    for ep in sorted(endpoints):
        url = base_url + ep + '/'
        print('=> Testing endpoint %s with %s: %s' % (
            url, endpoints[ep]['method'], pformat(data))
        )
        if endpoints[ep]['method'] == 'POST':
            res = requests.post(url, json=data)
        elif endpoints[ep]['method'] == 'GET':
            res = requests.get(url, params=data)
        else:
            raise Exception('Unimplemented method: %s'
                            '' % endpoints[ep]['method'])
        print('RESULT: HTTP %d' % res.status_code)
        for h in sorted(res.headers):
            print('%s: %s' % (h, res.headers[h]))
        print("\n%s\n" % res.content)