Statistics
| Branch: | Tag: | Revision:

root / snf-tools / synnefo_tools / burnin / common.py @ 9355a604

History | View | Annotate | Download (24.9 kB)

1
# Copyright 2013 GRNET S.A. All rights reserved.
2
#
3
# Redistribution and use in source and binary forms, with or
4
# without modification, are permitted provided that the following
5
# conditions are met:
6
#
7
#   1. Redistributions of source code must retain the above
8
#      copyright notice, this list of conditions and the following
9
#      disclaimer.
10
#
11
#   2. Redistributions in binary form must reproduce the above
12
#      copyright notice, this list of conditions and the following
13
#      disclaimer in the documentation and/or other materials
14
#      provided with the distribution.
15
#
16
# THIS SOFTWARE IS PROVIDED BY GRNET S.A. ``AS IS'' AND ANY EXPRESS
17
# OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
18
# WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR
19
# PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL GRNET S.A OR
20
# CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
21
# SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
22
# LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF
23
# USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED
24
# AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT
25
# LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN
26
# ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
27
# POSSIBILITY OF SUCH DAMAGE.
28
#
29
# The views and conclusions contained in the software and
30
# documentation are those of the authors and should not be
31
# interpreted as representing official policies, either expressed
32
# or implied, of GRNET S.A.
33

    
34
"""
35
Common utils for burnin tests
36

37
"""
38

    
39
import re
40
import shutil
41
import unittest
42
import datetime
43
import tempfile
44
import traceback
45

    
46
from kamaki.clients.cyclades import CycladesClient, CycladesNetworkClient
47
from kamaki.clients.astakos import AstakosClient, parse_endpoints
48
from kamaki.clients.compute import ComputeClient
49
from kamaki.clients.pithos import PithosClient
50
from kamaki.clients.image import ImageClient
51

    
52
from synnefo_tools.burnin.logger import Log
53

    
54

    
55
# --------------------------------------------------------------------
56
# Global variables
57
logger = None   # pylint: disable=invalid-name
58
success = None  # pylint: disable=invalid-name
59
SNF_TEST_PREFIX = "snf-test-"
60
CONNECTION_RETRY_LIMIT = 2
61
SYSTEM_USERS = ["images@okeanos.grnet.gr", "images@demo.synnefo.org"]
62
KB = 2**10
63
MB = 2**20
64
GB = 2**30
65

    
66

    
67
# --------------------------------------------------------------------
68
# BurninTestResult class
69
class BurninTestResult(unittest.TestResult):
70
    """Modify the TextTestResult class"""
71
    def __init__(self):
72
        super(BurninTestResult, self).__init__()
73

    
74
        # Test parameters
75
        self.failfast = True
76

    
77
    def startTest(self, test):  # noqa
78
        """Called when the test case test is about to be run"""
79
        super(BurninTestResult, self).startTest(test)
80
        logger.log(test.__class__.__name__, test.shortDescription())
81

    
82
    # pylint: disable=no-self-use
83
    def _test_failed(self, test, err):
84
        """Test failed"""
85
        # Get class name
86
        if test.__class__.__name__ == "_ErrorHolder":
87
            class_name = test.id().split('.')[-1].rstrip(')')
88
        else:
89
            class_name = test.__class__.__name__
90
        err_msg = str(test) + "... failed (%s)."
91
        timestamp = datetime.datetime.strftime(
92
            datetime.datetime.now(), "%a %b %d %Y %H:%M:%S")
93
        logger.error(class_name, err_msg, timestamp)
94
        (err_type, err_value, err_trace) = err
95
        trcback = traceback.format_exception(err_type, err_value, err_trace)
96
        logger.info(class_name, trcback)
97

    
98
    def addError(self, test, err):  # noqa
99
        """Called when the test case test raises an unexpected exception"""
100
        super(BurninTestResult, self).addError(test, err)
101
        self._test_failed(test, err)
102

    
103
    def addFailure(self, test, err):  # noqa
104
        """Called when the test case test signals a failure"""
105
        super(BurninTestResult, self).addFailure(test, err)
106
        self._test_failed(test, err)
107

    
108

    
109
# --------------------------------------------------------------------
110
# Helper Classes
111
# pylint: disable=too-few-public-methods
112
# pylint: disable=too-many-instance-attributes
113
class Clients(object):
114
    """Our kamaki clients"""
115
    auth_url = None
