test_command=OS_STDOUT_CAPTURE=${OS_STDOUT_CAPTURE:-1} \
OS_STDERR_CAPTURE=${OS_STDERR_CAPTURE:-1} \
OS_TEST_TIMEOUT=${OS_TEST_TIMEOUT:-60} \
- ${PYTHON:-python} -m subunit.run discover -t ./ ./cinder/tests $LISTOPT $IDOPTION
+ ${PYTHON:-python} -m subunit.run discover -t ./ ${OS_TEST_PATH:-./cinder/tests/unit} $LISTOPT $IDOPTION
test_id_option=--load-list $IDFILE
test_list_option=--list
# License for the specific language governing permissions and limitations
# under the License.
-from oslo_log import log as logging
from oslo_serialization import jsonutils
from oslo_utils import netutils
import requests
from six.moves import urllib
-from cinder.i18n import _, _LI
-
-
-LOG = logging.getLogger(__name__)
+from cinder.i18n import _
class OpenStackApiException(Exception):
relative_url = parsed_url.path
if parsed_url.query:
relative_url = relative_url + "?" + parsed_url.query
- LOG.info(_LI("Doing %(method)s on %(relative_url)s"),
- {'method': method, 'relative_url': relative_url})
- if body:
- LOG.info(_LI("Body: %s") % body)
if port:
_url = "%s://%s:%d%s" % (scheme, hostname, int(port), relative_url)
headers=headers)
http_status = response.status_code
- LOG.debug("%(auth_uri)s => code %(http_status)s",
- {'auth_uri': auth_uri, 'http_status': http_status})
if http_status == 401:
raise OpenStackApiAuthenticationException(response=response)
response = self.request(full_uri, **kwargs)
http_status = response.status_code
- LOG.debug("%(relative_uri)s => code %(http_status)s",
- {'relative_uri': relative_uri, 'http_status': http_status})
if check_response_status:
if http_status not in check_response_status:
def _decode_json(self, response):
body = response.text
- LOG.debug("Decoding JSON: %s" % (body))
if body:
return jsonutils.loads(body)
else:
# under the License.
"""
-Provides common functionality for integrated unit tests
+Provides common functionality for functional tests
"""
import os.path
import random
import fixtures
import mock
from oslo_config import cfg
-from oslo_log import log as logging
from cinder import service
from cinder import test # For the flags
-from cinder.tests.unit.integrated.api import client
+from cinder.tests.functional.api import client
CONF = cfg.CONF
-LOG = logging.getLogger(__name__)
def generate_random_alphanumeric(length):
candidate = prefix + generate_random_alphanumeric(8)
if candidate not in items:
return candidate
- LOG.debug("Random collision on %s", candidate)
-class _IntegratedTestBase(test.TestCase):
+class _FunctionalTestBase(test.TestCase):
def setUp(self):
- super(_IntegratedTestBase, self).setUp()
+ super(_FunctionalTestBase, self).setUp()
f = self._get_flags()
self.flags(**f)
def _start_api_service(self):
default_conf = os.path.abspath(os.path.join(
- os.path.dirname(__file__), '..', '..', '..', '..',
+ os.path.dirname(__file__), '..', '..', '..',
'etc/cinder/api-paste.ini'))
CONF.api_paste_config = default_conf
self.osapi = service.WSGIService("osapi_volume")
# FIXME(ja): this is not the auth url - this is the service url
# FIXME(ja): this needs fixed in nova as well
self.auth_url = 'http://%s:%s/v2' % (self.osapi.host, self.osapi.port)
- LOG.warning(self.auth_url)
def _get_flags(self):
"""An opportunity to setup flags, before the services are started."""
server = {}
image = self.api.get_images()[0]
- LOG.debug("Image: %s", image)
if 'imageRef' in image:
image_href = image['imageRef']
# Set a valid flavorId
flavor = self.api.get_flavors()[0]
- LOG.debug("Using flavor: %s", flavor)
server['flavorRef'] = 'http://fake.server/%s' % flavor['id']
# Set a valid server name
-# Copyright (c) 2011 X.commerce, a business unit of eBay Inc.
-# Copyright 2011 OpenStack Foundation
+# Copyright 2011 Justin Santa Barbara
# All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# License for the specific language governing permissions and limitations
# under the License.
+
import iso8601
from lxml import etree
from oslo_config import cfg
from cinder.api import extensions
from cinder.api.v1 import router
from cinder.api import xmlutil
-from cinder import test
+from cinder.tests.functional import functional_helpers
NS = "{http://docs.openstack.org/common/api/v1.0}"
+CONF = cfg.CONF
-CONF = cfg.CONF
+class ExtensionTestCase(functional_helpers._FunctionalTestBase):
+ def _get_flags(self):
+ f = super(ExtensionTestCase, self)._get_flags()
+ f['osapi_volume_extension'] = CONF.osapi_volume_extension[:]
+ f['osapi_volume_extension'].append(
+ 'cinder.tests.functional.api.foxinsocks.Foxinsocks')
+ return f
-class ExtensionTestCase(test.TestCase):
- def setUp(self):
- super(ExtensionTestCase, self).setUp()
- ext_list = CONF.osapi_volume_extension[:]
- fox = ('cinder.tests.unit.api.extensions.foxinsocks.Foxinsocks')
- if fox not in ext_list:
- ext_list.append(fox)
- self.flags(osapi_volume_extension=ext_list)
+class ExtensionsTest(ExtensionTestCase):
+ def test_get_foxnsocks(self):
+ """Simple check that fox-n-socks works."""
+ response = self.api.api_request('/foxnsocks')
+ foxnsocks = response.text
+ self.assertEqual('Try to say this Mr. Knox, sir...', foxnsocks)
class ExtensionControllerTest(ExtensionTestCase):
return controller_extensions
-class ExtensionControllerIdFormatTest(test.TestCase):
+class ExtensionControllerIdFormatTest(ExtensionTestCase):
def _bounce_id(self, test_id):
# License for the specific language governing permissions and limitations
# under the License.
-from oslo_log import log as logging
+from cinder.tests.functional import functional_helpers
-from cinder.tests.unit.integrated import integrated_helpers
-
-LOG = logging.getLogger(__name__)
-
-
-class LoginTest(integrated_helpers._IntegratedTestBase):
+class LoginTest(functional_helpers._FunctionalTestBase):
def test_login(self):
"""Simple check - we list volumes - so we know we're logged in."""
volumes = self.api.get_volumes()
- for volume in volumes:
- LOG.debug("volume: %s" % volume)
+ self.assertIsNotNone(volumes)
import time
-from oslo_log import log as logging
import testtools
+from cinder import service
+from cinder.tests.functional.api import client
+from cinder.tests.functional import functional_helpers
from cinder.tests.unit import fake_driver
-from cinder.tests.unit.integrated.api import client
-from cinder.tests.unit.integrated import integrated_helpers
-LOG = logging.getLogger(__name__)
-
-
-class VolumesTest(integrated_helpers._IntegratedTestBase):
+class VolumesTest(functional_helpers._FunctionalTestBase):
def setUp(self):
super(VolumesTest, self).setUp()
fake_driver.LoggingVolumeDriver.clear_logs()
+ def _start_api_service(self):
+ self.osapi = service.WSGIService("osapi_volume")
+ self.osapi.start()
+ self.auth_url = 'http://%s:%s/v2' % (self.osapi.host, self.osapi.port)
+
def _get_flags(self):
f = super(VolumesTest, self)._get_flags()
f['volume_driver'] = \
def test_get_volumes_summary(self):
"""Simple check that listing volumes works."""
volumes = self.api.get_volumes(False)
- for volume in volumes:
- LOG.debug("volume: %s", volume)
+ self.assertIsNotNone(volumes)
def test_get_volumes(self):
"""Simple check that listing volumes works."""
volumes = self.api.get_volumes()
- for volume in volumes:
- LOG.debug("volume: %s", volume)
+ self.assertIsNotNone(volumes)
def _poll_while(self, volume_id, continue_states, max_retries=5):
"""Poll (briefly) while the state is in continue_states."""
found_volume = self.api.get_volume(volume_id)
except client.OpenStackApiNotFoundException:
found_volume = None
- LOG.debug("Got 404, proceeding")
break
- LOG.debug("Found %s", found_volume)
-
self.assertEqual(volume_id, found_volume['id'])
if found_volume['status'] not in continue_states:
# Create volume
created_volume = self.api.post_volume({'volume': {'size': 1}})
- LOG.debug("created_volume: %s", created_volume)
self.assertTrue(created_volume['id'])
created_volume_id = created_volume['id']
# Should be gone
self.assertFalse(found_volume)
- LOG.debug("Logs: %s", fake_driver.LoggingVolumeDriver.all_logs())
-
create_actions = fake_driver.LoggingVolumeDriver.logs_like(
'create_volume',
id=created_volume_id)
- LOG.debug("Create_Actions: %s", create_actions)
self.assertEqual(1, len(create_actions))
create_action = create_actions[0]
created_volume = self.api.post_volume(
{'volume': {'size': 1,
'metadata': metadata}})
- LOG.debug("created_volume: %s", created_volume)
self.assertTrue(created_volume['id'])
created_volume_id = created_volume['id']
created_volume = self.api.post_volume(
{'volume': {'size': 1,
'availability_zone': availability_zone}})
- LOG.debug("created_volume: %s", created_volume)
self.assertTrue(created_volume['id'])
created_volume_id = created_volume['id']
# under the License.
from lxml import etree
-from oslo_log import log as logging
from cinder.api import common
-from cinder.tests.unit.integrated import integrated_helpers
+from cinder.tests.functional import functional_helpers
-LOG = logging.getLogger(__name__)
-
-
-class XmlTests(integrated_helpers._IntegratedTestBase):
+class XmlTests(functional_helpers._FunctionalTestBase):
"""Some basic XML sanity checks."""
# FIXME(ja): does cinder need limits?
response = self.api.api_request('/volumes', headers=headers,
stream=True)
data = response.raw
- LOG.warning("data: %s", data)
root = etree.parse(data).getroot()
self.assertEqual(common.XML_NS_V2, root.nsmap.get(None))
+++ /dev/null
-# Copyright 2011 Justin Santa Barbara
-# 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.
-
-
-from oslo_config import cfg
-from oslo_log import log as logging
-
-from cinder.tests.unit.integrated import integrated_helpers
-
-
-CONF = cfg.CONF
-
-LOG = logging.getLogger(__name__)
-
-
-class ExtensionsTest(integrated_helpers._IntegratedTestBase):
- def _get_flags(self):
- f = super(ExtensionsTest, self)._get_flags()
- f['osapi_volume_extension'] = CONF.osapi_volume_extension[:]
- f['osapi_volume_extension'].append(
- 'cinder.tests.unit.api.extensions.foxinsocks.Foxinsocks')
- return f
-
- def test_get_foxnsocks(self):
- """Simple check that fox-n-socks works."""
- response = self.api.api_request('/foxnsocks')
- foxnsocks = response.text
- self.assertEqual('Try to say this Mr. Knox, sir...', foxnsocks)
# to ncpu, to specify something else use
# the concurrency=<n> option.
# call ie: 'tox -epy27 -- --concurrency=4'
-commands = ostestr {posargs}
+commands = python setup.py testr --testr-args='{posargs}'
whitelist_externals = bash
passenv = *_proxy *_PROXY
commands =
ostestr --whitelist_file=tests-py3.txt
+[testenv:functional]
+setenv =
+ OS_TEST_PATH = ./cinder/tests/functional
+
[testenv:py34-constraints]
install_command = {[testenv:common-constraints]install_command}
commands = {[testenv:py34]commands}
Code Review / openstack-build / cinder-build.git / commitdiff