KaneAI - World’s First E2E Software Testing Agent.Book a Demo Join TestMu Conference

How to use test_lock method in avocado

Best Python code snippet using avocado_python

test_lock.py

Source:test_lock.py

1import etcd2try:3    import mock4except ImportError:5    from unittest import mock6from etcd.tests.unit import TestClientApiBase7class TestClientLock(TestClientApiBase):8    def recursive_read(self):9        nodes = [10            {"key": "/_locks/test_lock/1", "value": "2qwwwq",11             "modifiedIndex":33,"createdIndex":33},12            {"key": "/_locks/test_lock/34", "value": self.locker.uuid,13             "modifiedIndex":34,"createdIndex":34},14        ]15        d = {16            "action": "get",17            "node": {"dir": True,18                     "nodes": [{"key":"/_locks/test_lock", "dir": True,19                                "nodes": nodes}]}20        }21        self._mock_api(200, d)22    def setUp(self):23        super(TestClientLock, self).setUp()24        self.locker = etcd.Lock(self.client, 'test_lock')25    def test_initialization(self):26        """27        Verify the lock gets initialized correctly28        """29        self.assertEquals(self.locker.name, u'test_lock')30        self.assertEquals(self.locker.path, u'/_locks/test_lock')31        self.assertEquals(self.locker.is_taken, False)32    def test_acquire(self):33        """34        Acquiring a precedingly inexistent lock works.35        """36        l = etcd.Lock(self.client, 'test_lock')37        l._find_lock = mock.MagicMock(spec=l._find_lock, return_value=False)38        l._acquired = mock.MagicMock(spec=l._acquired, return_value=True)39        # Mock the write40        d = {41            u'action': u'set',42            u'node': {43                u'modifiedIndex': 190,44                u'key': u'/_locks/test_lock/1',45                u'value': l.uuid46            }47        }48        self._mock_api(200, d)49        self.assertEquals(l.acquire(), True)50        self.assertEquals(l._sequence, '1')51    def test_is_acquired(self):52        """53        Test is_acquired54        """55        self.locker._sequence = '1'56        d = {57            u'action': u'get',58            u'node': {59                u'modifiedIndex': 190,60                u'key': u'/_locks/test_lock/1',61                u'value': self.locker.uuid62            }63        }64        self._mock_api(200, d)65        self.locker.is_taken = True66        self.assertEquals(self.locker.is_acquired, True)67    def test_is_not_acquired(self):68        """69        Test is_acquired failures70        """71        self.locker._sequence = '2'72        self.locker.is_taken = False73        self.assertEquals(self.locker.is_acquired, False)74        self.locker.is_taken = True75        self._mock_exception(etcd.EtcdKeyNotFound, self.locker.lock_key)76        self.assertEquals(self.locker.is_acquired, False)77        self.assertEquals(self.locker.is_taken, False)78    def test_acquired(self):79        """80        Test the acquiring primitives81        """82        self.locker._sequence = '4'83        retval = ('/_locks/test_lock/4', None)84        self.locker._get_locker = mock.MagicMock(85            spec=self.locker._get_locker, return_value=retval)86        self.assertTrue(self.locker._acquired())87        self.assertTrue(self.locker.is_taken)88        retval = ('/_locks/test_lock/1', '/_locks/test_lock/4')89        self.locker._get_locker = mock.MagicMock(return_value=retval)90        self.assertFalse(self.locker._acquired(blocking=False))91        self.assertFalse(self.locker.is_taken)92        d = {93            u'action': u'delete',94            u'node': {95                u'modifiedIndex': 190,96                u'key': u'/_locks/test_lock/1',97                u'value': self.locker.uuid98            }99        }100        self._mock_api(200, d)101        returns = [('/_locks/test_lock/1', '/_locks/test_lock/4'),  ('/_locks/test_lock/4', None)]102        def side_effect():103            return returns.pop()104        self.locker._get_locker = mock.MagicMock(105            spec=self.locker._get_locker, side_effect=side_effect)106        self.assertTrue(self.locker._acquired())107    def test_acquired_no_timeout(self):108        self.locker._sequence = 4109        returns = [('/_locks/test_lock/4', None), ('/_locks/test_lock/1', '/_locks/test_lock/4')]110        def side_effect():111            return returns.pop()112        d = {113            u'action': u'get',114            u'node': {115                u'modifiedIndex': 190,116                u'key': u'/_locks/test_lock/4',117                u'value': self.locker.uuid118            }119        }120        self._mock_api(200, d)121        self.locker._get_locker = mock.create_autospec(122            self.locker._get_locker, side_effect=side_effect)123        self.assertTrue(self.locker._acquired())124    def test_lock_key(self):125        """126        Test responses from the lock_key property127        """128        with self.assertRaises(ValueError):129            self.locker.lock_key130        self.locker._sequence = '5'131        self.assertEquals(u'/_locks/test_lock/5',self.locker.lock_key)132    def test_set_sequence(self):133        self.locker._set_sequence('/_locks/test_lock/10')134        self.assertEquals('10', self.locker._sequence)135    def test_find_lock(self):136        d = {137            u'action': u'get',138            u'node': {139                u'modifiedIndex': 190,140                u'key': u'/_locks/test_lock/1',141                u'value': self.locker.uuid142            }143        }144        self._mock_api(200, d)145        self.locker._sequence = '1'146        self.assertTrue(self.locker._find_lock())147        # Now let's pretend the lock is not there148        self._mock_exception(etcd.EtcdKeyNotFound, self.locker.lock_key)149        self.assertFalse(self.locker._find_lock())150        self.locker._sequence = None151        self.recursive_read()152        self.assertTrue(self.locker._find_lock())153        self.assertEquals(self.locker._sequence, '34')154    def test_get_locker(self):155        self.recursive_read()156        self.assertEquals((u'/_locks/test_lock/1', u'/_locks/test_lock/1'),157                          self.locker._get_locker())158        with self.assertRaises(etcd.EtcdLockExpired):159            self.locker._sequence = '35'160            self.locker._get_locker()161    def test_release(self):162        d = {163            u'action': u'delete',164            u'node': {165                u'modifiedIndex': 190,166                u'key': u'/_locks/test_lock/1',167                u'value': self.locker.uuid168            }169        }170        self._mock_api(200, d)171        self.locker._sequence = 1172        self.locker.is_taken = True173        self.locker.release()...