116
    token = None
117
    # Astakos
118
    astakos = None
119
    retry = CONNECTION_RETRY_LIMIT
120
    # Compute
121
    compute = None
122
    compute_url = None
123
    # Cyclades
124
    cyclades = None
125
    # Network
126
    network = None
127
    network_url = None
128
    # Pithos
129
    pithos = None
130
    pithos_url = None
131
    # Image
132
    image = None
133
    image_url = None
134

    
135
    def initialize_clients(self):
136
        """Initialize all the Kamaki Clients"""
137
        self.astakos = AstakosClient(self.auth_url, self.token)
138
        self.astakos.CONNECTION_RETRY_LIMIT = self.retry
139

    
140
        endpoints = self.astakos.authenticate()
141

    
142
        self.compute_url = _get_endpoint_url(endpoints, "compute")
143
        self.compute = ComputeClient(self.compute_url, self.token)
144
        self.compute.CONNECTION_RETRY_LIMIT = self.retry
145

    
146
        self.cyclades = CycladesClient(self.compute_url, self.token)
147
        self.cyclades.CONNECTION_RETRY_LIMIT = self.retry
148

    
149
        self.network_url = _get_endpoint_url(endpoints, "network")
150
        self.network = CycladesNetworkClient(self.network_url, self.token)
151
        self.network.CONNECTION_RETRY_LIMIT = self.retry
152

    
153
        self.pithos_url = _get_endpoint_url(endpoints, "object-store")
154
        self.pithos = PithosClient(self.pithos_url, self.token)
155
        self.pithos.CONNECTION_RETRY_LIMIT = self.retry
156

    
157
        self.image_url = _get_endpoint_url(endpoints, "image")
158
        self.image = ImageClient(self.image_url, self.token)
159
        self.image.CONNECTION_RETRY_LIMIT = self.retry
160

    
161

    
162
def _get_endpoint_url(endpoints, endpoint_type):
163
    """Get the publicURL for the specified endpoint"""
164

    
165
    service_catalog = parse_endpoints(endpoints, ep_type=endpoint_type)
166
    return service_catalog[0]['endpoints'][0]['publicURL']
167

    
168

    
169
class Proper(object):
170
    """A descriptor used by tests implementing the TestCase class
171

172
    Since each instance of the TestCase will only be used to run a single
173
    test method (a new fixture is created for each test) the attributes can
174
    not be saved in the class instances. Instead we use descriptors.
175

176
    """
177
    def __init__(self, value=None):
178
        self.val = value
179

    
180
    def __get__(self, obj, objtype=None):
181
        return self.val
182

    
183
    def __set__(self, obj, value):
184
        self.val = value
185

    
186

    
187
# --------------------------------------------------------------------
188
# BurninTests class
189
# pylint: disable=too-many-public-methods
190
class BurninTests(unittest.TestCase):
191
    """Common class that all burnin tests should implement"""
192
    clients = Clients()
193
    run_id = None
194
    use_ipv6 = None
195
    action_timeout = None
196
    action_warning = None
197
    query_interval = None
198
    system_user = None
199
    images = None
200
    flavors = None
201
    delete_stale = False
202
    temp_directory = None
203
    failfast = None
204

    
205
    quotas = Proper(value=None)
206

    
207
    @classmethod
208
    def setUpClass(cls):  # noqa
209
        """Initialize BurninTests"""
210
        cls.suite_name = cls.__name__
211
        logger.testsuite_start(cls.suite_name)
212

    
213
        # Set test parameters
214
        cls.longMessage = True
215

    
216
    def test_000_clients_setup(self):
217
        """Initializing astakos/cyclades/pithos clients"""
218
        # Update class attributes
219
        self.clients.initialize_clients()
220
        self.info("Astakos auth url is %s", self.clients.auth_url)
221
        self.info("Cyclades url is %s", self.clients.compute_url)
222
        self.info("Network url is %s", self.clients.network_url)
223
        self.info("Pithos url is %s", self.clients.pithos_url)
224
        self.info("Image url is %s", self.clients.image_url)
225

    
226
        user_uuid = self._get_uuid()
227
        self.quotas = self._get_quotas()
228
        for puuid, quotas in self.quotas.items():
229
            project_name = self._get_project_name(puuid, user_uuid)
230
            self.info("  Project %s:", project_name)
231
            self.info("    Disk usage is         %s bytes",
232
                      quotas['cyclades.disk']['usage'])
