¡@

Home 

OpenStack Study: utils.py

OpenStack Index

**** CubicPower OpenStack Study ****

# Copyright 2010-2011 OpenStack Foundation

# All Rights Reserved.

#

# Licensed under the Apache License, Version 2.0 (the "License"); you may

# not use this file except in compliance with the License. You may obtain

# a copy of the License at

#

# http://www.apache.org/licenses/LICENSE-2.0

#

# Unless required by applicable law or agreed to in writing, software

# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT

# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the

# License for the specific language governing permissions and limitations

# under the License.

"""Common utilities used in testing"""

import errno

import functools

import os

import shlex

import shutil

import socket

import subprocess

import fixtures

from oslo.config import cfg

import six

import stubout

import testtools

import webob

from glance.common import config

from glance.common import exception

from glance.common import property_utils

from glance.common import wsgi

from glance import context

from glance.db.sqlalchemy import api as db_api

from glance.db.sqlalchemy import models as db_models

from glance.openstack.common import jsonutils

from glance.openstack.common import timeutils

CONF = cfg.CONF

**** CubicPower OpenStack Study ****

class BaseTestCase(testtools.TestCase):

**** CubicPower OpenStack Study ****

    def setUp(self):

        super(BaseTestCase, self).setUp()

        #NOTE(bcwaldon): parse_args has to be called to register certain

        # command-line options - specifically we need config_dir for

        # the following policy tests

        config.parse_args(args=[])

        self.addCleanup(CONF.reset)

        self.stubs = stubout.StubOutForTesting()

        self.stubs.Set(exception, '_FATAL_EXCEPTION_FORMAT_ERRORS', True)

        self.test_dir = self.useFixture(fixtures.TempDir()).path

**** CubicPower OpenStack Study ****

    def tearDown(self):

        self.stubs.UnsetAll()

        self.stubs.SmartUnsetAll()

        super(BaseTestCase, self).tearDown()

**** CubicPower OpenStack Study ****

    def set_property_protections(self, use_policies=False):

        self.unset_property_protections()

        conf_file = "property-protections.conf"

        if use_policies:

            conf_file = "property-protections-policies.conf"

            self.config(property_protection_rule_format="policies")

        self.property_file = self._copy_data_file(conf_file, self.test_dir)

        self.config(property_protection_file=self.property_file)

**** CubicPower OpenStack Study ****

    def unset_property_protections(self):

        for section in property_utils.CONFIG.sections():

            property_utils.CONFIG.remove_section(section)

**** CubicPower OpenStack Study ****

    def _copy_data_file(self, file_name, dst_dir):

        src_file_name = os.path.join('glance/tests/etc', file_name)

        shutil.copy(src_file_name, dst_dir)

        dst_file_name = os.path.join(dst_dir, file_name)

        return dst_file_name

**** CubicPower OpenStack Study ****

    def set_property_protection_rules(self, rules):

        f = open(self.property_file, 'w')

        for rule_key in rules.keys():

            f.write('[%s]\n' % rule_key)

            for operation in rules[rule_key].keys():

                roles_str = ','.join(rules[rule_key][operation])

                f.write('%s = %s\n' % (operation, roles_str))

        f.close()

**** CubicPower OpenStack Study ****

    def config(self, **kw):

        """

        Override some configuration values.

        The keyword arguments are the names of configuration options to

        override and their values.

        If a group argument is supplied, the overrides are applied to

        the specified configuration option group.

        All overrides are automatically cleared at the end of the current

        test by the fixtures cleanup process.

        """

        group = kw.pop('group', None)

        for k, v in kw.iteritems():

            CONF.set_override(k, v, group)

**** CubicPower OpenStack Study ****

class requires(object):

"""Decorator that initiates additional test setup/teardown."""

**** CubicPower OpenStack Study ****

    def __init__(self, setup=None, teardown=None):

        self.setup = setup

        self.teardown = teardown

**** CubicPower OpenStack Study ****

    def __call__(self, func):

        def _runner(*args, **kw):

            if self.setup:

                self.setup(args[0])

            func(*args, **kw)

            if self.teardown:

                self.teardown(args[0])

        _runner.__name__ = func.__name__

        _runner.__doc__ = func.__doc__

        return _runner

