← Back to branch summary

~ubuntu-branches/ubuntu/trusty/cinder/trusty

« back to all changes in this revision

Viewing changes to cinder/tests/test_misc.py

  • Committer: Package Import Robot
  • Author(s): Chuck Short
  • Date: 2012-05-22 09:57:46 UTC
  • Revision ID: package-import@ubuntu.com-20120522095746-9lm71yvzltjybk4b
Tags: upstream-2012.2~f1~20120503.2
Import upstream version 2012.2~f1~20120503.2

Show diffs side-by-side

added added

removed removed

Lines of Context:
cinder/tests/test_misc.py
 
1
# vim: tabstop=4 shiftwidth=4 softtabstop=4
 
2
 
 
3
#    Copyright 2010 OpenStack LLC
 
4
#
 
5
#    Licensed under the Apache License, Version 2.0 (the "License"); you may
 
6
#    not use this file except in compliance with the License. You may obtain
 
7
#    a copy of the License at
 
8
#
 
9
#         http://www.apache.org/licenses/LICENSE-2.0
 
10
#
 
11
#    Unless required by applicable law or agreed to in writing, software
 
12
#    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
 
13
#    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
 
14
#    License for the specific language governing permissions and limitations
 
15
#    under the License.
 
16
 
 
17
import commands
 
18
import errno
 
19
import glob
 
20
import os
 
21
import select
 
22
 
 
23
from eventlet import greenpool
 
24
from eventlet import greenthread
 
25
import lockfile
 
26
 
 
27
from cinder import exception
 
28
from cinder import test
 
29
from cinder import utils
 
30
 
 
31
 
 
32
class ExceptionTestCase(test.TestCase):
 
33
    @staticmethod
 
34
    def _raise_exc(exc):
 
35
        raise exc()
 
36
 
 
37
    def test_exceptions_raise(self):
 
38
        for name in dir(exception):
 
39
            exc = getattr(exception, name)
 
40
            if isinstance(exc, type):
 
41
                self.assertRaises(exc, self._raise_exc, exc)
 
42
 
 
43
 
 
44
class ProjectTestCase(test.TestCase):
 
45
    def test_authors_up_to_date(self):
 
46
        topdir = os.path.normpath(os.path.dirname(__file__) + '/../../')
 
47
        missing = set()
 
48
        contributors = set()
 
49
        mailmap = utils.parse_mailmap(os.path.join(topdir, '.mailmap'))
 
50
        authors_file = open(os.path.join(topdir,
 
51
                                         'Authors'), 'r').read().lower()
 
52
 
 
53
        if os.path.exists(os.path.join(topdir, '.git')):
 
54
            for email in commands.getoutput('git log --format=%ae').split():
 
55
                if not email:
 
56
                    continue
 
57
                if "jenkins" in email and "openstack.org" in email:
 
58
                    continue
 
59
                email = '<' + email.lower() + '>'
 
60
                contributors.add(utils.str_dict_replace(email, mailmap))
 
61
        else:
 
62
            return
 
63
 
 
64
        for contributor in contributors:
 
65
            if contributor == 'cinder-core':
 
66
                continue
 
67
            if not contributor in authors_file:
 
68
                missing.add(contributor)
 
69
 
 
70
        self.assertTrue(len(missing) == 0,
 
71
                        '%r not listed in Authors' % missing)
 
72
 
 
73
    def test_all_migrations_have_downgrade(self):
 
74
        topdir = os.path.normpath(os.path.dirname(__file__) + '/../../')
 
75
        py_glob = os.path.join(topdir, "cinder", "db", "sqlalchemy",
 
76
                               "migrate_repo", "versions", "*.py")
 
77
        missing_downgrade = []
 
78
        for path in glob.iglob(py_glob):
 
79
            has_upgrade = False
 
80
            has_downgrade = False
 
81
            with open(path, "r") as f:
 
82
                for line in f:
 
83
                    if 'def upgrade(' in line:
 
84
                        has_upgrade = True
 
85
                    if 'def downgrade(' in line:
 
86
                        has_downgrade = True
 