233
            self.info("    VM usage is           %s",
234
                      quotas['cyclades.vm']['usage'])
235
            self.info("    DiskSpace usage is    %s bytes",
236
                      quotas['pithos.diskspace']['usage'])
237
            self.info("    Ram usage is          %s bytes",
238
                      quotas['cyclades.ram']['usage'])
239
            self.info("    Floating IPs usage is %s",
240
                      quotas['cyclades.floating_ip']['usage'])
241
            self.info("    CPU usage is          %s",
242
                      quotas['cyclades.cpu']['usage'])
243
            self.info("    Network usage is      %s",
244
                      quotas['cyclades.network.private']['usage'])
245

    
246
    def _run_tests(self, tcases):
247
        """Run some generated testcases"""
248
        global success  # pylint: disable=invalid-name, global-statement
249

    
250
        for tcase in tcases:
251
            self.info("Running testsuite %s", tcase.__name__)
252
            success = run_test(tcase) and success
253
            if self.failfast and not success:
254
                break
255

    
256
    # ----------------------------------
257
    # Loggers helper functions
258
    def log(self, msg, *args):
259
        """Pass the section value to logger"""
260
        logger.log(self.suite_name, msg, *args)
261

    
262
    def info(self, msg, *args):
263
        """Pass the section value to logger"""
264
        logger.info(self.suite_name, msg, *args)
265

    
266
    def debug(self, msg, *args):
267
        """Pass the section value to logger"""
268
        logger.debug(self.suite_name, msg, *args)
269

    
270
    def warning(self, msg, *args):
271
        """Pass the section value to logger"""
272
        logger.warning(self.suite_name, msg, *args)
273

    
274
    def error(self, msg, *args):
275
        """Pass the section value to logger"""
276
        logger.error(self.suite_name, msg, *args)
277

    
278
    # ----------------------------------
279
    # Helper functions that every testsuite may need
280
    def _get_uuid(self):
281
        """Get our uuid"""
282
        authenticate = self.clients.astakos.authenticate()
283
        uuid = authenticate['access']['user']['id']
284
        self.info("User's uuid is %s", uuid)
285
        return uuid
286

    
287
    def _get_username(self):
288
        """Get our User Name"""
289
        authenticate = self.clients.astakos.authenticate()
290
        username = authenticate['access']['user']['name']
291
        self.info("User's name is %s", username)
292
        return username
293

    
294
    def _create_tmp_directory(self):
295
        """Create a tmp directory"""
296
        temp_dir = tempfile.mkdtemp(dir=self.temp_directory)
297
        self.info("Temp directory %s created", temp_dir)
298
        return temp_dir
299

    
300
    def _remove_tmp_directory(self, tmp_dir):
301
        """Remove a tmp directory"""
302
        try:
303
            shutil.rmtree(tmp_dir)
304
            self.info("Temp directory %s deleted", tmp_dir)
305
        except OSError:
306
            pass
307

    
308
    def _get_uuid_of_system_user(self):
309
        """Get the uuid of the system user
310

311
        This is the user that upload the 'official' images.
312

313
        """
314
        self.info("Getting the uuid of the system user")
315
        system_users = None
316
        if self.system_user is not None:
317
            try:
318
                su_type, su_value = parse_typed_option(self.system_user)
319
                if su_type == "name":
320
                    system_users = [su_value]
321
                elif su_type == "id":
322
                    self.info("System user's uuid is %s", su_value)
323
                    return su_value
324
                else:
325
                    self.error("Unrecognized system-user type %s", su_type)
326
                    self.fail("Unrecognized system-user type")
327
            except ValueError:
328
                msg = "Invalid system-user format: %s. Must be [id|name]:.+"
329
                self.warning(msg, self.system_user)
330

    
331
        if system_users is None:
332
            system_users = SYSTEM_USERS
333

    
334
        uuids = self.clients.astakos.get_uuids(system_users)
335
        for su_name in system_users:
336
            self.info("Trying username %s", su_name)
337
            if su_name in uuids:
338
                self.info("System user's uuid is %s", uuids[su_name])
339
                return uuids[su_name]
340

    
341
        self.warning("No system user found")
342
        return None
343

    
344
    def _skip_if(self, condition, msg):
345
        """Skip tests"""
346
        if condition:
347
            self.info("Test skipped: %s" % msg)