**** CubicPower OpenStack Study ****

        def _runner(*args, **kw):

            if self.setup:

                self.setup(args[0])

            func(*args, **kw)

            if self.teardown:

                self.teardown(args[0])

        _runner.__name__ = func.__name__

        _runner.__doc__ = func.__doc__

        return _runner

**** CubicPower OpenStack Study ****

class depends_on_exe(object):

"""Decorator to skip test if an executable is unavailable"""

**** CubicPower OpenStack Study ****

    def __init__(self, exe):

        self.exe = exe

**** CubicPower OpenStack Study ****

    def __call__(self, func):

        def _runner(*args, **kw):

            cmd = 'which %s' % self.exe

            exitcode, out, err = execute(cmd, raise_error=False)

            if exitcode != 0:

                args[0].disabled_message = 'test requires exe: %s' % self.exe

                args[0].disabled = True

            func(*args, **kw)

        _runner.__name__ = func.__name__

        _runner.__doc__ = func.__doc__

        return _runner

def skip_if_disabled(func):

    """Decorator that skips a test if test case is disabled."""

    @functools.wraps(func)

**** CubicPower OpenStack Study ****

        def _runner(*args, **kw):

            cmd = 'which %s' % self.exe

            exitcode, out, err = execute(cmd, raise_error=False)

            if exitcode != 0:

                args[0].disabled_message = 'test requires exe: %s' % self.exe

                args[0].disabled = True

            func(*args, **kw)

        _runner.__name__ = func.__name__

        _runner.__doc__ = func.__doc__

        return _runner

def skip_if_disabled(func):

    """Decorator that skips a test if test case is disabled."""

    @functools.wraps(func)

**** CubicPower OpenStack Study ****

    def wrapped(*a, **kwargs):

        func.__test__ = False

        test_obj = a[0]

        message = getattr(test_obj, 'disabled_message',

                          'Test disabled')

        if getattr(test_obj, 'disabled', False):

            test_obj.skipTest(message)

        func(*a, **kwargs)

    return wrapped

def fork_exec(cmd,

              exec_env=None,

              logfile=None):

    """

    Execute a command using fork/exec.

    This is needed for programs system executions that need path

    searching but cannot have a shell as their parent process, for

    example: glance-api.  When glance-api starts it sets itself as

    the parent process for its own process group.  Thus the pid that

    a Popen process would have is not the right pid to use for killing

    the process group.  This patch gives the test env direct access

    to the actual pid.

    :param cmd: Command to execute as an array of arguments.

    :param exec_env: A dictionary representing the environment with

                     which to run the command.

    :param logile: A path to a file which will hold the stdout/err of

                   the child process.

    """

    env = os.environ.copy()

    if exec_env is not None:

        for env_name, env_val in exec_env.items():

            if callable(env_val):

                env[env_name] = env_val(env.get(env_name))

            else:

                env[env_name] = env_val

    pid = os.fork()

    if pid == 0:

        if logfile:

            fds = [1, 2]

            with open(logfile, 'r+b') as fptr:

                for desc in fds:  # close fds

                    try:

                        os.dup2(fptr.fileno(), desc)

                    except OSError:

                        pass

        args = shlex.split(cmd)

        os.execvpe(args[0], args, env)

    else:

        return pid

def wait_for_fork(pid,

                  raise_error=True,

                  expected_exitcode=0):

    """

    Wait for a process to complete

    This function will wait for the given pid to complete.  If the

    exit code does not match that of the expected_exitcode an error

    is raised.

    """

    rc = 0

    try:

        (pid, rc) = os.waitpid(pid, 0)

        rc = os.WEXITSTATUS(rc)

        if rc != expected_exitcode:

            raise RuntimeError('The exit code %d is not %d'

                               % (rc, expected_exitcode))

    except Exception:

        if raise_error:

            raise

    return rc

