summaryrefslogtreecommitdiffstats
path: root/tests/testValidation.py
blob: b46c9f80b76f5a63e250a0ac903543854498b247 (plain)
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
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
#
# Copyright 2014 Red Hat, Inc.
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation; either version 2 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program; if not, write to the Free Software
# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301 USA
#
# Refer to the README and COPYING files for full details of the license
#

## This framework is mostly copied from vdsm test framework

import os
from nose.plugins.skip import SkipTest
from functools import wraps
from nose.plugins import Plugin
import subprocess


class SlowTestsPlugin(Plugin):
    """Skips tests that might be too slow to be run for quick iteration
    builds"""
    name = 'slowtests'
    enabled = False

    def add_options(self, parser, env=os.environ):
        env_opt = 'NOSE_SKIP_SLOW_TESTS'
        if env is None:
            default = False
        else:
            default = env.get(env_opt)

        parser.add_option('--without-slow-tests',
                          action='store_true',
                          default=default,
                          dest='disable_slow_tests',
                          help='Some tests might take a long time to run, ' +
                               'use this to skip slow tests automatically.' +
                               '  [%s]' % env_opt)

    def configure(self, options, conf):
        Plugin.configure(self, options, conf)
        if options.disable_slow_tests:
            SlowTestsPlugin.enabled = True


class StressTestsPlugin(Plugin):
    """
    Denotes a test which stresses the resources of the system under test. Such
    tests should probably not be run in parallel.  This plugin provides a
    mechanism for parallel testing applications to skip stress tests.
    """
    name = 'nonparalleltests'
    enabled = False

    def add_options(self, parser, env=os.environ):
        env_opt = 'NOSE_SKIP_STRESS_TESTS'
        if env is None:
            default = False
        else:
            default = env.get(env_opt)

        parser.add_option('--without-stress-tests',
                          action='store_true',
                          default=default,
                          dest='disable_stress_tests',
                          help='Some tests stress the resources of the ' +
                               'system under test.  Use this option to skip' +
                               'these tests (eg. when doing parallel' +
                               'testing [%s]' % env_opt)

    def configure(self, options, conf):
        Plugin.configure(self, options, conf)
        if options.disable_stress_tests:
            StressTestsPlugin.enabled = True


def ValidateRunningAsRoot(f):
    @wraps(f)
    def wrapper(*args, **kwargs):
        if os.geteuid() != 0:
            raise SkipTest("This test must be run as root")

        return f(*args, **kwargs)

    return wrapper


def slowtest(f):
    @wraps(f)
    def wrapper(*args, **kwargs):
        if SlowTestsPlugin.enabled:
            raise SkipTest("Slow tests have been disabled")

        return f(*args, **kwargs)

    return wrapper


def brokentest(msg="Test failed but it is known to be broken"):
    def wrap(f):
        @wraps(f)
        def wrapper(*args, **kwargs):
            try:
                return f(*args, **kwargs)
            except:
                raise SkipTest(msg)
        return wrapper

    return wrap


def stresstest(f):
    @wraps(f)
    def wrapper(*args, **kwargs):
        if StressTestsPlugin.enabled:
            raise SkipTest("Stress tests have been disabled")

        return f(*args, **kwargs)

    return wrapper


def checkSudo(cmd):
    p = subprocess.Popen(['sudo', '-l', '-n'] + cmd,
                         stdin=subprocess.PIPE, stdout=subprocess.PIPE,
                         stderr=subprocess.PIPE)
    out, err = p.communicate()

    if p.returncode != 0:
        raise SkipTest("Test requires SUDO configuration (%s)" % err.strip())