Get fakes and mox working so that any http request/response sequence can be

test scripted
This commit is contained in:
Steve Baker 2012-10-12 16:50:43 +13:00
parent 70619f8945
commit 3c9c859e81
7 changed files with 213 additions and 55 deletions

View File

@ -1,26 +1,31 @@
# vim: tabstop=4 shiftwidth=4 softtabstop=4
# 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
# 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
# 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.
# 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.
import inspect
import os
#NOTE(bcwaldon): this try/except block is needed to run setup.py due to
# its need to import local code before installing required dependencies
try:
import heatclient.client
Client = heatclient.client.Client
except ImportError:
import warnings
warnings.warn("Could not import heatclient.client", ImportWarning)
import heatclient.version
def _get_heatclient_version():
"""Read version from versioninfo file."""
mod_abspath = inspect.getabsfile(inspect.currentframe())
heatclient_path = os.path.dirname(mod_abspath)
version_path = os.path.join(heatclient_path, 'versioninfo')
__version__ = heatclient.version.version_info.deferred_version_string()
if os.path.exists(version_path):
version = open(version_path).read().strip()
else:
version = "Unknown, couldn't find versioninfo file at %s"\
% version_path
return version
__version__ = _get_heatclient_version()

View File

@ -15,16 +15,26 @@ Command-line interface to the OpenStack Images API.
"""
import argparse
import glob
import httplib2
import imp
import itertools
import os
import pkgutil
import sys
import logging
import re
import sys
from keystoneclient.v2_0 import client as ksclient
import heatclient
from heatclient import exc
from heatclient import client as heatclient
from heatclient.common import utils
logger = logging.getLogger(__name__)
class HeatShell(object):
@ -192,21 +202,6 @@ class HeatShell(object):
subparser.add_argument(*args, **kwargs)
subparser.set_defaults(func=callback)
# TODO(dtroyer): move this into the common client support?
# Compatibility check to remove API version as the trailing component
# in a service endpoint; also removes a trailing '/'
def _strip_version(self, endpoint):
"""Strip a version from the last component of an endpoint if present"""
# Get rid of trailing '/' if present
if endpoint.endswith('/'):
endpoint = endpoint[:-1]
url_bits = endpoint.split('/')
# regex to match 'v1' or 'v2.0' etc
if re.match('v\d+\.?\d*', url_bits[-1]):
endpoint = '/'.join(url_bits[:-1])
return endpoint
def _get_ksclient(self, **kwargs):
"""Get an endpoint and auth token from Keystone.
@ -225,15 +220,27 @@ class HeatShell(object):
def _get_endpoint(self, client, **kwargs):
"""Get an endpoint using the provided keystone client."""
endpoint = client.service_catalog.url_for(
return client.service_catalog.url_for(
service_type=kwargs.get('service_type') or 'orchestration',
endpoint_type=kwargs.get('endpoint_type') or 'publicURL')
return self._strip_version(endpoint)
def _setup_debugging(self, debug):
if not debug:
return
streamhandler = logging.StreamHandler()
streamformat = "%(levelname)s (%(module)s:%(lineno)d) %(message)s"
streamhandler.setFormatter(logging.Formatter(streamformat))
logger.setLevel(logging.DEBUG)
logger.addHandler(streamhandler)
httplib2.debuglevel = 1
def main(self, argv):
# Parse args once to find version
parser = self.get_base_parser()
(options, args) = parser.parse_known_args(argv)
self._setup_debugging(options.debug)
# build available subcommands based on version
api_version = options.heat_api_version
@ -254,10 +261,6 @@ class HeatShell(object):
self.do_help(args)
return 0
LOG = logging.getLogger('heatclient')
LOG.addHandler(logging.StreamHandler())
LOG.setLevel(logging.DEBUG if args.debug else logging.INFO)
heat_url = args.heat_url
auth_reqd = (utils.is_authentication_required(args.func) and
not (args.os_auth_token and heat_url))

View File

@ -122,11 +122,6 @@ class StackManager(base.Manager):
# else:
# return body
#
# def delete(self, image):
# """Delete an image."""
# self._delete("/v1/images/%s" % base.getid(image))
#
#
# def update(self, image, **kwargs):
# """Update an image
#

View File

@ -1 +1 @@
0.0.8.622719c
0.0.9.4ba8e46

66
tests/fakes.py Normal file
View File

@ -0,0 +1,66 @@
import httplib
from keystoneclient.v2_0 import client as ksclient
def script_keystone_client():
ksclient.Client(auth_url='http://no.where',
insecure=False,
password='password',
tenant_id='',
tenant_name='tenant_name',
username='username').AndReturn(
FakeKeystone('abcd1234'))
def script_get(url):
httplib.HTTPConnection.request('GET',
url,
headers=fake_headers())
def script_response(status, reason, headers, body):
httplib.HTTPConnection.getresponse().AndReturn(
FakeHTTPResponse(status, reason, headers, body))
def fake_headers():
return {'X-Auth-Token': 'abcd1234',
'Content-Type': 'application/json',
'Accept': 'application/json',
'User-Agent': 'python-heatclient'}
class FakeServiceCatalog():
def url_for(self, endpoint_type, service_type):
return 'http://192.168.1.5:8004/v1/f14b41234'
class FakeKeystone():
service_catalog = FakeServiceCatalog()
def __init__(self, auth_token):
self.auth_token = auth_token
class FakeHTTPResponse():
version = 1.1
def __init__(self, status, reason, headers, body):
self.headers = headers
self.body = body
self.status = status
self.reason = reason
def getheader(self, name, default=None):
return self.headers.get(name, default)
def getheaders(self):
return self.headers.items()
def read(self, amt=None):
b = self.body
self.body = None
return b

View File

@ -1,7 +0,0 @@
import unittest
class fooTest(unittest.TestCase):
def test_foo(self):
self.assertTrue(True)

96
tests/test_shell.py Normal file
View File

@ -0,0 +1,96 @@
import cStringIO
import os
import httplib2
import httplib
import sys
import mox
import unittest
from keystoneclient.v2_0 import client as ksclient
from heatclient import exc
import heatclient.shell
import fakes
class ShellTest(unittest.TestCase):
# Patch os.environ to avoid required auth info.
def setUp(self):
self.m = mox.Mox()
self.m.StubOutWithMock(ksclient, 'Client')
self.m.StubOutWithMock(httplib.HTTPConnection, 'request')
self.m.StubOutWithMock(httplib.HTTPConnection, 'getresponse')
global _old_env
fake_env = {
'OS_USERNAME': 'username',
'OS_PASSWORD': 'password',
'OS_TENANT_NAME': 'tenant_name',
'OS_AUTH_URL': 'http://no.where',
}
_old_env, os.environ = os.environ, fake_env.copy()
def tearDown(self):
self.m.UnsetStubs()
global _old_env
os.environ = _old_env
def shell(self, argstr):
orig = sys.stdout
try:
sys.stdout = cStringIO.StringIO()
_shell = heatclient.shell.HeatShell()
_shell.main(argstr.split())
except SystemExit:
exc_type, exc_value, exc_traceback = sys.exc_info()
self.assertEqual(exc_value.code, 0)
finally:
out = sys.stdout.getvalue()
sys.stdout.close()
sys.stdout = orig
return out
def test_help_unknown_command(self):
self.assertRaises(exc.CommandError, self.shell, 'help foofoo')
def test_debug(self):
httplib2.debuglevel = 0
self.shell('--debug help')
self.assertEqual(httplib2.debuglevel, 1)
def test_help(self):
required = [
'^usage: heat',
'(?m)^See "heat help COMMAND" for help on a specific command',
]
for argstr in ['--help', 'help']:
help_text = self.shell(argstr)
for r in required:
self.assertRegexpMatches(help_text, r)
def test_help_on_subcommand(self):
required = [
'^usage: heat list',
"(?m)^List the user's stacks",
]
argstrings = [
'help list',
]
for argstr in argstrings:
help_text = self.shell(argstr)
for r in required:
self.assertRegexpMatches(help_text, r)
def test_list(self):
fakes.script_keystone_client()
fakes.script_get('/v1/f14b41234/stacks?limit=20')
fakes.script_response(200,
'success, yo',
{'content-type': 'application/json'},
'{"stacks": {}}')
self.m.ReplayAll()
list_text = self.shell('list')