def execute(cmd,

            raise_error=True,

            no_venv=False,

            exec_env=None,

            expect_exit=True,

            expected_exitcode=0,

            context=None):

    """

    Executes a command in a subprocess. Returns a tuple

    of (exitcode, out, err), where out is the string output

    from stdout and err is the string output from stderr when

    executing the command.

    :param cmd: Command string to execute

    :param raise_error: If returncode is not 0 (success), then

                        raise a RuntimeError? Default: True)

    :param no_venv: Disable the virtual environment

    :param exec_env: Optional dictionary of additional environment

                     variables; values may be callables, which will

                     be passed the current value of the named

                     environment variable

    :param expect_exit: Optional flag true iff timely exit is expected

    :param expected_exitcode: expected exitcode from the launcher

    :param context: additional context for error message

    """

    env = os.environ.copy()

    if exec_env is not None:

        for env_name, env_val in exec_env.items():

            if callable(env_val):

                env[env_name] = env_val(env.get(env_name))

            else:

                env[env_name] = env_val

    # If we're asked to omit the virtualenv, and if one is set up,

    # restore the various environment variables

    if no_venv and 'VIRTUAL_ENV' in env:

        # Clip off the first element of PATH

        env['PATH'] = env['PATH'].split(os.pathsep, 1)[-1]

        del env['VIRTUAL_ENV']

    # Make sure that we use the programs in the

    # current source directory's bin/ directory.

    path_ext = [os.path.join(os.getcwd(), 'bin')]

    # Also jack in the path cmd comes from, if it's absolute

    executable = cmd.split()[0]

    if os.path.isabs(executable):

        path_ext.append(os.path.dirname(executable))

    env['PATH'] = ':'.join(path_ext) + ':' + env['PATH']

    process = subprocess.Popen(cmd,

                               shell=True,

                               stdin=subprocess.PIPE,

                               stdout=subprocess.PIPE,

                               stderr=subprocess.PIPE,

                               env=env)

    if expect_exit:

        result = process.communicate()

        (out, err) = result

        exitcode = process.returncode

    else:

        out = ''

        err = ''

        exitcode = 0

    if exitcode != expected_exitcode and raise_error:

        msg = "Command %(cmd)s did not succeed. Returned an exit "\

              "code of %(exitcode)d."\

              "\n\nSTDOUT: %(out)s"\

              "\n\nSTDERR: %(err)s" % {'cmd': cmd, 'exitcode': exitcode,

                                       'out': out, 'err': err}

        if context:

            msg += "\n\nCONTEXT: %s" % context

        raise RuntimeError(msg)

    return exitcode, out, err

def find_executable(cmdname):

    """

    Searches the path for a given cmdname.  Returns an absolute

    filename if an executable with the given name exists in the path,

    or None if one does not.

    :param cmdname: The bare name of the executable to search for

    """

    # Keep an eye out for the possibility of an absolute pathname

    if os.path.isabs(cmdname):

        return cmdname

    # Get a list of the directories to search

    path = ([os.path.join(os.getcwd(), 'bin')] +

            os.environ['PATH'].split(os.pathsep))

    # Search through each in turn

    for elem in path:

        full_path = os.path.join(elem, cmdname)

        if os.access(full_path, os.X_OK):

            return full_path

    # No dice...

    return None

def get_unused_port():

    """

    Returns an unused port on localhost.

    """

    port, s = get_unused_port_and_socket()

    s.close()

    return port

def get_unused_port_and_socket():

    """

    Returns an unused port on localhost and the open socket

    from which it was created.

    """

    s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)

    s.bind(('localhost', 0))

    addr, port = s.getsockname()

    return (port, s)

def xattr_writes_supported(path):

    """

    Returns True if the we can write a file to the supplied

    path and subsequently write a xattr to that file.

    """

    try:

        import xattr

    except ImportError:

        return False

**** CubicPower OpenStack Study ****

def fork_exec(cmd,

              exec_env=None,

              logfile=None):

    """

    Execute a command using fork/exec.

    This is needed for programs system executions that need path

    searching but cannot have a shell as their parent process, for

    example: glance-api.  When glance-api starts it sets itself as

    the parent process for its own process group.  Thus the pid that

    a Popen process would have is not the right pid to use for killing

    the process group.  This patch gives the test env direct access

    to the actual pid.

    :param cmd: Command to execute as an array of arguments.

    :param exec_env: A dictionary representing the environment with

                     which to run the command.

    :param logile: A path to a file which will hold the stdout/err of

                   the child process.

    """

    env = os.environ.copy()

    if exec_env is not None:

        for env_name, env_val in exec_env.items():

            if callable(env_val):

                env[env_name] = env_val(env.get(env_name))

            else:

                env[env_name] = env_val

    pid = os.fork()

    if pid == 0:

        if logfile:

            fds = [1, 2]

            with open(logfile, 'r+b') as fptr:

                for desc in fds:  # close fds

                    try:

                        os.dup2(fptr.fileno(), desc)

                    except OSError:

                        pass

        args = shlex.split(cmd)

        os.execvpe(args[0], args, env)

    else:

        return pid

**** CubicPower OpenStack Study ****

def wait_for_fork(pid,

                  raise_error=True,

                  expected_exitcode=0):

    """

    Wait for a process to complete

    This function will wait for the given pid to complete.  If the

    exit code does not match that of the expected_exitcode an error

    is raised.

    """

    rc = 0

    try:

        (pid, rc) = os.waitpid(pid, 0)

        rc = os.WEXITSTATUS(rc)

        if rc != expected_exitcode:

            raise RuntimeError('The exit code %d is not %d'

                               % (rc, expected_exitcode))

    except Exception:

        if raise_error:

            raise

    return rc

**** CubicPower OpenStack Study ****

def execute(cmd,

            raise_error=True,

            no_venv=False,

            exec_env=None,

            expect_exit=True,

            expected_exitcode=0,

            context=None):

    """

    Executes a command in a subprocess. Returns a tuple

    of (exitcode, out, err), where out is the string output

    from stdout and err is the string output from stderr when

    executing the command.

    :param cmd: Command string to execute

    :param raise_error: If returncode is not 0 (success), then

                        raise a RuntimeError? Default: True)

    :param no_venv: Disable the virtual environment

    :param exec_env: Optional dictionary of additional environment

                     variables; values may be callables, which will

                     be passed the current value of the named

                     environment variable

    :param expect_exit: Optional flag true iff timely exit is expected

    :param expected_exitcode: expected exitcode from the launcher

    :param context: additional context for error message

    """

    env = os.environ.copy()

    if exec_env is not None:

        for env_name, env_val in exec_env.items():

            if callable(env_val):

                env[env_name] = env_val(env.get(env_name))

            else:

                env[env_name] = env_val

    # If we're asked to omit the virtualenv, and if one is set up,

    # restore the various environment variables

    if no_venv and 'VIRTUAL_ENV' in env:

        # Clip off the first element of PATH

        env['PATH'] = env['PATH'].split(os.pathsep, 1)[-1]

        del env['VIRTUAL_ENV']

    # Make sure that we use the programs in the

    # current source directory's bin/ directory.

    path_ext = [os.path.join(os.getcwd(), 'bin')]

    # Also jack in the path cmd comes from, if it's absolute

    executable = cmd.split()[0]

    if os.path.isabs(executable):

        path_ext.append(os.path.dirname(executable))

    env['PATH'] = ':'.join(path_ext) + ':' + env['PATH']

    process = subprocess.Popen(cmd,

                               shell=True,

                               stdin=subprocess.PIPE,

                               stdout=subprocess.PIPE,

                               stderr=subprocess.PIPE,

                               env=env)

    if expect_exit:

        result = process.communicate()

        (out, err) = result

        exitcode = process.returncode

    else:

        out = ''

        err = ''

        exitcode = 0

    if exitcode != expected_exitcode and raise_error:

        msg = "Command %(cmd)s did not succeed. Returned an exit "\

              "code of %(exitcode)d."\

              "\n\nSTDOUT: %(out)s"\

              "\n\nSTDERR: %(err)s" % {'cmd': cmd, 'exitcode': exitcode,

                                       'out': out, 'err': err}

        if context:

            msg += "\n\nCONTEXT: %s" % context

        raise RuntimeError(msg)

    return exitcode, out, err

**** CubicPower OpenStack Study ****

def find_executable(cmdname):

    """

    Searches the path for a given cmdname.  Returns an absolute

    filename if an executable with the given name exists in the path,

    or None if one does not.

    :param cmdname: The bare name of the executable to search for

    """

    # Keep an eye out for the possibility of an absolute pathname

    if os.path.isabs(cmdname):

        return cmdname

    # Get a list of the directories to search

    path = ([os.path.join(os.getcwd(), 'bin')] +

            os.environ['PATH'].split(os.pathsep))

    # Search through each in turn

    for elem in path:

        full_path = os.path.join(elem, cmdname)

        if os.access(full_path, os.X_OK):

            return full_path

    # No dice...

    return None

**** CubicPower OpenStack Study ****

def get_unused_port():

    """

    Returns an unused port on localhost.

    """

    port, s = get_unused_port_and_socket()

    s.close()

    return port