348
            self.skipTest(msg)
349

    
350
    # ----------------------------------
351
    # Flavors
352
    def _get_list_of_flavors(self, detail=False):
353
        """Get (detailed) list of flavors"""
354
        if detail:
355
            self.info("Getting detailed list of flavors")
356
        else:
357
            self.info("Getting simple list of flavors")
358
        flavors = self.clients.compute.list_flavors(detail=detail)
359
        return flavors
360

    
361
    def _find_flavors(self, patterns, flavors=None):
362
        """Find a list of suitable flavors to use
363

364
        The patterns is a list of `typed_options'. A list of all flavors
365
        matching this patterns will be returned.
366

367
        """
368
        if flavors is None:
369
            flavors = self._get_list_of_flavors(detail=True)
370

    
371
        ret_flavors = []
372
        for ptrn in patterns:
373
            try:
374
                flv_type, flv_value = parse_typed_option(ptrn)
375
            except ValueError:
376
                msg = "Invalid flavor format: %s. Must be [id|name]:.+"
377
                self.warning(msg, ptrn)
378
                continue
379

    
380
            if flv_type == "name":
381
                # Filter flavor by name
382
                msg = "Trying to find a flavor with name %s"
383
                self.info(msg, flv_value)
384
                filtered_flvs = \
385
                    [f for f in flavors if
386
                     re.search(flv_value, f['name'], flags=re.I) is not None]
387
            elif flv_type == "id":
388
                # Filter flavors by id
389
                msg = "Trying to find a flavor with id %s"
390
                self.info(msg, flv_value)
391
                filtered_flvs = \
392
                    [f for f in flavors if str(f['id']) == flv_value]
393
            else:
394
                self.error("Unrecognized flavor type %s", flv_type)
395
                self.fail("Unrecognized flavor type")
396

    
397
            # Append and continue
398
            ret_flavors.extend(filtered_flvs)
399

    
400
        self.assertGreater(len(ret_flavors), 0,
401
                           "No matching flavors found")
402
        return ret_flavors
403

    
404
    # ----------------------------------
405
    # Images
406
    def _get_list_of_images(self, detail=False):
407
        """Get (detailed) list of images"""
408
        if detail:
409
            self.info("Getting detailed list of images")
410
        else:
411
            self.info("Getting simple list of images")
412
        images = self.clients.image.list_public(detail=detail)
413
        # Remove images registered by burnin
414
        images = [img for img in images
415
                  if not img['name'].startswith(SNF_TEST_PREFIX)]
416
        return images
417

    
418
    def _get_list_of_sys_images(self, images=None):
419
        """Get (detailed) list of images registered by system user or by me"""
420
        self.info("Getting list of images registered by system user or by me")
421
        if images is None:
422
            images = self._get_list_of_images(detail=True)
423

    
424
        su_uuid = self._get_uuid_of_system_user()
425
        my_uuid = self._get_uuid()
426
        ret_images = [i for i in images
427
                      if i['owner'] == su_uuid or i['owner'] == my_uuid]
428

    
429
        return ret_images
430

    
431
    def _find_images(self, patterns, images=None):
432
        """Find a list of suitable images to use
433

434
        The patterns is a list of `typed_options'. A list of all images
435
        matching this patterns will be returned.
436

437
        """
438
        if images is None:
439
            images = self._get_list_of_sys_images()
440

    
441
        ret_images = []
442
        for ptrn in patterns:
443
            try:
444
                img_type, img_value = parse_typed_option(ptrn)
445
            except ValueError:
446
                msg = "Invalid image format: %s. Must be [id|name]:.+"
447
                self.warning(msg, ptrn)
448
                continue
449

    
450
            if img_type == "name":
451
                # Filter image by name
452
                msg = "Trying to find an image with name %s"
453
                self.info(msg, img_value)
454
                filtered_imgs = \
455
                    [i for i in images if
456
                     re.search(img_value, i['name'], flags=re.I) is not None]
457
            elif img_type == "id":
458
                # Filter images by id
459
                msg = "Trying to find an image with id %s"
460
                self.info(msg, img_value)
461
                filtered_imgs = \
462
                    [i for i in images if
463
                     i['id'].lower() == img_value.lower()]
464
            else:
465
                self.error("Unrecognized image type %s", img_type)
466
                self.fail("Unrecognized image type")
467

    
468
            # Append and continue
469
            ret_images.extend(filtered_imgs)