test_deadlock.py

Source:test_deadlock.py

...4import pypar as mpi5import scipy as sc6Nnode = mpi.size()7myrank = mpi.rank()8def test_lock(Nmpi,fields,pbc_opt=None):9    if myrank == 0:10	print 'PBC : %s, start' % pbc_opt11    mpi.barrier()12    for i in xrange(len(fields)):13	fields[i][:,:,:6] = 1.14	fields[i][:,:,6:] = 0.15	#print 'I`m', myrank,'Field %s Direction x1 sum before = '%i,fields[i][:,:,6].sum()16	#print 'I`m', myrank,'Field %s Direction x2 sum before = '%i,fields[i][:,:,7].sum()17	#print 'I`m', myrank,'Field %s Direction y1 sum before = '%i,fields[i][:,:,8].sum()18	#print 'I`m', myrank,'Field %s Direction y2 sum before = '%i,fields[i][:,:,9].sum()19	#print 'I`m', myrank,'Field %s Direction z1 sum before = '%i,fields[i][:,:,10].sum()20	#print 'I`m', myrank,'Field %s Direction z2 sum before = '%i,fields[i][:,:,11].sum()21    mpi.barrier()22    if myrank != 0:23	targets = MPI.calc_mpitarget(Nmpi, myrank)24	targets_pbc = MPI.calc_mpitarget_pbc(Nmpi, myrank, pbc_opt)25	message_range = MPI.test_making_message_range()26	MPI.test_mpi_exchange(fields, Nmpi, myrank, targets, message_range)27	MPI.test_mpi_exchange_pbc(fields, myrank,targets_pbc, message_range, pbc_opt)28	for i in xrange(len(fields)):29	    print 'I`m', myrank,'Field %s Direction x1 sum after = '%i,fields[i][:,:,6].sum()30	    print 'I`m', myrank,'Field %s Direction x2 sum after = '%i,fields[i][:,:,7].sum()31	    print 'I`m', myrank,'Field %s Direction y1 sum after = '%i,fields[i][:,:,8].sum()32	    print 'I`m', myrank,'Field %s Direction y2 sum after = '%i,fields[i][:,:,9].sum()33	    print 'I`m', myrank,'Field %s Direction z1 sum after = '%i,fields[i][:,:,10].sum()34	    print 'I`m', myrank,'Field %s Direction z2 sum after = '%i,fields[i][:,:,11].sum()35    mpi.barrier()36    if myrank == 0:37	print 'PBC : %s, Done' % pbc_opt38	print39	print40	print41if __name__ == '__main__':42    Nx = 100043    Ny = 10044    fields = []45    fields.append(sc.zeros((Nx,Ny,12),'f'))46    fields.append(sc.zeros((Nx,Ny,12),'f'))47    fields.append(sc.zeros((Nx,Ny,12),'f'))48    Nmpi = (8,1,1)49    # one field50    #test_lock(Nmpi,fields[:1])51    #test_lock(Nmpi,fields[:1],'x')52    #test_lock(Nmpi,fields[:1],'xy')53    #test_lock(Nmpi,fields[:1],'xyz')54    # two fields55    #test_lock(Nmpi,fields[:2])56    #test_lock(Nmpi,fields[:2],'x')57    #test_lock(Nmpi,fields[:2],'xy')58    #test_lock(Nmpi,fields[:2],'xyz')59    # three fields60    #test_lock(Nmpi,fields[:3])61    #test_lock(Nmpi,fields[:3],'x')62    #test_lock(Nmpi,fields[:3],'xy')63    test_lock(Nmpi,fields[:3],'xyz')64    Nmpi = (2,4,1)65    # one field66    #test_lock(Nmpi,fields[:1])67    #test_lock(Nmpi,fields[:1],'x')68    #test_lock(Nmpi,fields[:1],'xy')69    #test_lock(Nmpi,fields[:1],'xyz')70    # two fields71    #test_lock(Nmpi,fields[:2])72    #test_lock(Nmpi,fields[:2],'x')73    #test_lock(Nmpi,fields[:2],'xy')74    #test_lock(Nmpi,fields[:2],'xyz')75    # three fields76    #test_lock(Nmpi,fields[:3])77    #test_lock(Nmpi,fields[:3],'x')78    #test_lock(Nmpi,fields[:3],'xy')79    #test_lock(Nmpi,fields[:3],'xyz')80    Nmpi = (2,2,2)81    # one field82    #test_lock(Nmpi,fields[:1])83    #test_lock(Nmpi,fields[:1],'x')84    #test_lock(Nmpi,fields[:1],'xy')85    #test_lock(Nmpi,fields[:1],'xyz')86    # two fields87    #test_lock(Nmpi,fields[:2])88    #test_lock(Nmpi,fields[:2],'x')89    #test_lock(Nmpi,fields[:2],'xy')90    #test_lock(Nmpi,fields[:2],'xyz')91    # three fields92    #test_lock(Nmpi,fields[:3])93    #test_lock(Nmpi,fields[:3],'x')94    #test_lock(Nmpi,fields[:3],'xy')95    #test_lock(Nmpi,fields[:3],'xyz')...

Automation Testing Tutorials

Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.