**** CubicPower OpenStack Study ****

def get_unused_port_and_socket():

    """

    Returns an unused port on localhost and the open socket

    from which it was created.

    """

    s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)

    s.bind(('localhost', 0))

    addr, port = s.getsockname()

    return (port, s)

**** CubicPower OpenStack Study ****

def xattr_writes_supported(path):

    """

    Returns True if the we can write a file to the supplied

    path and subsequently write a xattr to that file.

    """

    try:

        import xattr

    except ImportError:

        return False

**** CubicPower OpenStack Study ****

    def set_xattr(path, key, value):

        xattr.setxattr(path, "user.%s" % key, str(value))

    # We do a quick attempt to write a user xattr to a temporary file

    # to check that the filesystem is even enabled to support xattrs

    fake_filepath = os.path.join(path, 'testing-checkme')

    result = True

    with open(fake_filepath, 'wb') as fake_file:

        fake_file.write("XXX")

        fake_file.flush()

    try:

        set_xattr(fake_filepath, 'hits', '1')

    except IOError as e:

        if e.errno == errno.EOPNOTSUPP:

            result = False

    else:

        # Cleanup after ourselves...

        if os.path.exists(fake_filepath):

            os.unlink(fake_filepath)

    return result

def minimal_headers(name, public=True):

    headers = {

        'Content-Type': 'application/octet-stream',

        'X-Image-Meta-Name': name,

        'X-Image-Meta-disk_format': 'raw',

        'X-Image-Meta-container_format': 'ovf',

    }

    if public:

        headers['X-Image-Meta-Is-Public'] = 'True'

    return headers

def minimal_add_command(port, name, suffix='', public=True):

    visibility = 'is_public=True' if public else ''

    return ("bin/glance --port=%d add %s"

            " disk_format=raw container_format=ovf"

            " name=%s %s" % (port, visibility, name, suffix))

**** CubicPower OpenStack Study ****

def minimal_headers(name, public=True):

    headers = {

        'Content-Type': 'application/octet-stream',

        'X-Image-Meta-Name': name,

        'X-Image-Meta-disk_format': 'raw',

        'X-Image-Meta-container_format': 'ovf',

    }

    if public:

        headers['X-Image-Meta-Is-Public'] = 'True'

    return headers

**** CubicPower OpenStack Study ****

def minimal_add_command(port, name, suffix='', public=True):

    visibility = 'is_public=True' if public else ''

    return ("bin/glance --port=%d add %s"

            " disk_format=raw container_format=ovf"

            " name=%s %s" % (port, visibility, name, suffix))

**** CubicPower OpenStack Study ****

class RegistryAPIMixIn(object):

**** CubicPower OpenStack Study ****

    def create_fixtures(self):

        for fixture in self.FIXTURES:

            db_api.image_create(self.context, fixture)

            with open(os.path.join(self.test_dir, fixture['id']),

                      'wb') as image:

                image.write("chunk00000remainder")

**** CubicPower OpenStack Study ****

    def destroy_fixtures(self):

        db_models.unregister_models(db_api.get_engine())

        db_models.register_models(db_api.get_engine())

**** CubicPower OpenStack Study ****

    def get_fixture(self, **kwargs):

        fixture = {'name': 'fake public image',

                   'status': 'active',

                   'disk_format': 'vhd',

                   'container_format': 'ovf',

                   'is_public': True,

                   'size': 20,

                   'checksum': None}

        fixture.update(kwargs)

        return fixture

**** CubicPower OpenStack Study ****

    def get_minimal_fixture(self, **kwargs):

        fixture = {'name': 'fake public image',

                   'is_public': True,

                   'disk_format': 'vhd',

                   'container_format': 'ovf'}

        fixture.update(kwargs)

        return fixture

**** CubicPower OpenStack Study ****

    def get_extra_fixture(self, id, name, **kwargs):

        created_at = kwargs.pop('created_at', timeutils.utcnow())

        updated_at = kwargs.pop('updated_at', created_at)

        return self.get_fixture(

            id=id, name=name, deleted=False, deleted_at=None,

            created_at=created_at, updated_at=updated_at,

            **kwargs)

**** CubicPower OpenStack Study ****

    def get_api_response_ext(self, http_resp, url='/images', headers={},

                             body=None, method=None, api=None,

                             content_type=None):

        if api is None:

            api = self.api

        req = webob.Request.blank(url)

        for k, v in headers.iteritems():

            req.headers[k] = v

        if method:

            req.method = method

        if body:

            req.body = body

        if content_type == 'json':

            req.content_type = 'application/json'

        elif content_type == 'octet':

            req.content_type = 'application/octet-stream'

        res = req.get_response(api)

        self.assertEqual(res.status_int, http_resp)

        return res

**** CubicPower OpenStack Study ****

    def assertEqualImages(self, res, uuids, key='images', unjsonify=True):

        images = jsonutils.loads(res.body)[key] if unjsonify else res

        self.assertEqual(len(images), len(uuids))

        for i, value in enumerate(uuids):

            self.assertEqual(images[i]['id'], value)

**** CubicPower OpenStack Study ****

class FakeAuthMiddleware(wsgi.Middleware):

**** CubicPower OpenStack Study ****

    def __init__(self, app, is_admin=False):

        super(FakeAuthMiddleware, self).__init__(app)

        self.is_admin = is_admin

**** CubicPower OpenStack Study ****

    def process_request(self, req):

        auth_tok = req.headers.get('X-Auth-Token')

        user = None

        tenant = None

        roles = []

        if auth_tok:

            user, tenant, role = auth_tok.split(':')

            if tenant.lower() == 'none':

                tenant = None

            roles = [role]

            req.headers['X-User-Id'] = user

            req.headers['X-Tenant-Id'] = tenant

            req.headers['X-Roles'] = role

            req.headers['X-Identity-Status'] = 'Confirmed'

        kwargs = {

            'user': user,

            'tenant': tenant,

            'roles': roles,

            'is_admin': self.is_admin,

            'auth_tok': auth_tok,

        }

        req.context = context.RequestContext(**kwargs)

**** CubicPower OpenStack Study ****

class FakeHTTPResponse(object):

**** CubicPower OpenStack Study ****

    def __init__(self, status=200, headers=None, data=None, *args, **kwargs):

        data = data or 'I am a teapot, short and stout\n'

        self.data = six.StringIO(data)

        self.read = self.data.read

        self.status = status

        self.headers = headers or {'content-length': len(data)}

**** CubicPower OpenStack Study ****

    def getheader(self, name, default=None):

        return self.headers.get(name.lower(), default)

**** CubicPower OpenStack Study ****

    def getheaders(self):

        return self.headers or {}

**** CubicPower OpenStack Study ****

    def read(self, amt):

        self.data.read(amt)

**** CubicPower OpenStack Study ****

class Httplib2WsgiAdapter(object):

**** CubicPower OpenStack Study ****

    def __init__(self, app):

        self.app = app

**** CubicPower OpenStack Study ****

    def request(self, uri, method="GET", body=None, headers=None):

        req = webob.Request.blank(uri, method=method, headers=headers)

        req.body = body

        resp = req.get_response(self.app)

        return Httplib2WebobResponse(resp), resp.body

**** CubicPower OpenStack Study ****

class Httplib2WebobResponse(object):

**** CubicPower OpenStack Study ****

    def __init__(self, webob_resp):

        self.webob_resp = webob_resp

    @property

**** CubicPower OpenStack Study ****

    def status(self):

        return self.webob_resp.status_code

**** CubicPower OpenStack Study ****

    def __getitem__(self, key):

        return self.webob_resp.headers[key]

**** CubicPower OpenStack Study ****

    def get(self, key):

        return self.webob_resp.headers[key]

    @property

**** CubicPower OpenStack Study ****

    def allow(self):

        return self.webob_resp.allow

    @allow.setter

**** CubicPower OpenStack Study ****

    def allow(self, allowed):

        if type(allowed) is not str:

            raise TypeError('Allow header should be a str')

        self.webob_resp.allow = allowed

**** CubicPower OpenStack Study ****

class HttplibWsgiAdapter(object):

**** CubicPower OpenStack Study ****

    def __init__(self, app):

        self.app = app

        self.req = None

**** CubicPower OpenStack Study ****

    def request(self, method, url, body=None, headers={}):

        self.req = webob.Request.blank(url, method=method, headers=headers)

        self.req.body = body

**** CubicPower OpenStack Study ****

    def getresponse(self):

        response = self.req.get_response(self.app)

        return FakeHTTPResponse(response.status_code, response.headers,

                                response.body)