Skip to content
This repository
tree: 60e0506e44
Fetching contributors…

Octocat-spinner-32-eaf2f5

Cannot retrieve contributors at this time

file 96 lines (76 sloc) 2.935 kb
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
import unittest
import re
from ddlockclient import DDLockClient, DDLock, eurl

servers = ['localhost']


class TestError(Exception):
    pass


class ClientTest(unittest.TestCase):

    def setUp(self):
        self.c = DDLockClient(servers=servers)

    def _lock(self, name):
        return self.c.trylock(name, 0) # no block

    def test_init(self):
        self.assertTrue(isinstance(self.c, DDLockClient),
                        "Got a client object")

    def test_a(self):
        lock = self._lock('test_a')
        self.assertTrue(isinstance(lock, DDLock),
                        "Got a lock for 'test_a'")

    def test_a2(self):
        lock = self._lock('test_a')
        self.assertTrue(isinstance(lock, DDLock),
                        "Got a lock for 'test_a' again")

    def test_b(self):
        lock = self._lock('test_b')
        self.assertTrue(isinstance(lock, DDLock),
                        "Got a lock for 'test_b'")
        rv = lock.release()
        self.assertTrue(rv, "Lock release succeeded")

        rv = None
        try:
            rv = lock.release()
            self.fail("Expected an error")
        except Exception, e:
            self.assertTrue(re.search('ERR didnthave',
                            str(e)),
                            "release() die if it couldn't release")
        self.assertEquals(rv, None, "no return value")

        lock2 = self._lock('test_b')
        self.assertTrue(isinstance(lock2, DDLock),
                        "Got a lock for 'test_b' again")

    def test_c(self):
        lock = self._lock('test_c')
        self.assertTrue(isinstance(lock, DDLock),
                        "Got a lock for 'test_c'")
        lock2 = self._lock('test_c')
        self.assertEquals(lock2,
                          None,
                          "Got no lock for 'test_c' again without release")

    def test_d(self):
        lock = self._lock('test_d')
        self.assertTrue(isinstance(lock, DDLock),
                        "Got a lock for 'test_d'")
        # a lock will be released when going out of with statement
        with lock:
            pass

        lock2 = self._lock('test_d')
        self.assertTrue(isinstance(lock2, DDLock),
                        "Got a lock for 'test_d' again")

    def test_e(self):
        lock = self._lock('test')
        self.assertTrue(isinstance(lock, DDLock),
                        "Got a lock for 'test'")
        try:
            raise TestError("test error")
        except Exception, e:
            self.assertTrue(isinstance(e, TestError))
        finally:
            lock.release()

        lock = self._lock('test')
        self.assertTrue(isinstance(lock, DDLock),
                        "able to lock 'test' again")

    def test_eurl(self):
        name = eurl('foo|bar')
        self.assertEqual(name, "foo%7Cbar")
        name = eurl('foo|bar/baz')
        self.assertEqual(name, "foo%7Cbar%2Fbaz")
Something went wrong with that request. Please try again.