87
 
 
88
                if has_upgrade and not has_downgrade:
 
89
                    fname = os.path.basename(path)
 
90
                    missing_downgrade.append(fname)
 
91
 
 
92
        helpful_msg = (_("The following migrations are missing a downgrade:"
 
93
                         "\n\t%s") % '\n\t'.join(sorted(missing_downgrade)))
 
94
        self.assert_(not missing_downgrade, helpful_msg)
 
95
 
 
96
 
 
97
class LockTestCase(test.TestCase):
 
98
    def test_synchronized_wrapped_function_metadata(self):
 
99
        @utils.synchronized('whatever')
 
100
        def foo():
 
101
            """Bar"""
 
102
            pass
 
103
        self.assertEquals(foo.__doc__, 'Bar', "Wrapped function's docstring "
 
104
                                              "got lost")
 
105
        self.assertEquals(foo.__name__, 'foo', "Wrapped function's name "
 
106
                                               "got mangled")
 
107
 
 
108
    def test_synchronized_internally(self):
 
109
        """We can lock across multiple green threads"""
 
110
        saved_sem_num = len(utils._semaphores)
 
111
        seen_threads = list()
 
112
 
 
113
        @utils.synchronized('testlock2', external=False)
 
114
        def f(id):
 
115
            for x in range(10):
 
116
                seen_threads.append(id)
 
117
                greenthread.sleep(0)
 
118
 
 
119
        threads = []
 
120
        pool = greenpool.GreenPool(10)
 
121
        for i in range(10):
 
122
            threads.append(pool.spawn(f, i))
 
123
 
 
124
        for thread in threads:
 
125
            thread.wait()
 
126
 
 
127
        self.assertEquals(len(seen_threads), 100)
 
128
        # Looking at the seen threads, split it into chunks of 10, and verify
 
129
        # that the last 9 match the first in each chunk.
 
130
        for i in range(10):
 
131
            for j in range(9):
 
132
                self.assertEquals(seen_threads[i * 10],
 
133
                                  seen_threads[i * 10 + 1 + j])
 
134
 
 
135
        self.assertEqual(saved_sem_num, len(utils._semaphores),
 
136
                         "Semaphore leak detected")
 
137
 
 
138
    def test_nested_external_fails(self):
 
139
        """We can not nest external syncs"""
 
140
 
 
141
        @utils.synchronized('testlock1', external=True)
 
142
        def outer_lock():
 
143
 
 
144
            @utils.synchronized('testlock2', external=True)
 
145
            def inner_lock():
 
146
                pass
 
147
            inner_lock()
 
148
        try:
 
149
            self.assertRaises(lockfile.NotMyLock, outer_lock)
 
150
        finally:
 
151
            utils.cleanup_file_locks()
 
152
 
 
153
    def test_synchronized_externally(self):
 
154
        """We can lock across multiple processes"""
 
155
        rpipe1, wpipe1 = os.pipe()
 
156
        rpipe2, wpipe2 = os.pipe()
 
157
 
 
158
        @utils.synchronized('testlock1', external=True)
 
159
        def f(rpipe, wpipe):
 
160
            try:
 
161
                os.write(wpipe, "foo")
 
162
            except OSError, e:
 
163
                self.assertEquals(e.errno, errno.EPIPE)
 
164
                return
 
165
 
 
166
            rfds, _wfds, _efds = select.select([rpipe], [], [], 1)
 
167
            self.assertEquals(len(rfds), 0, "The other process, which was"
 
168
                                            " supposed to be locked, "
 
169
                                            "wrote on its end of the "
 
170
                                            "pipe")
 
171
            os.close(rpipe)
 
172
 
 
173
        pid = os.fork()
 
174
        if pid > 0:
 
175
            os.close(wpipe1)
 
176
            os.close(rpipe2)
 
177
 
 
178
            f(rpipe1, wpipe2)
 
179
        else:
 
180
            os.close(rpipe1)
 
181
            os.close(wpipe2)
 
182
 
 
183
            f(rpipe2, wpipe1)
 
184
            os._exit(0)