470

    
471
        self.assertGreater(len(ret_images), 0,
472
                           "No matching images found")
473
        return ret_images
474

    
475
    # ----------------------------------
476
    # Pithos
477
    def _set_pithos_account(self, account):
478
        """Set the Pithos account"""
479
        assert account, "No pithos account was given"
480

    
481
        self.info("Setting Pithos account to %s", account)
482
        self.clients.pithos.account = account
483

    
484
    def _set_pithos_container(self, container):
485
        """Set the Pithos container"""
486
        assert container, "No pithos container was given"
487

    
488
        self.info("Setting Pithos container to %s", container)
489
        self.clients.pithos.container = container
490

    
491
    def _get_list_of_containers(self, account=None):
492
        """Get list of containers"""
493
        if account is not None:
494
            self._set_pithos_account(account)
495
        self.info("Getting list of containers")
496
        return self.clients.pithos.list_containers()
497

    
498
    def _create_pithos_container(self, container):
499
        """Create a pithos container
500

501
        If the container exists, nothing will happen
502

503
        """
504
        assert container, "No pithos container was given"
505

    
506
        self.info("Creating pithos container %s", container)
507
        self.clients.pithos.container = container
508
        self.clients.pithos.container_put()
509

    
510
    # ----------------------------------
511
    # Quotas
512
    def _get_quotas(self):
513
        """Get quotas"""
514
        self.info("Getting quotas")
515
        return dict(self.clients.astakos.get_quotas())
516

    
517
    # pylint: disable=invalid-name
518
    # pylint: disable=too-many-arguments
519
    def _check_quotas(self, puuid=None, disk=None, vm=None, diskspace=None,
520
                      ram=None, ip=None, cpu=None, network=None):
521
        """Check that quotas' changes are consistent
522

523
        @param puuid: The uuid of the project, quotas are assigned to
524

525
        """
526

    
527
        assert any(v is None for v in
528
                   [disk, vm, diskspace, ram, ip, cpu, network]), \
529
            "_check_quotas require arguments"
530

    
531
        self.info("Check that quotas' changes are consistent")
532
        old_quotas = self.quotas
533
        new_quotas = self._get_quotas()
534
        self.quotas = new_quotas
535

    
536
        user_uuid = self._get_uuid()
537
        if puuid is None:
538
            puuid = user_uuid
539

    
540
        self.assertListEqual(sorted(old_quotas.keys()),
541
                             sorted(new_quotas.keys()))
542
        for project in old_quotas.keys():
543
            # Check Disk usage
544
            project_name = self._get_project_name(project, user_uuid)
545
            self._check_quotas_aux(old_quotas[project], new_quotas[project],
546
                                   project_name, "cyclades.disk",
547
                                   disk, project == puuid)
548
            # Check VM usage
549
            self._check_quotas_aux(old_quotas[project], new_quotas[project],
550
                                   project_name, "cyclades.vm",
551
                                   vm, project == puuid)
552
            # Check DiskSpace usage
553
            self._check_quotas_aux(old_quotas[project], new_quotas[project],
554
                                   project_name, "pithos.diskspace",
555
                                   diskspace, project == puuid)
556
            # Check Ram usage
557
            self._check_quotas_aux(old_quotas[project], new_quotas[project],
558
                                   project_name, "cyclades.ram",
559
                                   ram, project == puuid)
560
            # Check Floating IPs usage
561
            self._check_quotas_aux(old_quotas[project], new_quotas[project],
562
                                   project_name, "cyclades.floating_ip",
563
                                   ip, project == puuid)
564
            # Check CPU usage
565
            self._check_quotas_aux(old_quotas[project], new_quotas[project],
566
                                   project_name, "cyclades.cpu",
567
                                   cpu, project == puuid)
568
            # Check Network usage
569
            self._check_quotas_aux(old_quotas[project], new_quotas[project],
570
                                   project_name, "cyclades.network.private",
571
                                   network, project == puuid)
572

    
573
    def _check_quotas_aux(self, old_quotas, new_quotas,
574
                          project_name, resource, value, check):
575
        """Auxiliary function for _check_quotas"""
576
        old_value = old_quotas[resource]['usage']
577
        new_value = new_quotas[resource]['usage']
578
        if check and value is not None:
579
            assert isinstance(value, int), \
580
                "%s value has to be integer" % resource
581
            old_value += value
