openstack/glance
Code Issues Proposed changes

tests.unit.test_misc.execute -> tests.utils.execute after merge

Browse Source
This commit is contained in:
jaypipes@gmail.com 2011-03-24 19:03:17 -04:00
commit 85f91b771c
16 changed files with 949 additions and 370 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
MANIFEST.in
View File

@ -1,9 +1,8 @@
include LICENSE run_tests.sh ChangeLog
include run_tests.sh
include README builddeb.sh
include MANIFEST.in pylintrc
include tests/__init__.py
include tests/stubs.py
include tests/test_data.py
include tests/utils.py
include run_tests.py
include glance/registry/db/migrate_repo/migrate.cfg

4
README
View File

@ -29,10 +29,10 @@ documentation for more details).
Now that Glance is installed, you can start the service. The easiest way to
do that is by using the `glance-combined` utility which runs both the
do that is by using the `glance-control` utility which runs both the
`glance-api` and `glance-registry` services::
glance-combined
glance-control all start
Once both services are running, you can now use the `glance-upload` tool to

2
bin/glance
View File

@ -270,7 +270,7 @@ to spell field names correctly. :)"""
# Have to handle "boolean" values specially...
if 'is_public' in fields:
image_meta['is_public'] = utils.int_from_bool_as_string(
image_meta['is_public'] = utils.bool_from_string(
fields.pop('is_public'))
# Add custom attributes, which are all the arguments remaining

15
glance/common/wsgi.py
View File

@ -35,6 +35,17 @@ import webob.dec
import webob.exc
class WritableLogger(object):
"""A thin wrapper that responds to `write` and logs."""
def __init__(self, logger, level=logging.DEBUG):
self.logger = logger
self.level = level
def write(self, msg):
self.logger.log(self.level, msg.strip("\n"))
def run_server(application, port):
"""Run a WSGI server with the given application."""
sock = eventlet.listen(('0.0.0.0', port))
@ -61,7 +72,9 @@ class Server(object):
def _run(self, application, socket):
"""Start a WSGI server in a new green thread."""
eventlet.wsgi.server(socket, application, custom_pool=self.pool)
logger = logging.getLogger('eventlet.wsgi.server')
eventlet.wsgi.server(socket, application, custom_pool=self.pool,
log=WritableLogger(logger))
class Middleware(object):

4
glance/server.py
View File

@ -163,7 +163,9 @@ class Controller(wsgi.Controller):
yield chunk
res = Response(app_iter=image_iterator(),
content_type="text/plain")
content_type="application/octet-stream")
# Using app_iter blanks content-length, so we set it here...
res.headers.add('Content-Length', image['size'])
utils.inject_image_meta_into_headers(res, image)
return req.get_response(res)

2
glance/store/http.py
View File

@ -24,7 +24,7 @@ class HTTPBackend(glance.store.Backend):
""" An implementation of the HTTP Backend Adapter """
@classmethod
def get(cls, parsed_uri, expected_size, conn_class=None):
def get(cls, parsed_uri, expected_size, options=None, conn_class=None):
"""Takes a parsed uri for an HTTP resource, fetches it, and yields the
data.
"""

1
setup.py
View File

@ -87,7 +87,6 @@ setup(
],
scripts=['bin/glance',
'bin/glance-api',
'bin/glance-combined',
'bin/glance-control',
'bin/glance-manage',
'bin/glance-registry',

225
tests/functional/__init__.py Normal file
View File

@ -0,0 +1,225 @@
# vim: tabstop=4 shiftwidth=4 softtabstop=4
# Copyright 2011 OpenStack, LLC
# 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.
"""
Base test class for running non-stubbed tests (functional tests)
The FunctionalTest class contains helper methods for starting the API
and Registry server, grabbing the logs of each, cleaning up pidfiles,
and spinning down the servers.
"""
import datetime
import os
import random
import shutil
import signal
import socket
import tempfile
import time
import unittest
from tests.utils import execute, get_unused_port
class FunctionalTest(unittest.TestCase):
"""
Base test class for any test that wants to test the actual
servers and clients and not just the stubbed out interfaces
"""
def setUp(self):
self.test_id = random.randint(0, 100000)
self.test_dir = os.path.join("/", "tmp", "test.%d" % self.test_id)
self.api_port = get_unused_port()
self.api_pid_file = os.path.join(self.test_dir,
"glance-api.pid")
self.api_log_file = os.path.join(self.test_dir, "apilog")
self.registry_port = get_unused_port()
self.registry_pid_file = ("/tmp/test.%d/glance-registry.pid"
% self.test_id)
self.registry_log_file = os.path.join(self.test_dir, "registrylog")
self.image_dir = "/tmp/test.%d/images" % self.test_id
self.sql_connection = os.environ.get('GLANCE_SQL_CONNECTION',
"sqlite://")
self.pid_files = [self.api_pid_file,
self.registry_pid_file]
self.files_to_destroy = []
def tearDown(self):
self.cleanup()
def cleanup(self):
"""
Makes sure anything we created or started up in the
tests are destroyed or spun down
"""
for pid_file in self.pid_files:
if os.path.exists(pid_file):
pid = int(open(pid_file).read().strip())
try:
os.killpg(pid, signal.SIGTERM)
except:
pass # Ignore if the process group is dead
os.unlink(pid_file)
for f in self.files_to_destroy:
if os.path.exists(f):
os.unlink(f)
def start_servers(self, **kwargs):
"""
Starts the API and Registry servers (bin/glance-api and
bin/glance-registry) on unused ports and returns a tuple
of the (api_port, registry_port, conf_file_name).
Any kwargs passed to this method will override the configuration
value in the conf file used in starting the servers.
"""
self.cleanup()
conf_override = self.__dict__.copy()
if kwargs:
conf_override.update(**kwargs)
# A config file to use just for this test...we don't want
# to trample on currently-running Glance servers, now do we?
conf_file = tempfile.NamedTemporaryFile()
conf_contents = """[DEFAULT]
verbose = True
debug = True
[app:glance-api]
paste.app_factory = glance.server:app_factory
filesystem_store_datadir=%(image_dir)s
default_store = file
bind_host = 0.0.0.0
bind_port = %(api_port)s
registry_host = 0.0.0.0
registry_port = %(registry_port)s
log_file = %(api_log_file)s
[app:glance-registry]
paste.app_factory = glance.registry.server:app_factory
bind_host = 0.0.0.0
bind_port = %(registry_port)s
log_file = %(registry_log_file)s
sql_connection = %(sql_connection)s
sql_idle_timeout = 3600
""" % conf_override
conf_file.write(conf_contents)
conf_file.flush()
self.conf_file_name = conf_file.name
# Start up the API and default registry server
cmd = ("./bin/glance-control api start "
"%(conf_file_name)s --pid-file=%(api_pid_file)s"
% self.__dict__)
exitcode, out, err = execute(cmd)
self.assertEqual(0, exitcode,
"Failed to spin up the API server. "
"Got: %s" % err)
self.assertTrue("Starting glance-api with" in out)
cmd = ("./bin/glance-control registry start "
"%(conf_file_name)s --pid-file=%(registry_pid_file)s"
% self.__dict__)
exitcode, out, err = execute(cmd)
self.assertEqual(0, exitcode,
"Failed to spin up the Registry server. "
"Got: %s" % err)
self.assertTrue("Starting glance-registry with" in out)
self.wait_for_servers()
return self.api_port, self.registry_port, self.conf_file_name
def ping_server(self, port):
"""
Simple ping on the port. If responsive, return True, else
return False.
:note We use raw sockets, not ping here, since ping uses ICMP and
has no concept of ports...
"""
s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
try:
s.connect(("127.0.0.1", port))
s.close()
return True
except socket.error, e:
return False
def wait_for_servers(self, timeout=3):
"""
Tight loop, waiting for both API and registry server to be
available on the ports. Returns when both are pingable. There
is a timeout on waiting for the servers to come up.
:param timeout: Optional, defaults to 3
"""
now = datetime.datetime.now()
timeout_time = now + datetime.timedelta(seconds=timeout)
while (timeout_time > now):
if self.ping_server(self.api_port) and\
self.ping_server(self.registry_port):
return
now = datetime.datetime.now()
time.sleep(0.05)
self.assertFalse(True, "Failed to start servers.")
def stop_servers(self):
"""
Called to stop the started servers in a normal fashion. Note
that cleanup() will stop the servers using a fairly draconian
method of sending a SIGTERM signal to the servers. Here, we use
the glance-control stop method to gracefully shut the server down.
This method also asserts that the shutdown was clean, and so it
is meant to be called during a normal test case sequence.
"""
# Spin down the API and default registry server
cmd = ("./bin/glance-control api start "
"%(conf_file_name)s --pid-file=%(api_pid_file)s"
% self.__dict__)
exitcode, out, err = execute(cmd)
self.assertEqual(0, exitcode,
"Failed to spin down the API server. "
"Got: %s" % err)
cmd = ("./bin/glance-control registry start "
"%(conf_file_name)s --pid-file=%(registry_pid_file)s"
% self.__dict__)
exitcode, out, err = execute(cmd)
self.assertEqual(0, exitcode,
"Failed to spin down the Registry server. "
"Got: %s" % err)
# If all went well, then just remove the test directory.
# We only want to check the logs and stuff if something
# went wrong...
if os.path.exists(self.test_dir):
shutil.rmtree(self.test_dir)

152
tests/functional/test_bin_glance.py Normal file
View File

@ -0,0 +1,152 @@
# vim: tabstop=4 shiftwidth=4 softtabstop=4
# Copyright 2011 OpenStack, LLC
# 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.
"""Functional test case that utilizes the bin/glance CLI tool"""
import os
import unittest
from tests import functional
from tests.utils import execute
class TestBinGlance(functional.FunctionalTest):
"""Functional tests for the bin/glance CLI tool"""
def test_add_list_delete_list(self):
"""
We test the following:
0. Verify no public images in index
1. Add a public image with a location attr
and no image data
2. Check that image exists in index
3. Delete the image
4. Verify no longer in index
"""
self.cleanup()
api_port, reg_port, conf_file = self.start_servers()
# 0. Verify no public images
cmd = "bin/glance --port=%d index" % api_port
exitcode, out, err = execute(cmd)
self.assertEqual(0, exitcode)
self.assertEqual('No public images found.', out.strip())
# 1. Add public image
cmd = "bin/glance --port=%d add is_public=True name=MyImage" % api_port
exitcode, out, err = execute(cmd)
self.assertEqual(0, exitcode)
self.assertEqual('Added new image with ID: 1', out.strip())
# 2. Verify image added as public image
cmd = "bin/glance --port=%d index" % api_port
exitcode, out, err = execute(cmd)
self.assertEqual(0, exitcode)
lines = out.split("\n")
first_line = lines[0]
image_data_line = lines[3]
self.assertEqual('Found 1 public images...', first_line)
self.assertTrue('MyImage' in image_data_line)
# 3. Delete the image
cmd = "bin/glance --port=%d delete 1" % api_port
exitcode, out, err = execute(cmd)
self.assertEqual(0, exitcode)
self.assertEqual('Deleted image 1', out.strip())
# 4. Verify no public images
cmd = "bin/glance --port=%d index" % api_port
exitcode, out, err = execute(cmd)
self.assertEqual(0, exitcode)
self.assertEqual('No public images found.', out.strip())
self.stop_servers()
def test_add_list_update_list(self):
"""
Test for LP Bug #736295
We test the following:
0. Verify no public images in index
1. Add a NON-public image
2. Check that image does not appear in index
3. Update the image to be public
4. Check that image now appears in index
"""
self.cleanup()
api_port, reg_port, conf_file = self.start_servers()
# 0. Verify no public images
cmd = "bin/glance --port=%d index" % api_port
exitcode, out, err = execute(cmd)
self.assertEqual(0, exitcode)
self.assertEqual('No public images found.', out.strip())
# 1. Add public image
cmd = "bin/glance --port=%d add name=MyImage" % api_port
exitcode, out, err = execute(cmd)
self.assertEqual(0, exitcode)
self.assertEqual('Added new image with ID: 1', out.strip())
# 2. Verify image added as public image
cmd = "bin/glance --port=%d index" % api_port
exitcode, out, err = execute(cmd)
self.assertEqual(0, exitcode)
self.assertEqual('No public images found.', out.strip())
# 3. Update the image to make it public
cmd = "bin/glance --port=%d update 1 is_public=True" % api_port
exitcode, out, err = execute(cmd)
self.assertEqual(0, exitcode)
self.assertEqual('Updated image 1', out.strip())
# 4. Verify image 1 in list of public images
cmd = "bin/glance --port=%d index" % api_port
exitcode, out, err = execute(cmd)
self.assertEqual(0, exitcode)
lines = out.split("\n")
first_line = lines[0]
self.assertEqual('Found 1 public images...', first_line)
image_data_line = lines[3]
self.assertTrue('MyImage' in image_data_line)
self.stop_servers()

302
tests/functional/test_curl_api.py Normal file
View File

@ -0,0 +1,302 @@
# vim: tabstop=4 shiftwidth=4 softtabstop=4
# Copyright 2011 OpenStack, LLC
# 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.
"""Functional test case that utilizes cURL against the API server"""
import json
import os
import unittest
from tests import functional
from tests.utils import execute
FIVE_KB = 5 * 1024
class TestCurlApi(functional.FunctionalTest):
"""Functional tests using straight cURL against the API server"""
def test_get_head_simple_post(self):
"""
We test the following sequential series of actions:
0. GET /images
- Verify no public images
1. GET /images/detail
- Verify no public images
2. HEAD /images/1
- Verify 404 returned
3. POST /images with public image named Image1 with a location
attribute and no custom properties
- Verify 201 returned
4. HEAD /images/1
- Verify HTTP headers have correct information we just added
5. GET /images/1
- Verify all information on image we just added is correct
6. GET /images
- Verify the image we just added is returned
7. GET /images/detail
- Verify the image we just added is returned
8. PUT /images/1 with custom properties of "distro" and "arch"
- Verify 200 returned
9. GET /images/1
- Verify updated information about image was stored
"""
self.cleanup()
api_port, reg_port, conf_file = self.start_servers()
# 0. GET /images
# Verify no public images
cmd = "curl -g http://0.0.0.0:%d/images" % api_port
exitcode, out, err = execute(cmd)
self.assertEqual(0, exitcode)
self.assertEqual('{"images": []}', out.strip())
# 1. GET /images/detail
# Verify no public images
cmd = "curl -g http://0.0.0.0:%d/images/detail" % api_port
exitcode, out, err = execute(cmd)
self.assertEqual(0, exitcode)
self.assertEqual('{"images": []}', out.strip())
# 2. HEAD /images/1
# Verify 404 returned
cmd = "curl -i -X HEAD http://0.0.0.0:%d/images/1" % api_port
exitcode, out, err = execute(cmd)
self.assertEqual(0, exitcode)
lines = out.split("\r\n")
status_line = lines[0]
self.assertEqual("HTTP/1.1 404 Not Found", status_line)
# 3. POST /images with public image named Image1
# attribute and no custom properties. Verify a 200 OK is returned
image_data = "*" * FIVE_KB
cmd = ("curl -i -X POST "
"-H 'Expect: ' " # Necessary otherwise sends 100 Continue
"-H 'Content-Type: application/octet-stream' "
"-H 'X-Image-Meta-Name: Image1' "
"-H 'X-Image-Meta-Is-Public: True' "
"--data-binary \"%s\" "
"http://0.0.0.0:%d/images") % (image_data, api_port)
exitcode, out, err = execute(cmd)
self.assertEqual(0, exitcode)
lines = out.split("\r\n")
status_line = lines[0]
self.assertEqual("HTTP/1.1 200 OK", status_line)
# 4. HEAD /images
# Verify image found now
cmd = "curl -i -X HEAD http://0.0.0.0:%d/images/1" % api_port
exitcode, out, err = execute(cmd)
self.assertEqual(0, exitcode)
lines = out.split("\r\n")
status_line = lines[0]
self.assertEqual("HTTP/1.1 200 OK", status_line)
self.assertTrue("X-Image-Meta-Name: Image1" in out)
# 5. GET /images/1
# Verify all information on image we just added is correct
cmd = "curl -i -g http://0.0.0.0:%d/images/1" % api_port
exitcode, out, err = execute(cmd)
self.assertEqual(0, exitcode)
lines = out.split("\r\n")
self.assertEqual("HTTP/1.1 200 OK", lines.pop(0))
# Handle the headers
image_headers = {}
std_headers = {}
other_lines = []
for line in lines:
if line.strip() == '':
continue
if line.startswith("X-Image"):
pieces = line.split(":")
key = pieces[0].strip()
value = ":".join(pieces[1:]).strip()
image_headers[key] = value
elif ':' in line:
pieces = line.split(":")
key = pieces[0].strip()
value = ":".join(pieces[1:]).strip()
std_headers[key] = value
else:
other_lines.append(line)
expected_image_headers = {
'X-Image-Meta-Id': '1',
'X-Image-Meta-Name': 'Image1',
'X-Image-Meta-Is_public': 'True',
'X-Image-Meta-Status': 'active',
'X-Image-Meta-Disk_format': '',
'X-Image-Meta-Container_format': '',
'X-Image-Meta-Size': str(FIVE_KB),
'X-Image-Meta-Location': 'file://%s/1' % self.image_dir}
expected_std_headers = {
'Content-Length': str(FIVE_KB),
'Content-Type': 'application/octet-stream'}
for expected_key, expected_value in expected_image_headers.items():
self.assertTrue(expected_key in image_headers,
"Failed to find key %s in image_headers"
% expected_key)
self.assertEqual(expected_value, image_headers[expected_key],
"For key '%s' expected header value '%s'. Got '%s'"
% (expected_key,
expected_value,
image_headers[expected_key]))
for expected_key, expected_value in expected_std_headers.items():
self.assertTrue(expected_key in std_headers,
"Failed to find key %s in std_headers"
% expected_key)
self.assertEqual(expected_value, std_headers[expected_key],
"For key '%s' expected header value '%s'. Got '%s'"
% (expected_key,
expected_value,
std_headers[expected_key]))
# Now the image data...
expected_image_data = "*" * FIVE_KB
# Should only be a single "line" left, and
# that's the image data
self.assertEqual(1, len(other_lines))
self.assertEqual(expected_image_data, other_lines[0])
# 6. GET /images
# Verify no public images
cmd = "curl -g http://0.0.0.0:%d/images" % api_port
exitcode, out, err = execute(cmd)
self.assertEqual(0, exitcode)
expected_result = {"images": [
{"container_format": None,
"disk_format": None,
"id": 1,
"name": "Image1",
"size": 5120}]}
self.assertEqual(expected_result, json.loads(out.strip()))
# 7. GET /images/detail
# Verify image and all its metadata
cmd = "curl -g http://0.0.0.0:%d/images/detail" % api_port
exitcode, out, err = execute(cmd)
self.assertEqual(0, exitcode)
expected_image = {
"status": "active",
"name": "Image1",
"deleted": False,
"container_format": None,
"disk_format": None,
"id": 1,
"location": "file://%s/1" % self.image_dir,
"is_public": True,
"deleted_at": None,
"properties": {},
"size": 5120}
image = json.loads(out.strip())['images'][0]
for expected_key, expected_value in expected_image.items():
self.assertTrue(expected_key in image,
"Failed to find key %s in image"
% expected_key)
self.assertEqual(expected_value, expected_image[expected_key],
"For key '%s' expected header value '%s'. Got '%s'"
% (expected_key,
expected_value,
image[expected_key]))
# 8. PUT /images/1 with custom properties of "distro" and "arch"
# Verify 200 returned
cmd = ("curl -i -X PUT "
"-H 'X-Image-Meta-Property-Distro: Ubuntu' "
"-H 'X-Image-Meta-Property-Arch: x86_64' "
"http://0.0.0.0:%d/images/1") % api_port
exitcode, out, err = execute(cmd)
self.assertEqual(0, exitcode)
lines = out.split("\r\n")
status_line = lines[0]
self.assertEqual("HTTP/1.1 200 OK", status_line)
# 9. GET /images/detail
# Verify image and all its metadata
cmd = "curl -g http://0.0.0.0:%d/images/detail" % api_port
exitcode, out, err = execute(cmd)
self.assertEqual(0, exitcode)
expected_image = {
"status": "active",
"name": "Image1",
"deleted": False,
"container_format": None,
"disk_format": None,
"id": 1,
"location": "file://%s/1" % self.image_dir,
"is_public": True,
"deleted_at": None,
"properties": {'distro': 'Ubuntu', 'arch': 'x86_64'},
"size": 5120}
image = json.loads(out.strip())['images'][0]
for expected_key, expected_value in expected_image.items():
self.assertTrue(expected_key in image,
"Failed to find key %s in image"
% expected_key)
self.assertEqual(expected_value, image[expected_key],
"For key '%s' expected header value '%s'. Got '%s'"
% (expected_key,
expected_value,
image[expected_key]))
self.stop_servers()

79
tests/functional/test_logging.py Normal file
View File

@ -0,0 +1,79 @@
# vim: tabstop=4 shiftwidth=4 softtabstop=4
# Copyright 2011 OpenStack, LLC
# 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.
import os
import unittest
from tests import functional
from tests.utils import execute
class TestLogging(functional.FunctionalTest):
"""Tests that logging can be configured correctly"""
def test_logfile(self):
"""
A test that logging can be configured properly from the
glance.conf file with the log_file option.
We start both servers daemonized with a temporary config
file that has some logging options in it.
We then use curl to issue a few requests and verify that each server's
logging statements were logged to the one log file
"""
self.cleanup()
api_port, reg_port, conf_file = self.start_servers()
cmd = "curl -X POST -H 'Content-Type: application/octet-stream' "\
"-H 'X-Image-Meta-Name: ImageName' "\
"-H 'X-Image-Meta-Disk-Format: Invalid' "\
"http://0.0.0.0:%d/images" % api_port
ignored, out, err = execute(cmd)
self.assertTrue('Invalid disk format' in out,
"Could not find 'Invalid disk format' "
"in output: %s" % out)
self.assertTrue(os.path.exists(self.api_log_file),
"API Logfile %s does not exist!"
% self.api_log_file)
self.assertTrue(os.path.exists(self.registry_log_file),
"Registry Logfile %s does not exist!"
% self.registry_log_file)
api_logfile_contents = open(self.api_log_file, 'rb').read()
registry_logfile_contents = open(self.registry_log_file, 'rb').read()
# Check that BOTH the glance API and registry server
# modules are logged to their respective logfiles.
self.assertTrue('[glance.server]'
in api_logfile_contents,
"Could not find '[glance.server]' "
"in API logfile: %s" % api_logfile_contents)
self.assertTrue('[glance.registry.server]'
in registry_logfile_contents,
"Could not find '[glance.registry.server]' "
"in Registry logfile: %s" % registry_logfile_contents)
# Test that the error we caused above is in the log
self.assertTrue('Invalid disk format' in api_logfile_contents,
"Could not find 'Invalid disk format' "
"in API logfile: %s" % api_logfile_contents)
self.stop_servers()

63
tests/functional/test_misc.py Normal file
View File

@ -0,0 +1,63 @@
# vim: tabstop=4 shiftwidth=4 softtabstop=4
# Copyright 2011 OpenStack, LLC
# 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.
import unittest
from tests import functional
from tests.utils import execute
class TestMiscellaneous(functional.FunctionalTest):
"""Some random tests for various bugs and stuff"""
def test_exception_not_eaten_from_registry_to_api(self):
"""
A test for LP bug #704854 -- Exception thrown by registry
server is consumed by API server.
We start both servers daemonized.
We then use curl to try adding an image that does not
meet validation requirements on the registry server and test
that the error returned from the API server to curl is appropriate
We also fire the glance-upload tool against the API server
and verify that glance-upload doesn't eat the exception either...
"""
self.cleanup()
api_port, reg_port, conf_file = self.start_servers()
cmd = "curl -g http://0.0.0.0:%d/images" % api_port
exitcode, out, err = execute(cmd)
self.assertEqual(0, exitcode)
self.assertEqual('{"images": []}', out.strip())
cmd = "curl -X POST -H 'Content-Type: application/octet-stream' "\
"-H 'X-Image-Meta-Name: ImageName' "\
"-H 'X-Image-Meta-Disk-Format: Invalid' "\
"http://0.0.0.0:%d/images" % api_port
ignored, out, err = execute(cmd)
self.assertTrue('Invalid disk format' in out,
"Could not find 'Invalid disk format' "
"in output: %s" % out)
self.stop_servers()

2
tests/unit/test_migrations.py
View File

@ -36,7 +36,7 @@ from sqlalchemy.pool import NullPool
from glance.common import exception
import glance.registry.db.migration as migration_api
from tests.unit.test_misc import execute
from tests.utils import execute
class TestMigrations(unittest.TestCase):

352
tests/unit/test_misc.py
View File

@ -1,352 +0,0 @@
# vim: tabstop=4 shiftwidth=4 softtabstop=4
# Copyright 2011 OpenStack, LLC
# 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.
import os
import shutil
import signal
import subprocess
import tempfile
import time
import unittest
from glance import utils
def execute(cmd):
env = os.environ.copy()
# Make sure that we use the programs in the
# current source directory's bin/ directory.
env['PATH'] = os.path.join(os.getcwd(), 'bin') + ':' + env['PATH']
process = subprocess.Popen(cmd,
shell=True,
stdin=subprocess.PIPE,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
env=env)
result = process.communicate()
(out, err) = result
exitcode = process.returncode
if process.returncode != 0:
msg = "Command %(cmd)s did not succeed. Returned an exit "\
"code of %(exitcode)d."\
"\n\nSTDOUT: %(out)s"\
"\n\nSTDERR: %(err)s" % locals()
raise RuntimeError(msg)
return exitcode, out, err
class TestMiscellaneous(unittest.TestCase):
"""Some random tests for various bugs and stuff"""
def tearDown(self):
self._cleanup_test_servers()
def _cleanup_test_servers(self):
# Clean up any leftover test servers...
pid_files = ('glance-api.pid', 'glance-registry.pid')
for pid_file in pid_files:
if os.path.exists(pid_file):
pid = int(open(pid_file).read().strip())
try:
os.killpg(pid, signal.SIGTERM)
except:
pass # Ignore if the process group is dead
os.unlink(pid_file)
def test_headers_are_unicode(self):
"""
Verifies that the headers returned by conversion code are unicode.
Headers are passed via http in non-testing mode, which automatically
converts them to unicode. Verifying that the method does the
conversion proves that we aren't passing data that works in tests
but will fail in production.
"""
fixture = {'name': 'fake public image',
'is_public': True,
'type': 'kernel',
'size': 19,
'location': "file:///tmp/glance-tests/2",
'properties': {'distro': 'Ubuntu 10.04 LTS'}}
headers = utils.image_meta_to_http_headers(fixture)
for k, v in headers.iteritems():
self.assert_(isinstance(v, unicode), "%s is not unicode" % v)
def test_data_passed_properly_through_headers(self):
"""
Verifies that data is the same after being passed through headers
"""
fixture = {'name': 'fake public image',
'is_public': True,
'deleted': False,
'type': 'kernel',
'name': None,
'size': 19,
'location': "file:///tmp/glance-tests/2",
'properties': {'distro': 'Ubuntu 10.04 LTS'}}
headers = utils.image_meta_to_http_headers(fixture)
class FakeResponse():
pass
response = FakeResponse()
response.headers = headers
result = utils.get_image_meta_from_headers(response)
for k, v in fixture.iteritems():
self.assertEqual(v, result[k])
def test_exception_not_eaten_from_registry_to_api(self):
"""
A test for LP bug #704854 -- Exception thrown by registry
server is consumed by API server.
We start both servers daemonized.
We then use curl to try adding an image that does not
meet validation requirements on the registry server and test
that the error returned from the API server to curl is appropriate
We also fire the glance-upload tool against the API server
and verify that glance-upload doesn't eat the exception either...
"""
self._cleanup_test_servers()
# Port numbers hopefully not used by anything...
api_port = 32001
reg_port = 32000
image_dir = "/tmp/test.images.%d" % api_port
sql_connection = os.environ.get('GLANCE_SQL_CONNECTION', "sqlite://")
if os.path.exists(image_dir):
shutil.rmtree(image_dir)
# A config file to use just for this test...we don't want
# to trample on currently-running Glance servers, now do we?
with tempfile.NamedTemporaryFile() as conf_file:
conf_contents = """[DEFAULT]
verbose = True
debug = True
[app:glance-api]
paste.app_factory = glance.server:app_factory
filesystem_store_datadir=%(image_dir)s
default_store = file
bind_host = 0.0.0.0
bind_port = %(api_port)s
registry_host = 0.0.0.0
registry_port = %(reg_port)s
[app:glance-registry]
paste.app_factory = glance.registry.server:app_factory
bind_host = 0.0.0.0
bind_port = %(reg_port)s
sql_connection = %(sql_connection)s
sql_idle_timeout = 3600
""" % locals()
conf_file.write(conf_contents)
conf_file.flush()
conf_file_name = conf_file.name
venv = ""
if 'VIRTUAL_ENV' in os.environ:
venv = "tools/with_venv.sh "
# Start up the API and default registry server
cmd = venv + "./bin/glance-control api start "\
"%s --pid-file=glance-api.pid" % conf_file_name
exitcode, out, err = execute(cmd)
self.assertEquals(0, exitcode)
self.assertTrue("Starting glance-api with" in out)
cmd = venv + "./bin/glance-control registry start "\
"%s --pid-file=glance-registry.pid" % conf_file_name
exitcode, out, err = execute(cmd)
self.assertEquals(0, exitcode)
self.assertTrue("Starting glance-registry with" in out)
time.sleep(2) # Gotta give some time for spin up...
cmd = "curl -g http://0.0.0.0:%d/images" % api_port
exitcode, out, err = execute(cmd)
self.assertEquals(0, exitcode)
self.assertEquals('{"images": []}', out.strip())
cmd = "curl -X POST -H 'Content-Type: application/octet-stream' "\
"-H 'X-Image-Meta-Name: ImageName' "\
"-H 'X-Image-Meta-Disk-Format: Invalid' "\
"http://0.0.0.0:%d/images" % api_port
ignored, out, err = execute(cmd)
self.assertTrue('Invalid disk format' in out,
"Could not find 'Invalid disk format' "
"in output: %s" % out)
# Spin down the API and default registry server
cmd = "./bin/glance-control api stop "\
"%s --pid-file=glance-api.pid" % conf_file_name
ignored, out, err = execute(cmd)
cmd = "./bin/glance-control registry stop "\
"%s --pid-file=glance-registry.pid" % conf_file_name
ignored, out, err = execute(cmd)
# TODO(jaypipes): Move this to separate test file once
# LP Bug#731304 moves execute() out to a common file, etc
class TestLogging(unittest.TestCase):
"""Tests that logging can be configured correctly"""
def setUp(self):
self.logfiles = []
def tearDown(self):
self._cleanup_test_servers()
self._cleanup_log_files()
def _cleanup_test_servers(self):
# Clean up any leftover test servers...
pid_files = ('glance-api.pid', 'glance-registry.pid')
for pid_file in pid_files:
if os.path.exists(pid_file):
pid = int(open(pid_file).read().strip())
try:
os.killpg(pid, signal.SIGTERM)
except:
pass # Ignore if the process group is dead
os.unlink(pid_file)
def _cleanup_log_files(self):
for f in self.logfiles:
if os.path.exists(f):
os.unlink(f)
def test_logfile(self):
"""
A test that logging can be configured properly from the
glance.conf file with the log_file option.
We start both servers daemonized with a temporary config
file that has some logging options in it.
We then use curl to issue a few requests and verify that each server's
logging statements were logged to the one log file
"""
logfile = "/tmp/test_logfile.log"
self.logfiles.append(logfile)
if os.path.exists(logfile):
os.unlink(logfile)
self._cleanup_test_servers()
# Port numbers hopefully not used by anything...
api_port = 32001
reg_port = 32000
image_dir = "/tmp/test.images.%d" % api_port
if os.path.exists(image_dir):
shutil.rmtree(image_dir)
# A config file to use just for this test...we don't want
# to trample on currently-running Glance servers, now do we?
with tempfile.NamedTemporaryFile() as conf_file:
conf_contents = """[DEFAULT]
verbose = True
debug = True
log_file = %(logfile)s
[app:glance-api]
paste.app_factory = glance.server:app_factory
filesystem_store_datadir=%(image_dir)s
default_store = file
bind_host = 0.0.0.0
bind_port = %(api_port)s
registry_host = 0.0.0.0
registry_port = %(reg_port)s
[app:glance-registry]
paste.app_factory = glance.registry.server:app_factory
bind_host = 0.0.0.0
bind_port = %(reg_port)s
sql_connection = sqlite://
sql_idle_timeout = 3600
""" % locals()
conf_file.write(conf_contents)
conf_file.flush()
conf_file_name = conf_file.name
venv = ""
if 'VIRTUAL_ENV' in os.environ:
venv = "tools/with_venv.sh "
# Start up the API and default registry server
cmd = venv + "./bin/glance-control api start "\
"%s --pid-file=glance-api.pid" % conf_file_name
exitcode, out, err = execute(cmd)
self.assertEquals(0, exitcode)
self.assertTrue("Starting glance-api with" in out)
cmd = venv + "./bin/glance-control registry start "\
"%s --pid-file=glance-registry.pid" % conf_file_name
exitcode, out, err = execute(cmd)
self.assertEquals(0, exitcode)
self.assertTrue("Starting glance-registry with" in out)
time.sleep(2) # Gotta give some time for spin up...
cmd = "curl -X POST -H 'Content-Type: application/octet-stream' "\
"-H 'X-Image-Meta-Name: ImageName' "\
"-H 'X-Image-Meta-Disk-Format: Invalid' "\
"http://0.0.0.0:%d/images" % api_port
ignored, out, err = execute(cmd)
self.assertTrue('Invalid disk format' in out,
"Could not find 'Invalid disk format' "
"in output: %s" % out)
self.assertTrue(os.path.exists(logfile),
"Logfile %s does not exist!" % logfile)
logfile_contents = open(logfile, 'rb').read()
# Check that BOTH the glance API and registry server
# modules are logged to the file.
self.assertTrue('[glance.server]' in logfile_contents,
"Could not find '[glance.server]' "
"in logfile: %s" % logfile_contents)
self.assertTrue('[glance.registry.server]' in logfile_contents,
"Could not find '[glance.registry.server]' "
"in logfile: %s" % logfile_contents)
# Test that the error we caused above is in the log
self.assertTrue('Invalid disk format' in logfile_contents,
"Could not find 'Invalid disk format' "
"in logfile: %s" % logfile_contents)
# Spin down the API and default registry server
cmd = "./bin/glance-control api stop "\
"%s --pid-file=glance-api.pid" % conf_file_name
ignored, out, err = execute(cmd)
cmd = "./bin/glance-control registry stop "\
"%s --pid-file=glance-registry.pid" % conf_file_name
ignored, out, err = execute(cmd)

67
tests/unit/test_utils.py Normal file
View File

@ -0,0 +1,67 @@
# vim: tabstop=4 shiftwidth=4 softtabstop=4
# Copyright 2011 OpenStack, LLC
# 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.
import unittest
from glance import utils
class TestUtils(unittest.TestCase):
"""Test routines in glance.utils"""
def test_headers_are_unicode(self):
"""
Verifies that the headers returned by conversion code are unicode.
Headers are passed via http in non-testing mode, which automatically
converts them to unicode. Verifying that the method does the
conversion proves that we aren't passing data that works in tests
but will fail in production.
"""
fixture = {'name': 'fake public image',
'is_public': True,
'type': 'kernel',
'size': 19,
'location': "file:///tmp/glance-tests/2",
'properties': {'distro': 'Ubuntu 10.04 LTS'}}
headers = utils.image_meta_to_http_headers(fixture)
for k, v in headers.iteritems():
self.assert_(isinstance(v, unicode), "%s is not unicode" % v)
def test_data_passed_properly_through_headers(self):
"""
Verifies that data is the same after being passed through headers
"""
fixture = {'name': 'fake public image',
'is_public': True,
'deleted': False,
'type': 'kernel',
'name': None,
'size': 19,
'location': "file:///tmp/glance-tests/2",
'properties': {'distro': 'Ubuntu 10.04 LTS'}}
headers = utils.image_meta_to_http_headers(fixture)
class FakeResponse():
pass
response = FakeResponse()
response.headers = headers
result = utils.get_image_meta_from_headers(response)
for k, v in fixture.iteritems():
self.assertEqual(v, result[k])

46
tests/utils.py
View File

@ -1,6 +1,6 @@
# vim: tabstop=4 shiftwidth=4 softtabstop=4
# Copyright 2010 OpenStack, LLC
# Copyright 2010-2011 OpenStack, LLC
# All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
@ -17,11 +17,41 @@
"""Common utilities used in testing"""
import os
import socket
import subprocess
def is_swift_available():
"""Returns True if Swift/Cloudfiles is importable"""
try:
import swift
return True
except ImportError:
return False
def execute(cmd):
env = os.environ.copy()
# Make sure that we use the programs in the
# current source directory's bin/ directory.
env['PATH'] = os.path.join(os.getcwd(), 'bin') + ':' + env['PATH']
process = subprocess.Popen(cmd,
shell=True,
stdin=subprocess.PIPE,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
env=env)
result = process.communicate()
(out, err) = result
exitcode = process.returncode
if process.returncode != 0:
msg = "Command %(cmd)s did not succeed. Returned an exit "\
"code of %(exitcode)d."\
"\n\nSTDOUT: %(out)s"\
"\n\nSTDERR: %(err)s" % locals()
raise RuntimeError(msg)
return exitcode, out, err
def get_unused_port():
"""
Returns an unused port on localhost.
"""
s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
s.bind(('localhost', 0))
addr, port = s.getsockname()
s.close()
return port