common.py 19.2 KB
Newer Older
Ilias Tsitsimpis's avatar
Ilias Tsitsimpis committed
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
# 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.

"""
Common utils for burnin tests

"""

Ilias Tsitsimpis's avatar
Ilias Tsitsimpis committed
39
40
41
import os
import re
import shutil
42
import unittest
Ilias Tsitsimpis's avatar
Ilias Tsitsimpis committed
43
import datetime
Ilias Tsitsimpis's avatar
Ilias Tsitsimpis committed
44
import tempfile
Ilias Tsitsimpis's avatar
Ilias Tsitsimpis committed
45
46
import traceback

47
from kamaki.clients.cyclades import CycladesClient
Ilias Tsitsimpis's avatar
Ilias Tsitsimpis committed
48
49
from kamaki.clients.astakos import AstakosClient
from kamaki.clients.compute import ComputeClient
Ilias Tsitsimpis's avatar
Ilias Tsitsimpis committed
50
from kamaki.clients.pithos import PithosClient
Ilias Tsitsimpis's avatar
Ilias Tsitsimpis committed
51
from kamaki.clients.image import ImageClient
Ilias Tsitsimpis's avatar
Ilias Tsitsimpis committed
52

Ilias Tsitsimpis's avatar
Ilias Tsitsimpis committed
53
from synnefo_tools.burnin.logger import Log
Ilias Tsitsimpis's avatar
Ilias Tsitsimpis committed
54
55
56
57
58


# --------------------------------------------------------------------
# Global variables
logger = None  # Invalid constant name. pylint: disable-msg=C0103
Ilias Tsitsimpis's avatar
Ilias Tsitsimpis committed
59
SNF_TEST_PREFIX = "snf-test-"
Ilias Tsitsimpis's avatar
Ilias Tsitsimpis committed
60
CONNECTION_RETRY_LIMIT = 2
Ilias Tsitsimpis's avatar
Ilias Tsitsimpis committed
61
SYSTEM_USERS = ["images@okeanos.grnet.gr", "images@demo.synnefo.org"]
Ilias Tsitsimpis's avatar
Ilias Tsitsimpis committed
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76


# --------------------------------------------------------------------
# BurninTestResult class
class BurninTestResult(unittest.TestResult):
    """Modify the TextTestResult class"""
    def __init__(self):
        super(BurninTestResult, self).__init__()

        # Test parameters
        self.failfast = True

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

    # Method could be a function. pylint: disable-msg=R0201
    def _test_failed(self, test, err):
        """Test failed"""
Ilias Tsitsimpis's avatar
Ilias Tsitsimpis committed
82
83
84
85
86
        # Get class name
        if test.__class__.__name__ == "_ErrorHolder":
            class_name = test.id().split('.')[-1].rstrip(')')
        else:
            class_name = test.__class__.__name__
87
        err_msg = str(test) + "... failed (%s)."
Ilias Tsitsimpis's avatar
Ilias Tsitsimpis committed
88
89
        timestamp = datetime.datetime.strftime(
            datetime.datetime.now(), "%a %b %d %Y %H:%M:%S")
Ilias Tsitsimpis's avatar
Ilias Tsitsimpis committed
90
        logger.error(class_name, err_msg, timestamp)
Ilias Tsitsimpis's avatar
Ilias Tsitsimpis committed
91
92
        (err_type, err_value, err_trace) = err
        trcback = traceback.format_exception(err_type, err_value, err_trace)
Ilias Tsitsimpis's avatar
Ilias Tsitsimpis committed
93
        logger.info(class_name, trcback)
Ilias Tsitsimpis's avatar
Ilias Tsitsimpis committed
94
95
96
97
98
99
100
101
102
103
104
105
106
107

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

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


# --------------------------------------------------------------------
# BurninTests class
108
109
# Too few public methods. pylint: disable-msg=R0903
# Too many instance attributes. pylint: disable-msg=R0902
Ilias Tsitsimpis's avatar
Ilias Tsitsimpis committed
110
111
112
113
class Clients(object):
    """Our kamaki clients"""
    auth_url = None
    token = None
Ilias Tsitsimpis's avatar
Ilias Tsitsimpis committed
114
    # Astakos
Ilias Tsitsimpis's avatar
Ilias Tsitsimpis committed
115
116
    astakos = None
    retry = CONNECTION_RETRY_LIMIT
Ilias Tsitsimpis's avatar
Ilias Tsitsimpis committed
117
    # Compute
Ilias Tsitsimpis's avatar
Ilias Tsitsimpis committed
118
119
    compute = None
    compute_url = None
Ilias Tsitsimpis's avatar
Ilias Tsitsimpis committed
120
121
122
123
124
    # Cyclades
    cyclades = None
    # Pithos
    pithos = None
    pithos_url = None
Ilias Tsitsimpis's avatar
Ilias Tsitsimpis committed
125
126
127
    # Image
    image = None
    image_url = None
Ilias Tsitsimpis's avatar
Ilias Tsitsimpis committed
128
129


Ilias Tsitsimpis's avatar
Ilias Tsitsimpis committed
130
131
132
# Too many public methods (45/20). pylint: disable-msg=R0904
class BurninTests(unittest.TestCase):
    """Common class that all burnin tests should implement"""
Ilias Tsitsimpis's avatar
Ilias Tsitsimpis committed
133
    clients = Clients()
Ilias Tsitsimpis's avatar
Ilias Tsitsimpis committed
134
    run_id = None
Ilias Tsitsimpis's avatar
Ilias Tsitsimpis committed
135
136
137
138
    use_ipv6 = None
    action_timeout = None
    action_warning = None
    query_interval = None
Ilias Tsitsimpis's avatar
Ilias Tsitsimpis committed
139
    system_user = None
140
141
    images = None
    flavors = None
142
    delete_stale = False
Ilias Tsitsimpis's avatar
Ilias Tsitsimpis committed
143

Ilias Tsitsimpis's avatar
Ilias Tsitsimpis committed
144
145
146
147
148
149
150
151
152
    @classmethod
    def setUpClass(cls):  # noqa
        """Initialize BurninTests"""
        cls.suite_name = cls.__name__
        logger.testsuite_start(cls.suite_name)

        # Set test parameters
        cls.longMessage = True

153
    def test_000_clients_setup(self):
Ilias Tsitsimpis's avatar
Ilias Tsitsimpis committed
154
155
        """Initializing astakos/cyclades/pithos clients"""
        # Update class attributes
Ilias Tsitsimpis's avatar
Ilias Tsitsimpis committed
156
157
158
159
160
161
162
163
164
165
166
        self.info("Astakos auth url is %s", self.clients.auth_url)
        self.clients.astakos = AstakosClient(
            self.clients.auth_url, self.clients.token)
        self.clients.astakos.CONNECTION_RETRY_LIMIT = self.clients.retry

        self.clients.compute_url = \
            self.clients.astakos.get_service_endpoints('compute')['publicURL']
        self.info("Cyclades url is %s", self.clients.compute_url)
        self.clients.compute = ComputeClient(
            self.clients.compute_url, self.clients.token)
        self.clients.compute.CONNECTION_RETRY_LIMIT = self.clients.retry
Ilias Tsitsimpis's avatar
Ilias Tsitsimpis committed
167

168
169
170
171
        self.clients.cyclades = CycladesClient(
            self.clients.compute_url, self.clients.token)
        self.clients.cyclades.CONNECTION_RETRY_LIMIT = self.clients.retry

Ilias Tsitsimpis's avatar
Ilias Tsitsimpis committed
172
173
174
175
176
177
178
        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

Ilias Tsitsimpis's avatar
Ilias Tsitsimpis committed
179
180
181
182
183
184
185
        self.clients.image_url = \
            self.clients.astakos.get_service_endpoints('image')['publicURL']
        self.info("Image url is %s", self.clients.image_url)
        self.clients.image = ImageClient(
            self.clients.image_url, self.clients.token)
        self.clients.image.CONNECTION_RETRY_LIMIT = self.clients.retry

Ilias Tsitsimpis's avatar
Ilias Tsitsimpis committed
186
187
    # ----------------------------------
    # Loggers helper functions
Ilias Tsitsimpis's avatar
Ilias Tsitsimpis committed
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
    def log(self, msg, *args):
        """Pass the section value to logger"""
        logger.log(self.suite_name, msg, *args)

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

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

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

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

Ilias Tsitsimpis's avatar
Ilias Tsitsimpis committed
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
    # ----------------------------------
    # Helper functions that every testsuite may need
    def _get_uuid(self):
        """Get our uuid"""
        authenticate = self.clients.astakos.authenticate()
        uuid = authenticate['access']['user']['id']
        self.info("User's uuid is %s", uuid)
        return uuid

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

Ilias Tsitsimpis's avatar
Ilias Tsitsimpis committed
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
    def _create_tmp_directory(self):
        """Create a tmp directory

        In my machine /tmp has not enough space for an image
        to be saves, so we are going to use the current directory.

        """
        temp_dir = tempfile.mkdtemp(dir=os.getcwd())
        self.info("Temp directory %s created", temp_dir)
        return temp_dir

    def _remove_tmp_directory(self, tmp_dir):
        """Remove a tmp directory"""
        try:
            shutil.rmtree(tmp_dir)
            self.info("Temp directory %s deleted", tmp_dir)
        except OSError:
            pass

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

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

        """
        self.info("Getting the uuid of the system user")
        system_users = None
        if self.system_user is not None:
            parsed_su = parse_typed_option(self.system_user)
            if parsed_su is None:
                msg = "Invalid system-user format: %s. Must be [id|name]:.+"
                self.warning(msg, self.system_user)
            else:
                su_type, su_value = parsed_su
                if su_type == "name":
                    system_users = [su_value]
                elif su_type == "id":
                    self.info("System user's uuid is %s", su_value)
                    return su_value
                else:
                    self.error("Unrecognized system-user type %s", su_type)
                    self.fail("Unrecognized system-user type")

        if system_users is None:
            system_users = SYSTEM_USERS

        uuids = self.clients.astakos.usernames2uuids(system_users)
        for su_name in system_users:
            self.info("Trying username %s", su_name)
            if su_name in uuids:
                self.info("System user's uuid is %s", uuids[su_name])
                return uuids[su_name]

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

280
281
282
283
284
285
    def _skip_if(self, condition, msg):
        """Skip tests"""
        if condition:
            self.info("Test skipped: %s" % msg)
            self.skipTest(msg)

Ilias Tsitsimpis's avatar
Ilias Tsitsimpis committed
286
287
    # ----------------------------------
    # Flavors
288
289
290
291
292
293
294
295
296
    def _get_list_of_flavors(self, detail=False):
        """Get (detailed) list of flavors"""
        if detail:
            self.info("Getting detailed list of flavors")
        else:
            self.info("Getting simple list of flavors")
        flavors = self.clients.compute.list_flavors(detail=detail)
        return flavors

297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
    def _find_flavors(self, patterns, flavors=None):
        """Find a list of suitable flavors to use

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

        """
        if flavors is None:
            flavors = self._get_list_of_flavors(detail=True)

        ret_flavors = []
        for ptrn in patterns:
            parsed_ptrn = parse_typed_option(ptrn)
            if parsed_ptrn is None:
                msg = "Invalid flavor format: %s. Must be [id|name]:.+"
                self.warning(msg, ptrn)
                continue
            flv_type, flv_value = parsed_ptrn
            if flv_type == "name":
                # Filter flavor by name
                msg = "Trying to find a flavor with name %s"
                self.info(msg, flv_value)
                filtered_flvs = \
                    [f for f in flavors if
                     re.search(flv_value, f['name'], flags=re.I) is not None]
            elif flv_type == "id":
                # Filter flavors by id
                msg = "Trying to find a flavor with id %s"
                self.info(msg, flv_value)
                filtered_flvs = \
                    [f for f in flavors if str(f['id']) == flv_value]
            else:
                self.error("Unrecognized flavor type %s", flv_type)
                self.fail("Unrecognized flavor type")

            # Append and continue
            ret_flavors.extend(filtered_flvs)

        self.assertGreater(len(ret_flavors), 0,
                           "No matching flavors found")
        return ret_flavors

Ilias Tsitsimpis's avatar
Ilias Tsitsimpis committed
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
    # ----------------------------------
    # Images
    def _get_list_of_images(self, detail=False):
        """Get (detailed) list of images"""
        if detail:
            self.info("Getting detailed list of images")
        else:
            self.info("Getting simple list of images")
        images = self.clients.image.list_public(detail=detail)
        # Remove images registered by burnin
        images = [img for img in images
                  if not img['name'].startswith(SNF_TEST_PREFIX)]
        return images

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

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

        return ret_images

366
367
    def _find_images(self, patterns, images=None):
        """Find a list of suitable images to use
