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

How to use get_lock method in yandex-tank

Best Python code snippet using yandex-tank

compilelock.py

Source:compilelock.py

...19def force_unlock():20    """21    Delete the compilation lock if someone else has it.22    """23    get_lock(min_wait=0, max_wait=0.001, timeout=0)24    release_lock()25@contextmanager26def lock_ctx(lock_dir=None, keep_lock=False, **kw):27    get_lock(lock_dir=lock_dir, **kw)28    yield29    if not keep_lock:30        release_lock()31# We define this name with an underscore so that python shutdown32# deletes this before non-underscore names (like os).  We need to do33# it this way to avoid errors on shutdown.34def _get_lock(lock_dir=None, **kw):35    """36    Obtain lock on compilation directory.37    Parameters38    ----------39    kw40        Additional arguments to be forwarded to the `lock` function when41        acquiring the lock.42    Notes43    -----44    We can lock only on 1 directory at a time.45    """46    if lock_dir is None:47        lock_dir = os.path.join(config.compiledir, 'lock_dir')48    if not hasattr(get_lock, 'n_lock'):...

test_lock.py

Source:test_lock.py

...7class TestLock(object):8    @pytest.fixture()9    def r_decoded(self, request):10        return _get_client(Redis, request=request, decode_responses=True)11    def get_lock(self, redis, *args, **kwargs):12        kwargs['lock_class'] = Lock13        return redis.lock(*args, **kwargs)14    def test_lock(self, r):15        lock = self.get_lock(r, 'foo')16        assert lock.acquire(blocking=False)17        assert r.get('foo') == lock.local.token18        assert r.ttl('foo') == -119        lock.release()20        assert r.get('foo') is None21    def test_lock_token(self, r):22        lock = self.get_lock(r, 'foo')23        assert lock.acquire(blocking=False, token='test')24        assert r.get('foo') == b'test'25        assert lock.local.token == b'test'26        assert r.ttl('foo') == -127        lock.release()28        assert r.get('foo') is None29        assert lock.local.token is None30    def test_locked(self, r):31        lock = self.get_lock(r, 'foo')32        assert lock.locked() is False33        lock.acquire(blocking=False)34        assert lock.locked() is True35        lock.release()36        assert lock.locked() is False37    def _test_owned(self, client):38        lock = self.get_lock(client, 'foo')39        assert lock.owned() is False40        lock.acquire(blocking=False)41        assert lock.owned() is True42        lock.release()43        assert lock.owned() is False44        lock2 = self.get_lock(client, 'foo')45        assert lock.owned() is False46        assert lock2.owned() is False47        lock2.acquire(blocking=False)48        assert lock.owned() is False49        assert lock2.owned() is True50        lock2.release()51        assert lock.owned() is False52        assert lock2.owned() is False53    def test_owned(self, r):54        self._test_owned(r)55    def test_owned_with_decoded_responses(self, r_decoded):56        self._test_owned(r_decoded)57    def test_competing_locks(self, r):58        lock1 = self.get_lock(r, 'foo')59        lock2 = self.get_lock(r, 'foo')60        assert lock1.acquire(blocking=False)61        assert not lock2.acquire(blocking=False)62        lock1.release()63        assert lock2.acquire(blocking=False)64        assert not lock1.acquire(blocking=False)65        lock2.release()66    def test_timeout(self, r):67        lock = self.get_lock(r, 'foo', timeout=10)68        assert lock.acquire(blocking=False)69        assert 8 < r.ttl('foo') <= 1070        lock.release()71    def test_float_timeout(self, r):72        lock = self.get_lock(r, 'foo', timeout=9.5)73        assert lock.acquire(blocking=False)74        assert 8 < r.pttl('foo') <= 950075        lock.release()76    def test_blocking_timeout(self, r):77        lock1 = self.get_lock(r, 'foo')78        assert lock1.acquire(blocking=False)79        lock2 = self.get_lock(r, 'foo', blocking_timeout=0.2)80        start = time.time()81        assert not lock2.acquire()82        assert (time.time() - start) > 0.283        lock1.release()84    def test_context_manager(self, r):85        # blocking_timeout prevents a deadlock if the lock can't be acquired86        # for some reason87        with self.get_lock(r, 'foo', blocking_timeout=0.2) as lock:88            assert r.get('foo') == lock.local.token89        assert r.get('foo') is None90    def test_context_manager_raises_when_locked_not_acquired(self, r):91        r.set('foo', 'bar')92        with pytest.raises(LockError):93            with self.get_lock(r, 'foo', blocking_timeout=0.1):94                pass95    def test_high_sleep_raises_error(self, r):96        "If sleep is higher than timeout, it should raise an error"97        with pytest.raises(LockError):98            self.get_lock(r, 'foo', timeout=1, sleep=2)99    def test_releasing_unlocked_lock_raises_error(self, r):100        lock = self.get_lock(r, 'foo')101        with pytest.raises(LockError):102            lock.release()103    def test_releasing_lock_no_longer_owned_raises_error(self, r):104        lock = self.get_lock(r, 'foo')105        lock.acquire(blocking=False)106        # manually change the token107        r.set('foo', 'a')108        with pytest.raises(LockNotOwnedError):109            lock.release()110        # even though we errored, the token is still cleared111        assert lock.local.token is None112    def test_extend_lock(self, r):113        lock = self.get_lock(r, 'foo', timeout=10)114        assert lock.acquire(blocking=False)115        assert 8000 < r.pttl('foo') <= 10000116        assert lock.extend(10)117        assert 16000 < r.pttl('foo') <= 20000118        lock.release()119    def test_extend_lock_float(self, r):120        lock = self.get_lock(r, 'foo', timeout=10.0)121        assert lock.acquire(blocking=False)122        assert 8000 < r.pttl('foo') <= 10000123        assert lock.extend(10.0)124        assert 16000 < r.pttl('foo') <= 20000125        lock.release()126    def test_extending_unlocked_lock_raises_error(self, r):127        lock = self.get_lock(r, 'foo', timeout=10)128        with pytest.raises(LockError):129            lock.extend(10)130    def test_extending_lock_with_no_timeout_raises_error(self, r):131        lock = self.get_lock(r, 'foo')132        assert lock.acquire(blocking=False)133        with pytest.raises(LockError):134            lock.extend(10)135        lock.release()136    def test_extending_lock_no_longer_owned_raises_error(self, r):137        lock = self.get_lock(r, 'foo', timeout=10)138        assert lock.acquire(blocking=False)139        r.set('foo', 'a')140        with pytest.raises(LockNotOwnedError):141            lock.extend(10)142    def test_reacquire_lock(self, r):143        lock = self.get_lock(r, 'foo', timeout=10)144        assert lock.acquire(blocking=False)145        assert r.pexpire('foo', 5000)146        assert r.pttl('foo') <= 5000147        assert lock.reacquire()148        assert 8000 < r.pttl('foo') <= 10000149        lock.release()150    def test_reacquiring_unlocked_lock_raises_error(self, r):151        lock = self.get_lock(r, 'foo', timeout=10)152        with pytest.raises(LockError):153            lock.reacquire()154    def test_reacquiring_lock_with_no_timeout_raises_error(self, r):155        lock = self.get_lock(r, 'foo')156        assert lock.acquire(blocking=False)157        with pytest.raises(LockError):158            lock.reacquire()159        lock.release()160    def test_reacquiring_lock_no_longer_owned_raises_error(self, r):161        lock = self.get_lock(r, 'foo', timeout=10)162        assert lock.acquire(blocking=False)163        r.set('foo', 'a')164        with pytest.raises(LockNotOwnedError):165            lock.reacquire()166class TestLockClassSelection(object):167    def test_lock_class_argument(self, r):168        class MyLock(object):169            def __init__(self, *args, **kwargs):170                pass171        lock = r.lock('foo', lock_class=MyLock)...

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.