1. Alexander Shorin
  2. hydra

Source

hydra / apps / test.py

# -*- coding: utf-8 -*-
#
# Copyright (C) 2013 Alexander Shorin
# All rights reserved.
#
# This software is licensed as described in the file COPYING, which
# you should have received as part of this distribution.
#

import couchdb
import getopt
import os
import logging
import platform
import socket
import sys
from getpass import getpass
from couchdb import json
from couchdb.http import extract_credentials, HTTPError

SYSTEM = platform.system().lower()

_HELP = """Usage: %(name)s [OPTIONS] URL

Arguments:

  URL              CouchDB database URL in next form:
                   http[s]://[user[:password]@]host[:port]/dbname
                   Note that setting password in URL will make it visible in
                   shell history.

Options:

  -h, --help       Display a this help message and exit.
  -u, --user=      User name to access CouchDB. Could be also defined in URL.
                   Password will be requested.

""" % dict(name=os.path.basename(sys.argv[0]))

_NO_URL = """URL argument must be specified.
""" + _HELP

_USER_DUPLICATE = """Multiple users defined, couldn't decide which one to use:
%s or %s
"""

class ColoredFormatter(logging.Formatter):

    def __init__(self, fmt=None, datefmt=None):
        self.colors = {
            'CRITICAL': '\x1b[1;31m',
            'ERROR': '\x1b[1;31m',
            'INFO': '\x1b[1;32m',
            'WARN': '\x1b[1;33m',
            'WARNING': '\x1b[1;33m',
            'DEBUG': '\x1b[1;37m',
            }
        self.reset = '\033[0m'
        logging.Formatter.__init__(self, fmt, datefmt)

    def format(self, record):
        if SYSTEM == 'linux':
            record.reset = self.reset
        else:
            record.reset = ''
        record.funcName = '[%s]' % record.funcName
        levelname = record.levelname
        if SYSTEM == 'linux' and levelname in self.colors:
            color = self.colors[levelname]
            levelname = ''.join((color, levelname[0] * 2, self.reset))
            record.levelname = '[%s]' % levelname
        else:
            record.levelname = '[%s]' % (levelname[0] * 2)
        return logging.Formatter.format(self, record)


def get_logger(name, level=logging.DEBUG):
    fmt = '%(reset)s%(levelname)s  %(message)s'
    instance = logging.Logger('couchdb.audit.%s' % name, level)
    handler = logging.StreamHandler(sys.stdout)
    handler.setFormatter(ColoredFormatter(fmt))
    instance.addHandler(handler)
    instance.propagate = False
    return instance


def run(url, credentials, target='database'):
    if target == 'ddoc':
        pass
    db = couchdb.Database(url)
    if credentials:
        db.resource.credentials = credentials
    root = logging.Logger('hydra')
    handler = logging.StreamHandler(sys.stdout)
    root.addHandler(handler)
    log = get_logger('hydra.tests')
    log.setLevel(logging.WARNING)

    root.info('* Database: %s', db.resource.url)

    try:
        db.info()['db_name']
    except socket.error, err:
        log.error('%s: %s', err.__class__.__name__, err)
        return
    except couchdb.HTTPError, err:
        log.error('%s: %s', err.__class__.__name__, err.args[0][1])
        return
    except KeyError, err:
        log.critical('%s: %s; Possible not database?',
                     err.__class__.__name__, err)
        return

    try:
        rows = db.view('_all_docs', startkey='_design/',  endkey='_design0')
        rows = list(rows)
    except HTTPError, err:
        log.critical('Unable to get design documents list: %s',
                     err.args[0][1])

    for row in rows:
        root.info('* DDoc: %s', db.resource(*row.id.split('/')).url)
        name = row['id'].split('/')[1]
        try:
            data = json.decode(db.show('%s/tests' % name, format='json')[1].read())
        except couchdb.ResourceNotFound:
            log.warn('No tests provided by `show/tests`')
            continue
        except couchdb.ServerError:
            continue
        except socket.error:
            continue

        for suite, cases in data['results']:
            if log.level <= logging.INFO:
                root.info('\n * Suite: %s', suite)
            for case, tests in cases:
                if log.level <= logging.INFO:
                    root.info('\n   * Case: %s', suite)
                for test, info in tests:
                    if info['status'] in ['failed','error']:
                        log.error('%s failed\n%s', test, info['details'])
                    elif info['status'] in ['skipped', 'not run']:
                        log.warn('%s skipped: %s', test, info['details'])
                    else:
                        log.info('%s passed', test)

        for key, value in sorted(data['stats'].items()):
            if not value:
                continue
            root.info('  * %s : %s', key, value)
    return 0


def main():
    try:
        options, arguments = getopt.gnu_getopt(
            sys.argv[1:], 'hu:', ['help', 'user=']
        )
    except getopt.GetoptError, err:
        sys.stdout.write(('%s\n\n' % err).capitalize())
        sys.stdout.write(_HELP)
        sys.stdout.flush()
        sys.exit(1)
    message = None

    user = None
    for option, value in options:
        if option in ['-h', '--help']:
            message = _HELP
        elif option in ['-u', '--user'] and value:
            user = value

    if message:
        sys.stdout.write(message)
        sys.stdout.flush()
        sys.exit(0)

    if not arguments:
        sys.stdout.write(_NO_URL)
        sys.stdout.flush()
        sys.exit(1)

    url = arguments[0]
    if not url.startswith(('http://', 'https://')):
        url = 'http://' + url
    _, credentials = extract_credentials(url)

    if credentials:
        if user is not None and credentials[0] != user:
            sys.stdout.write(_USER_DUPLICATE % (credentials[0], user))
            sys.stdout.flush()
            sys.exit(1)
        credentials = list(credentials)
    elif user:
        credentials = [user]

    if credentials and len(credentials) == 1:
        credentials.append(getpass('Enter password for %s: ' % credentials[0]))

    target = 'database'
    if '_design' in url:
        target = 'ddoc'

    if credentials:
        credentials = tuple(credentials)

    sys.exit(run(url, credentials, target))

if __name__ == '__main__':
    main()