Ilias Tsitsimpis's avatar
Ilias Tsitsimpis committed
368

369
370
        The patterns is a list of `typed_options'. A list of all images
        matching this patterns will be returned.
Ilias Tsitsimpis's avatar
Ilias Tsitsimpis committed
371
372
373
374
375

        """
        if images is None:
            images = self._get_list_of_sys_images()

376
        ret_images = []
Ilias Tsitsimpis's avatar
Ilias Tsitsimpis committed
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
        for ptrn in patterns:
            parsed_ptrn = parse_typed_option(ptrn)
            if parsed_ptrn is None:
                msg = "Invalid image format: %s. Must be [id|name]:.+"
                self.warning(msg, ptrn)
                continue
            img_type, img_value = parsed_ptrn
            if img_type == "name":
                # Filter image by name
                msg = "Trying to find an image with name %s"
                self.info(msg, img_value)
                filtered_imgs = \
                    [i for i in images if
                     re.search(img_value, i['name'], flags=re.I) is not None]
            elif img_type == "id":
                # Filter images by id
                msg = "Trying to find an image with id %s"
                self.info(msg, img_value)
                filtered_imgs = \
                    [i for i in images if
                     i['id'].lower() == img_value.lower()]
            else:
                self.error("Unrecognized image type %s", img_type)
                self.fail("Unrecognized image type")

402
403
            # Append and continue
            ret_images.extend(filtered_imgs)
Ilias Tsitsimpis's avatar
Ilias Tsitsimpis committed
404

405
406
407
        self.assertGreater(len(ret_images), 0,
                           "No matching images found")
        return ret_images
Ilias Tsitsimpis's avatar
Ilias Tsitsimpis committed
408
409
410

    # ----------------------------------
    # Pithos
Ilias Tsitsimpis's avatar
Ilias Tsitsimpis committed
411
    def _set_pithos_account(self, account):
Ilias Tsitsimpis's avatar
Ilias Tsitsimpis committed
412
        """Set the Pithos account"""
Ilias Tsitsimpis's avatar
Ilias Tsitsimpis committed
413
414
        assert account, "No pithos account was given"

Ilias Tsitsimpis's avatar
Ilias Tsitsimpis committed
415
        self.info("Setting Pithos account to %s", account)
Ilias Tsitsimpis's avatar
Ilias Tsitsimpis committed
416
417
        self.clients.pithos.account = account

Ilias Tsitsimpis's avatar
Ilias Tsitsimpis committed
418
419
420
421
422
423
424
    def _set_pithos_container(self, container):
        """Set the Pithos container"""
        assert container, "No pithos container was given"

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

Ilias Tsitsimpis's avatar
Ilias Tsitsimpis committed
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
    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()

Ilias Tsitsimpis's avatar
Ilias Tsitsimpis committed
444

Ilias Tsitsimpis's avatar
Ilias Tsitsimpis committed
445
446
# --------------------------------------------------------------------
# Initialize Burnin
447
def initialize(opts, testsuites, stale_testsuites):
Ilias Tsitsimpis's avatar
Ilias Tsitsimpis committed
448
449
450
451
452
453
454
    """Initalize burnin

    Initialize our logger and burnin state

    """
    # Initialize logger
    global logger  # Using global statement. pylint: disable-msg=C0103,W0603
455
    curr_time = datetime.datetime.now()
Ilias Tsitsimpis's avatar
Ilias Tsitsimpis committed
456
    logger = Log(opts.log_folder, verbose=opts.verbose,
457
                 use_colors=opts.use_colors, in_parallel=False,
458
                 quiet=opts.quiet, curr_time=curr_time)
Ilias Tsitsimpis's avatar
Ilias Tsitsimpis committed
459
460
461
462
463
464

    # Initialize clients
    Clients.auth_url = opts.auth_url
    Clients.token = opts.token

    # Pass the rest options to BurninTests
Ilias Tsitsimpis's avatar
Ilias Tsitsimpis committed
465
466
467
468
    BurninTests.use_ipv6 = opts.use_ipv6
    BurninTests.action_timeout = opts.action_timeout
    BurninTests.action_warning = opts.action_warning
    BurninTests.query_interval = opts.query_interval
Ilias Tsitsimpis's avatar
Ilias Tsitsimpis committed
469
    BurninTests.system_user = opts.system_user
470
471
    BurninTests.flavors = opts.flavors
    BurninTests.images = opts.images
472
    BurninTests.delete_stale = opts.delete_stale
Ilias Tsitsimpis's avatar
Ilias Tsitsimpis committed
473
    BurninTests.run_id = SNF_TEST_PREFIX + \
474
        datetime.datetime.strftime(curr_time, "%Y%m%d%H%M%S")
Ilias Tsitsimpis's avatar
Ilias Tsitsimpis committed
475
476

    # Choose tests to run
477
478
    if opts.show_stale:
        # We will run the stale_testsuites
479
        return (stale_testsuites, True)
480

Ilias Tsitsimpis's avatar
Ilias Tsitsimpis committed
481
482
483
484
485
486
    if opts.tests != "all":
        testsuites = opts.tests
    if opts.exclude_tests is not None:
        testsuites = [tsuite for tsuite in testsuites
                      if tsuite not in opts.exclude_tests]

487
    return (testsuites, opts.failfast)
Ilias Tsitsimpis's avatar
Ilias Tsitsimpis committed
488
489
490
491


# --------------------------------------------------------------------
# Run Burnin
492
def run_burnin(testsuites, failfast=False, final_report=False):
Ilias Tsitsimpis's avatar
Ilias Tsitsimpis committed
493
494
495
496
497
    """Run burnin testsuites"""
    global logger  # Using global. pylint: disable-msg=C0103,W0603,W0602

    success = True
    for tcase in testsuites:
498
        was_success = run_test(tcase)
499
        success = success and was_success
500
501
        if failfast and not success:
            break
Ilias Tsitsimpis's avatar
Ilias Tsitsimpis committed
502

503
504
505
    # Are we going to print final report?
    if final_report:
        logger.print_logfile_to_stdout()
Ilias Tsitsimpis's avatar
Ilias Tsitsimpis committed
506
507
508
509
510
511
512
    # Clean up our logger
    del(logger)

    # Return
    return 0 if success else 1


513
514
515
516
517
518
519
520
def run_test(tcase):
    """Run a testcase"""
    tsuite = unittest.TestLoader().loadTestsFromTestCase(tcase)
    results = tsuite.run(BurninTestResult())

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


Ilias Tsitsimpis's avatar
Ilias Tsitsimpis committed
521
522
# --------------------------------------------------------------------
# Helper functions
Ilias Tsitsimpis's avatar
Ilias Tsitsimpis committed
523
def was_successful(tsuite, success):
Ilias Tsitsimpis's avatar
Ilias Tsitsimpis committed
524
525
526
    """Handle whether a testsuite was succesful or not"""
    if success:
        logger.testsuite_success(tsuite)
Ilias Tsitsimpis's avatar
Ilias Tsitsimpis committed
527
        return True
Ilias Tsitsimpis's avatar
Ilias Tsitsimpis committed
528
529
    else:
        logger.testsuite_failure(tsuite)
Ilias Tsitsimpis's avatar
Ilias Tsitsimpis committed
530
        return False
Ilias Tsitsimpis's avatar
Ilias Tsitsimpis committed
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545


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

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

    """
    try:
        [type_, val] = value.strip().split(':')
        if type_ not in ["id", "name"]:
            raise ValueError
        return type_, val
    except ValueError:
        return None
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563


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

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

    """
    def __init__(self, value=None):
        self.val = value

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

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