← Back to branch summary

~ubuntu-branches/ubuntu/raring/nova/raring-proposed

« back to all changes in this revision

Viewing changes to nova/tests/test_misc.py

  • Committer: Package Import Robot
  • Author(s): Chuck Short, Adam Gandelman, Chuck Short
  • Date: 2012-11-23 09:04:58 UTC
  • mfrom: (1.1.66)
  • Revision ID: package-import@ubuntu.com-20121123090458-91565o7aev1i1h71
Tags: 2013.1~g1-0ubuntu1
https://launchpad.net/bugs/1073289
https://launchpad.net/bugs/1076442
[ Adam Gandelman ]
* debian/control: Ensure novaclient is upgraded with nova,
  require python-keystoneclient >= 1:2.9.0. (LP: #1073289)
* debian/patches/{ubuntu/*, rbd-security.patch}: Dropped, applied
  upstream.
* debian/control: Add python-testtools to Build-Depends.

[ Chuck Short ]
* New upstream version.
* Refreshed debian/patches/avoid_setuptools_git_dependency.patch.
* debian/rules: FTBFS if missing binaries.
* debian/nova-scheudler.install: Add missing rabbit-queues and
  nova-rpc-zmq-receiver.
* Remove nova-volume since it doesnt exist anymore, transition to cinder-*.
* debian/rules: install apport hook in the right place.
* debian/patches/ubuntu-show-tests.patch: Display test failures.
* debian/control: Add depends on genisoimage
* debian/control: Suggest guestmount.
* debian/control: Suggest websockify. (LP: #1076442)
* debian/nova.conf: Disable nova-volume service.
* debian/control: Depend on xen-system-* rather than the hypervisor.
* debian/control, debian/mans/nova-conductor.8, debian/nova-conductor.init,
  debian/nova-conductor.install, debian/nova-conductor.logrotate
  debian/nova-conductor.manpages, debian/nova-conductor.postrm
  debian/nova-conductor.upstart.in: Add nova-conductor service.
* debian/control: Add python-fixtures as a build deps.

Show diffs side-by-side

added added

removed removed

Lines of Context:
nova/tests/test_misc.py
24
24
 
25
25
from nova import exception
26
26
from nova import test
27
 
from nova import utils
28
27
 
29
28
 
30
29
class ExceptionTestCase(test.TestCase):
63
62
        helpful_msg = (_("The following migrations are missing a downgrade:"
64
63
                         "\n\t%s") % '\n\t'.join(sorted(missing_downgrade)))
65
64
        self.assert_(not missing_downgrade, helpful_msg)
66
 
 
67
 
 
68
 
class LockTestCase(test.TestCase):
69
 
    def test_synchronized_wrapped_function_metadata(self):
70
 
        @utils.synchronized('whatever')
71
 
        def foo():
72
 
            """Bar"""
73
 
            pass
74
 
        self.assertEquals(foo.__doc__, 'Bar', "Wrapped function's docstring "
75
 
                                              "got lost")
76
 
        self.assertEquals(foo.__name__, 'foo', "Wrapped function's name "
77
 
                                               "got mangled")
78
 
 
79
 
    def test_synchronized_internally(self):
80
 
        """We can lock across multiple green threads"""
81
 
        saved_sem_num = len(utils._semaphores)
82
 
        seen_threads = list()
83
 
 
84
 
        @utils.synchronized('testlock2', external=False)
85
 
        def f(id):
86
 
            for x in range(10):
87
 
                seen_threads.append(id)
88
 
                greenthread.sleep(0)
89
 
 
90
 
        threads = []
91
 
        pool = greenpool.GreenPool(10)
92
 
        for i in range(10):
93
 
            threads.append(pool.spawn(f, i))
94
 
 
95
 
        for thread in threads:
96
 
            thread.wait()
97
 
 
98
 
        self.assertEquals(len(seen_threads), 100)
99
 
        # Looking at the seen threads, split it into chunks of 10, and verify
100
 
        # that the last 9 match the first in each chunk.
101
 
        for i in range(10):
102
 
            for j in range(9):
103
 
                self.assertEquals(seen_threads[i * 10],
104
 
                                  seen_threads[i * 10 + 1 + j])
105
 
 
106
 
        self.assertEqual(saved_sem_num, len(utils._semaphores),
107
 
                         "Semaphore leak detected")
108
 
 
109
 
    def test_nested_external_works(self):
110
 
        """We can nest external syncs"""
111
 
        with utils.tempdir() as tempdir:
112
 
            self.flags(lock_path=tempdir)
113
 
            sentinel = object()
114
 
 
115
 
            @utils.synchronized('testlock1', external=True)
116
 
            def outer_lock():
117
 
 
118
 
                @utils.synchronized('testlock2', external=True)
119
 
                def inner_lock():
120
 
                    return sentinel
121
 
                return inner_lock()
122
 
 
123
 
            self.assertEqual(sentinel, outer_lock())
124
 
 
125
 
    def test_synchronized_externally(self):
126
 
        """We can lock across multiple processes"""
127
 
        with utils.tempdir() as tempdir:
128
 
            self.flags(lock_path=tempdir)
129
 
            rpipe1, wpipe1 = os.pipe()
130
 
            rpipe2, wpipe2 = os.pipe()
131
 
 
132
 
            @utils.synchronized('testlock1', external=True)
133
 
            def f(rpipe, wpipe):
134
 
                try:
135
 
                    os.write(wpipe, "foo")
136
 
                except OSError, e:
137
 
                    self.assertEquals(e.errno, errno.EPIPE)
138
 
                    return
139
 
 
140
 
                rfds, _wfds, _efds = select.select([rpipe], [], [], 1)
141
 
                self.assertEquals(len(rfds), 0, "The other process, which was"
142
 
                                                " supposed to be locked, "
143
 
                                                "wrote on its end of the "
144
 
                                                "pipe")
145
 
                os.close(rpipe)
146
 
 
147
 
            pid = os.fork()
148
 
            if pid > 0:
149
 
                os.close(wpipe1)
150
 
                os.close(rpipe2)
151
 
 
152
 
                f(rpipe1, wpipe2)
153
 
            else:
154
 
                os.close(rpipe1)
155
 
                os.close(wpipe2)
156
 
 
157
 
                f(rpipe2, wpipe1)
158
 
                os._exit(0)