582
        self.assertEqual(old_value, new_value,
583
                         "Project %s: %s quotas don't match" %
584
                         (project_name, resource))
585

    
586
    # ----------------------------------
587
    # Projects
588
    def _get_project_name(self, puuid, uuid=None):
589
        """Get the name of a project"""
590
        if uuid is None:
591
            uuid = self._get_uuid()
592
        if puuid == uuid:
593
            return "base"
594
        else:
595
            project_info = self.clients.astakos.get_project(puuid)
596
            return project_info['name']
597

    
598

    
599
# --------------------------------------------------------------------
600
# Initialize Burnin
601
def initialize(opts, testsuites, stale_testsuites):
602
    """Initalize burnin
603

604
    Initialize our logger and burnin state
605

606
    """
607
    # Initialize logger
608
    global logger  # pylint: disable=invalid-name, global-statement
609
    curr_time = datetime.datetime.now()
610
    logger = Log(opts.log_folder, verbose=opts.verbose,
611
                 use_colors=opts.use_colors, in_parallel=False,
612
                 log_level=opts.log_level, curr_time=curr_time)
613

    
614
    # Initialize clients
615
    Clients.auth_url = opts.auth_url
616
    Clients.token = opts.token
617

    
618
    # Pass the rest options to BurninTests
619
    BurninTests.use_ipv6 = opts.use_ipv6
620
    BurninTests.action_timeout = opts.action_timeout
621
    BurninTests.action_warning = opts.action_warning
622
    BurninTests.query_interval = opts.query_interval
623
    BurninTests.system_user = opts.system_user
624
    BurninTests.flavors = opts.flavors
625
    BurninTests.images = opts.images
626
    BurninTests.delete_stale = opts.delete_stale
627
    BurninTests.temp_directory = opts.temp_directory
628
    BurninTests.failfast = opts.failfast
629
    BurninTests.run_id = SNF_TEST_PREFIX + \
630
        datetime.datetime.strftime(curr_time, "%Y%m%d%H%M%S")
631

    
632
    # Choose tests to run
633
    if opts.show_stale:
634
        # We will run the stale_testsuites
635
        return (stale_testsuites, True)
636

    
637
    if opts.tests != "all":
638
        testsuites = opts.tests
639
    if opts.exclude_tests is not None:
640
        testsuites = [tsuite for tsuite in testsuites
641
                      if tsuite not in opts.exclude_tests]
642

    
643
    return (testsuites, opts.failfast)
644

    
645

    
646
# --------------------------------------------------------------------
647
# Run Burnin
648
def run_burnin(testsuites, failfast=False):
649
    """Run burnin testsuites"""
650
    # pylint: disable=invalid-name,global-statement
651
    # pylint: disable=global-variable-not-assigned
652
    global logger, success
653

    
654
    success = True
655
    run_tests(testsuites, failfast=failfast)
656

    
657
    # Clean up our logger
658
    del logger
659

    
660
    # Return
661
    return 0 if success else 1
662

    
663

    
664
def run_tests(tcases, failfast=False):
665
    """Run some testcases"""
666
    # pylint: disable=invalid-name,global-statement
667
    # pylint: disable=global-variable-not-assigned
668
    global success
669

    
670
    for tcase in tcases:
671
        was_success = run_test(tcase)
672
        success = success and was_success
673
        if failfast and not success:
674
            break
675

    
676

    
677
def run_test(tcase):
678
    """Run a testcase"""
679
    tsuite = unittest.TestLoader().loadTestsFromTestCase(tcase)
680
    results = tsuite.run(BurninTestResult())
681

    
682
    return was_successful(tcase.__name__, results.wasSuccessful())
683

    
684

    
685
# --------------------------------------------------------------------
686
# Helper functions
687
def was_successful(tsuite, successful):
688
    """Handle whether a testsuite was succesful or not"""
689
    if successful:
690
        logger.testsuite_success(tsuite)
691
        return True
692
    else:
693
        logger.testsuite_failure(tsuite)
694
        return False
695

    
696

    
697
def parse_typed_option(value):
698
    """Parse typed options (flavors and images)
699

700
    The options are in the form 'id:123-345' or 'name:^Debian Base$'
701

702
    """
703
    try:
704
        [type_, val] = value.strip().split(':')
705
        if type_ not in ["id", "name"]:
706
            raise ValueError
707
        return type_, val
708
    except ValueError:
709
        raise