Commit 72d3b842 authored by Ilias Tsitsimpis's avatar Ilias Tsitsimpis
Browse files

burnin: Add PithosTestSuite

parent 89cf47aa
......@@ -43,13 +43,15 @@ from synnefo_tools import version
from synnefo_tools.burnin import common
from synnefo_tools.burnin.astakos_tests import AstakosTestSuite
from synnefo_tools.burnin.cyclades_tests import FlavorsTestSuite
from synnefo_tools.burnin.pithos_tests import PithosTestSuite
# --------------------------------------------------------------------
# Define our TestSuites
TESTSUITES = [
AstakosTestSuite,
FlavorsTestSuite
FlavorsTestSuite,
PithosTestSuite,
]
TSUITES_NAMES = [tsuite.__name__ for tsuite in TESTSUITES]
......
......@@ -49,6 +49,7 @@ except ImportError:
from kamaki.clients.astakos import AstakosClient
from kamaki.clients.compute import ComputeClient
from kamaki.clients.pithos import PithosClient
from synnefo_tools.burnin.logger import Log
......@@ -78,13 +79,18 @@ class BurninTestResult(unittest.TestResult):
# Method could be a function. pylint: disable-msg=R0201
def _test_failed(self, test, err):
"""Test failed"""
# Get class name
if test.__class__.__name__ == "_ErrorHolder":
class_name = test.id().split('.')[-1].rstrip(')')
else:
class_name = test.__class__.__name__
err_msg = str(test) + "... failed (%s)."
timestamp = datetime.datetime.strftime(
datetime.datetime.now(), "%a %b %d %Y %H:%M:%S")
logger.error(test.__class__.__name__, err_msg, timestamp)
logger.error(class_name, err_msg, timestamp)
(err_type, err_value, err_trace) = err
trcback = traceback.format_exception(err_type, err_value, err_trace)
logger.info(test.__class__.__name__, trcback)
logger.info(class_name, trcback)
def addError(self, test, err): # noqa
"""Called when the test case test raises an unexpected exception"""
......@@ -104,20 +110,28 @@ class Clients(object):
"""Our kamaki clients"""
auth_url = None
token = None
# Astakos
astakos = None
retry = CONNECTION_RETRY_LIMIT
# Compute
compute = None
compute_url = None
# Cyclades
cyclades = None
# Pithos
pithos = None
pithos_url = None
# Too many public methods (45/20). pylint: disable-msg=R0904
class BurninTests(unittest.TestCase):
"""Common class that all burnin tests should implement"""
clients = Clients()
opts = None
run_id = None
use_ipv6 = None
action_timeout = None
action_warning = None
query_interval = None
@classmethod
def setUpClass(cls): # noqa
......@@ -154,6 +168,13 @@ class BurninTests(unittest.TestCase):
self.clients.compute_url, self.clients.token)
self.clients.compute.CONNECTION_RETRY_LIMIT = self.clients.retry
self.clients.pithos_url = self.clients.astakos.\
get_service_endpoints('object-store')['publicURL']
self.info("Pithos url is %s", self.clients.pithos_url)
self.clients.pithos = PithosClient(
self.clients.pithos_url, self.clients.token)
self.clients.pithos.CONNECTION_RETRY_LIMIT = self.clients.retry
# ----------------------------------
# Loggers helper functions
def log(self, msg, *args):
......@@ -201,6 +222,32 @@ class BurninTests(unittest.TestCase):
flavors = self.clients.compute.list_flavors(detail=detail)
return flavors
def _set_pithos_account(self, account):
"""Set the pithos account"""
assert account, "No pithos account was given"
self.info("Setting pithos account to %s", account)
self.clients.pithos.account = account
def _get_list_of_containers(self, account=None):
"""Get list of containers"""
if account is not None:
self._set_pithos_account(account)
self.info("Getting list of containers")
return self.clients.pithos.list_containers()
def _create_pithos_container(self, container):
"""Create a pithos container
If the container exists, nothing will happen
"""
assert container, "No pithos container was given"
self.info("Creating pithos container %s", container)
self.clients.pithos.container = container
self.clients.pithos.container_put()
# --------------------------------------------------------------------
# Initialize Burnin
......@@ -221,9 +268,12 @@ def initialize(opts, testsuites):
Clients.token = opts.token
# Pass the rest options to BurninTests
BurninTests.opts = opts
BurninTests.run_id = datetime.datetime.strftime(
datetime.datetime.now(), "%Y%m%d%H%M%S")
BurninTests.use_ipv6 = opts.use_ipv6
BurninTests.action_timeout = opts.action_timeout
BurninTests.action_warning = opts.action_warning
BurninTests.query_interval = opts.query_interval
BurninTests.run_id = SNF_TEST_PREFIX + \
datetime.datetime.strftime(datetime.datetime.now(), "%Y%m%d%H%M%S")
# Choose tests to run
if opts.tests != "all":
......
# Copyright 2013 GRNET S.A. All rights reserved.
#
# Redistribution and use in source and binary forms, with or
# without modification, are permitted provided that the following
# conditions are met:
#
# 1. Redistributions of source code must retain the above
# copyright notice, this list of conditions and the following
# disclaimer.
#
# 2. Redistributions in binary form must reproduce the above
# copyright notice, this list of conditions and the following
# disclaimer in the documentation and/or other materials
# provided with the distribution.
#
# THIS SOFTWARE IS PROVIDED BY GRNET S.A. ``AS IS'' AND ANY EXPRESS
# OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
# WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR
# PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL GRNET S.A OR
# CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
# SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
# LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF
# USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED
# AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT
# LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN
# ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
# POSSIBILITY OF SUCH DAMAGE.
#
# The views and conclusions contained in the software and
# documentation are those of the authors and should not be
# interpreted as representing official policies, either expressed
# or implied, of GRNET S.A.
"""
This is the burnin class that tests the Pithos functionality
"""
import random
import tempfile
from synnefo_tools.burnin import common
# Too many public methods. pylint: disable-msg=R0904
class PithosTestSuite(common.BurninTests):
"""Test Pithos functionality"""
containers = None
created_container = None
def test_001_list_containers(self):
"""Test container list actually returns containers"""
self._set_pithos_account(self._get_uuid())
containers = self._get_list_of_containers()
self._setattr("containers", containers)
self.assertGreater(len(self.containers), 0)
def test_002_unique_containers(self):
"""Test if containers have unique names"""
names = [n['name'] for n in self.containers]
names = sorted(names)
self.assertEqual(sorted(list(set(names))), names)
def test_003_create_container(self):
"""Test creating a new container"""
names = [n['name'] for n in self.containers]
while True:
rand_num = random.randint(1000, 9999)
rand_name = "%s%s" % (self.run_id, rand_num)
self.info("Trying container name %s", rand_name)
if rand_name not in names:
break
self.info("Container name %s already exists", rand_name)
# Create container
self._create_pithos_container(rand_name)
# Verify that container is created
containers = self._get_list_of_containers()
self.info("Verify that container %s is created", rand_name)
names = [n['name'] for n in containers]
self.assertIn(rand_name, names)
# Keep the name of the container so we can remove it
# at cleanup phase, if something goes wrong.
self._setattr("created_container", rand_name)
def test_004_upload_file(self):
"""Test uploading a txt file to Pithos"""
# Create a tmp file
with tempfile.TemporaryFile() as fout:
fout.write("This is a temp file")
fout.seek(0, 0)
# Upload the file,
# The container is the one choosen during the `create_container'
self.clients.pithos.upload_object("test.txt", fout)
def test_005_download_file(self):
"""Test downloading the file from Pithos"""
# Create a tmp directory to save the file
with tempfile.TemporaryFile() as fout:
self.clients.pithos.download_object("test.txt", fout)
# Now read the file
fout.seek(0, 0)
contents = fout.read()
# Compare results
self.info("Comparing contents with the uploaded file")
self.assertEqual(contents, "This is a temp file")
def test_006_remove(self):
"""Test removing files and containers from Pithos"""
self.info("Removing the file %s from container %s",
"test.txt", self.created_container)
# The container is the one choosen during the `create_container'
self.clients.pithos.del_object("test.txt")
self.info("Removing the container %s", self.created_container)
self.clients.pithos.purge_container()
# List containers
containers = self._get_list_of_containers()
self.info("Check that the container %s has been deleted",
self.created_container)
names = [n['name'] for n in containers]
self.assertNotIn(self.created_container, names)
# We successfully deleted our container, no need to do it
# in our clean up phase
self._setattr("created_container", None)
@classmethod
def tearDownClass(cls): # noqa
"""Clean up"""
if cls.created_container is not None:
cls.clients.pithos.del_container(delimiter='/')
cls.clients.pithos.purge_container()
Markdown is supported
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment