diff options
| author | alevitskii <[email protected]> | 2025-09-02 06:57:05 +0300 |
|---|---|---|
| committer | alevitskii <[email protected]> | 2025-09-02 07:19:27 +0300 |
| commit | 7e399723cf6d967e20c8f9d2ee975426636242c5 (patch) | |
| tree | abd5283daba11c07afc8fc16f02aec2c19e2272a /contrib/python/portalocker/py3/tests | |
| parent | 14e9c865541d5abe545cb496c0143e4905b00c78 (diff) | |
Drop LINTER param from styling macroses and cleanup deps
Drop LINTER param from styling macroses
commit_hash:00dd2e2ee103e509cff38f520d4779974abe39a7
Diffstat (limited to 'contrib/python/portalocker/py3/tests')
| -rw-r--r-- | contrib/python/portalocker/py3/tests/__init__.py | 0 | ||||
| -rw-r--r-- | contrib/python/portalocker/py3/tests/conftest.py | 23 | ||||
| -rw-r--r-- | contrib/python/portalocker/py3/tests/temporary_file_lock.py | 14 | ||||
| -rw-r--r-- | contrib/python/portalocker/py3/tests/test_combined.py | 15 | ||||
| -rw-r--r-- | contrib/python/portalocker/py3/tests/test_redis.py | 90 | ||||
| -rw-r--r-- | contrib/python/portalocker/py3/tests/test_semaphore.py | 22 | ||||
| -rw-r--r-- | contrib/python/portalocker/py3/tests/tests.py | 357 | ||||
| -rw-r--r-- | contrib/python/portalocker/py3/tests/ya.make | 23 |
8 files changed, 0 insertions, 544 deletions
diff --git a/contrib/python/portalocker/py3/tests/__init__.py b/contrib/python/portalocker/py3/tests/__init__.py deleted file mode 100644 index e69de29bb2d..00000000000 --- a/contrib/python/portalocker/py3/tests/__init__.py +++ /dev/null diff --git a/contrib/python/portalocker/py3/tests/conftest.py b/contrib/python/portalocker/py3/tests/conftest.py deleted file mode 100644 index cf59e2b32f4..00000000000 --- a/contrib/python/portalocker/py3/tests/conftest.py +++ /dev/null @@ -1,23 +0,0 @@ -import logging -import pytest -import random -import multiprocessing - -logger = logging.getLogger(__name__) - - -def tmpfile(tmp_path): - filename = tmp_path / str(random.random()) - yield str(filename) - try: - filename.unlink(missing_ok=True) - except PermissionError: - pass - - -def pytest_sessionstart(session): - # Force spawning the process so we don't accidently inherit locks. - # I'm not a 100% certain this will work correctly unfortunately... there - # is some potential for breaking tests - multiprocessing.set_start_method('spawn') diff --git a/contrib/python/portalocker/py3/tests/temporary_file_lock.py b/contrib/python/portalocker/py3/tests/temporary_file_lock.py deleted file mode 100644 index b250bad6510..00000000000 --- a/contrib/python/portalocker/py3/tests/temporary_file_lock.py +++ /dev/null @@ -1,14 +0,0 @@ -import os -import portalocker - - -def test_temporary_file_lock(tmpfile): - with portalocker.TemporaryFileLock(tmpfile): - pass - - assert not os.path.isfile(tmpfile) - - lock = portalocker.TemporaryFileLock(tmpfile) - lock.acquire() - del lock - diff --git a/contrib/python/portalocker/py3/tests/test_combined.py b/contrib/python/portalocker/py3/tests/test_combined.py deleted file mode 100644 index 594de74b9ba..00000000000 --- a/contrib/python/portalocker/py3/tests/test_combined.py +++ /dev/null @@ -1,15 +0,0 @@ -import sys - - -def test_combined(tmpdir): - from distutils import dist - import setup - - output_file = tmpdir.join('combined.py') - combine = setup.Combine(dist.Distribution()) - combine.output_file = str(output_file) - combine.run() - sys.path.append(output_file.dirname) - import combined - assert combined - diff --git a/contrib/python/portalocker/py3/tests/test_redis.py b/contrib/python/portalocker/py3/tests/test_redis.py deleted file mode 100644 index 694c9af17ca..00000000000 --- a/contrib/python/portalocker/py3/tests/test_redis.py +++ /dev/null @@ -1,90 +0,0 @@ -import _thread -import logging -import random -import time - -import pytest -from redis import client -from redis import exceptions - -import portalocker -from portalocker import redis -from portalocker import utils - -logger = logging.getLogger(__name__) - -try: - client.Redis().ping() -except (exceptions.ConnectionError, ConnectionRefusedError): - pytest.skip('Unable to connect to redis', allow_module_level=True) - - [email protected](autouse=True) -def set_redis_timeouts(monkeypatch): - monkeypatch.setattr(utils, 'DEFAULT_TIMEOUT', 0.0001) - monkeypatch.setattr(utils, 'DEFAULT_CHECK_INTERVAL', 0.0005) - monkeypatch.setattr(redis, 'DEFAULT_UNAVAILABLE_TIMEOUT', 0.01) - monkeypatch.setattr(redis, 'DEFAULT_THREAD_SLEEP_TIME', 0.001) - monkeypatch.setattr(_thread, 'interrupt_main', lambda: None) - - -def test_redis_lock(): - channel = str(random.random()) - - lock_a = redis.RedisLock(channel) - lock_a.acquire(fail_when_locked=True) - time.sleep(0.01) - - lock_b = redis.RedisLock(channel) - try: - with pytest.raises(portalocker.AlreadyLocked): - lock_b.acquire(fail_when_locked=True) - finally: - lock_a.release() - lock_a.connection.close() - - [email protected]('timeout', [None, 0, 0.001]) [email protected]('check_interval', [None, 0, 0.0005]) -def test_redis_lock_timeout(timeout, check_interval): - connection = client.Redis() - channel = str(random.random()) - lock_a = redis.RedisLock(channel) - lock_a.acquire(timeout=timeout, check_interval=check_interval) - - lock_b = redis.RedisLock(channel, connection=connection) - with pytest.raises(portalocker.AlreadyLocked): - try: - lock_b.acquire(timeout=timeout, check_interval=check_interval) - finally: - lock_a.release() - lock_a.connection.close() - - -def test_redis_lock_context(): - channel = str(random.random()) - - lock_a = redis.RedisLock(channel, fail_when_locked=True) - with lock_a: - time.sleep(0.01) - lock_b = redis.RedisLock(channel, fail_when_locked=True) - with pytest.raises(portalocker.AlreadyLocked): - with lock_b: - pass - - -def test_redis_relock(): - channel = str(random.random()) - - lock_a = redis.RedisLock(channel, fail_when_locked=True) - with lock_a: - time.sleep(0.01) - with pytest.raises(AssertionError): - lock_a.acquire() - time.sleep(0.01) - - lock_a.release() - - -if __name__ == '__main__': - test_redis_lock() diff --git a/contrib/python/portalocker/py3/tests/test_semaphore.py b/contrib/python/portalocker/py3/tests/test_semaphore.py deleted file mode 100644 index b0c57aa23c7..00000000000 --- a/contrib/python/portalocker/py3/tests/test_semaphore.py +++ /dev/null @@ -1,22 +0,0 @@ -import random -import pytest -import portalocker -from portalocker import utils - - [email protected]('timeout', [None, 0, 0.001]) [email protected]('check_interval', [None, 0, 0.0005]) -def test_bounded_semaphore(timeout, check_interval, monkeypatch): - n = 2 - name = random.random() - monkeypatch.setattr(utils, 'DEFAULT_TIMEOUT', 0.0001) - monkeypatch.setattr(utils, 'DEFAULT_CHECK_INTERVAL', 0.0005) - - semaphore_a = portalocker.BoundedSemaphore(n, name=name, timeout=timeout) - semaphore_b = portalocker.BoundedSemaphore(n, name=name, timeout=timeout) - semaphore_c = portalocker.BoundedSemaphore(n, name=name, timeout=timeout) - - semaphore_a.acquire(timeout=timeout) - semaphore_b.acquire() - with pytest.raises(portalocker.AlreadyLocked): - semaphore_c.acquire(check_interval=check_interval, timeout=timeout) diff --git a/contrib/python/portalocker/py3/tests/tests.py b/contrib/python/portalocker/py3/tests/tests.py deleted file mode 100644 index 7a00405966e..00000000000 --- a/contrib/python/portalocker/py3/tests/tests.py +++ /dev/null @@ -1,357 +0,0 @@ -from __future__ import print_function -from __future__ import with_statement - -import os -import dataclasses -import multiprocessing -import time -import typing - -import pytest -import portalocker -from portalocker import utils -from portalocker import LockFlags - - -def test_exceptions(tmpfile): - # Open the file 2 times - a = open(tmpfile, 'a') - b = open(tmpfile, 'a') - - # Lock exclusive non-blocking - lock_flags = portalocker.LOCK_EX | portalocker.LOCK_NB - - # First lock file a - portalocker.lock(a, lock_flags) - - # Now see if we can lock file b - with pytest.raises(portalocker.LockException): - portalocker.lock(b, lock_flags) - - # Cleanup - a.close() - b.close() - - -def test_utils_base(): - class Test(utils.LockBase): - pass - - -def test_with_timeout(tmpfile): - # Open the file 2 times - with pytest.raises(portalocker.AlreadyLocked): - with portalocker.Lock(tmpfile, timeout=0.1) as fh: - print('writing some stuff to my cache...', file=fh) - with portalocker.Lock( - tmpfile, timeout=0.1, mode='wb', - fail_when_locked=True - ): - pass - print('writing more stuff to my cache...', file=fh) - - -def test_without_timeout(tmpfile): - # Open the file 2 times - with pytest.raises(portalocker.LockException): - with portalocker.Lock(tmpfile, timeout=None) as fh: - print('writing some stuff to my cache...', file=fh) - with portalocker.Lock(tmpfile, timeout=None, mode='w'): - pass - print('writing more stuff to my cache...', file=fh) - - -def test_without_fail(tmpfile): - # Open the file 2 times - with pytest.raises(portalocker.LockException): - with portalocker.Lock(tmpfile, timeout=0.1) as fh: - print('writing some stuff to my cache...', file=fh) - lock = portalocker.Lock(tmpfile, timeout=0.1) - lock.acquire(check_interval=0.05, fail_when_locked=False) - - -def test_simple(tmpfile): - with open(tmpfile, 'w') as fh: - fh.write('spam and eggs') - - fh = open(tmpfile, 'r+') - portalocker.lock(fh, portalocker.LOCK_EX) - - fh.seek(13) - fh.write('foo') - - # Make sure we didn't overwrite the original text - fh.seek(0) - assert fh.read(13) == 'spam and eggs' - - portalocker.unlock(fh) - fh.close() - - -def test_truncate(tmpfile): - with open(tmpfile, 'w') as fh: - fh.write('spam and eggs') - - with portalocker.Lock(tmpfile, mode='a+') as fh: - # Make sure we didn't overwrite the original text - fh.seek(0) - assert fh.read(13) == 'spam and eggs' - - with portalocker.Lock(tmpfile, mode='w+') as fh: - # Make sure we truncated the file - assert fh.read() == '' - - -def test_class(tmpfile): - lock = portalocker.Lock(tmpfile) - lock2 = portalocker.Lock(tmpfile, fail_when_locked=False, timeout=0.01) - - with lock: - lock.acquire() - - with pytest.raises(portalocker.LockException): - with lock2: - pass - - with lock2: - pass - - -def test_acquire_release(tmpfile): - lock = portalocker.Lock(tmpfile) - lock2 = portalocker.Lock(tmpfile, fail_when_locked=False) - - lock.acquire() # acquire lock when nobody is using it - with pytest.raises(portalocker.LockException): - # another party should not be able to acquire the lock - lock2.acquire(timeout=0.01) - - # re-acquire a held lock is a no-op - lock.acquire() - - lock.release() # release the lock - lock.release() # second release does nothing - - -def test_rlock_acquire_release_count(tmpfile): - lock = portalocker.RLock(tmpfile) - # Twice acquire - h = lock.acquire() - assert not h.closed - lock.acquire() - assert not h.closed - - # Two release - lock.release() - assert not h.closed - lock.release() - assert h.closed - - -def test_rlock_acquire_release(tmpfile): - lock = portalocker.RLock(tmpfile) - lock2 = portalocker.RLock(tmpfile, fail_when_locked=False) - - lock.acquire() # acquire lock when nobody is using it - with pytest.raises(portalocker.LockException): - # another party should not be able to acquire the lock - lock2.acquire(timeout=0.01) - - # Now acquire again - lock.acquire() - - lock.release() # release the lock - lock.release() # second release does nothing - - -def test_release_unacquired(tmpfile): - with pytest.raises(portalocker.LockException): - portalocker.RLock(tmpfile).release() - - -def test_exlusive(tmpfile): - with open(tmpfile, 'w') as fh: - fh.write('spam and eggs') - - fh = open(tmpfile, 'r') - portalocker.lock(fh, portalocker.LOCK_EX | portalocker.LOCK_NB) - - # Make sure we can't read the locked file - with pytest.raises(portalocker.LockException): - with open(tmpfile, 'r') as fh2: - portalocker.lock(fh2, portalocker.LOCK_EX | portalocker.LOCK_NB) - fh2.read() - - # Make sure we can't write the locked file - with pytest.raises(portalocker.LockException): - with open(tmpfile, 'w+') as fh2: - portalocker.lock(fh2, portalocker.LOCK_EX | portalocker.LOCK_NB) - fh2.write('surprise and fear') - - # Make sure we can explicitly unlock the file - portalocker.unlock(fh) - fh.close() - - -def test_shared(tmpfile): - with open(tmpfile, 'w') as fh: - fh.write('spam and eggs') - - f = open(tmpfile, 'r') - portalocker.lock(f, portalocker.LOCK_SH | portalocker.LOCK_NB) - - # Make sure we can read the locked file - with open(tmpfile, 'r') as fh2: - portalocker.lock(fh2, portalocker.LOCK_SH | portalocker.LOCK_NB) - assert fh2.read() == 'spam and eggs' - - # Make sure we can't write the locked file - with pytest.raises(portalocker.LockException): - with open(tmpfile, 'w+') as fh2: - portalocker.lock(fh2, portalocker.LOCK_EX | portalocker.LOCK_NB) - fh2.write('surprise and fear') - - # Make sure we can explicitly unlock the file - portalocker.unlock(f) - f.close() - - -def test_blocking_timeout(tmpfile): - flags = LockFlags.SHARED - - with pytest.warns(UserWarning): - with portalocker.Lock(tmpfile, timeout=5, flags=flags): - pass - - lock = portalocker.Lock(tmpfile, flags=flags) - with pytest.warns(UserWarning): - lock.acquire(timeout=5) - - [email protected](os.name == 'nt', - reason='Windows uses an entirely different lockmechanism') -def test_nonblocking(tmpfile): - with open(tmpfile, 'w') as fh: - with pytest.raises(RuntimeError): - portalocker.lock(fh, LockFlags.NON_BLOCKING) - - -def shared_lock(filename, **kwargs): - with portalocker.Lock( - filename, - timeout=0.1, - fail_when_locked=False, - flags=LockFlags.SHARED | LockFlags.NON_BLOCKING, - ): - time.sleep(0.2) - return True - - -def shared_lock_fail(filename, **kwargs): - with portalocker.Lock( - filename, - timeout=0.1, - fail_when_locked=True, - flags=LockFlags.SHARED | LockFlags.NON_BLOCKING, - ): - time.sleep(0.2) - return True - - -def exclusive_lock(filename, **kwargs): - with portalocker.Lock( - filename, - timeout=0.1, - fail_when_locked=False, - flags=LockFlags.EXCLUSIVE | LockFlags.NON_BLOCKING, - ): - time.sleep(0.2) - return True - - [email protected](order=True) -class LockResult: - exception_class: typing.Union[typing.Type[Exception], None] = None - exception_message: typing.Union[str, None] = None - exception_repr: typing.Union[str, None] = None - - -def lock( - filename: str, - fail_when_locked: bool, - flags: LockFlags -) -> LockResult: - # Returns a case of True, False or FileNotFound - # https://thedailywtf.com/articles/what_is_truth_0x3f_ - # But seriously, the exception properties cannot be safely pickled so we - # only return string representations of the exception properties - try: - with portalocker.Lock( - filename, - timeout=0.1, - fail_when_locked=fail_when_locked, - flags=flags, - ): - time.sleep(0.2) - return LockResult() - - except Exception as exception: - # The exceptions cannot be pickled so we cannot return them through - # multiprocessing - return LockResult( - type(exception), - str(exception), - repr(exception), - ) - - [email protected]('fail_when_locked', [True, False]) -def test_shared_processes(tmpfile, fail_when_locked): - flags = LockFlags.SHARED | LockFlags.NON_BLOCKING - - with multiprocessing.Pool(processes=2) as pool: - args = tmpfile, fail_when_locked, flags - results = pool.starmap_async(lock, 2 * [args]) - - for result in results.get(timeout=3): - assert result == LockResult() - - [email protected]('fail_when_locked', [True, False]) -def test_exclusive_processes(tmpfile, fail_when_locked): - flags = LockFlags.EXCLUSIVE | LockFlags.NON_BLOCKING - - with multiprocessing.Pool(processes=2) as pool: - # filename, fail_when_locked, flags - args = tmpfile, fail_when_locked, flags - a, b = pool.starmap_async(lock, 2 * [args]).get(timeout=3) - - assert not a.exception_class or not b.exception_class - assert issubclass( - a.exception_class or b.exception_class, - portalocker.LockException - ) - - - os.name == 'nt', - reason='Locking on Windows requires a file object', -) -def test_lock_fileno(tmpfile): - # Open the file 2 times - a = open(tmpfile, 'a') - b = open(tmpfile, 'a') - - # Lock exclusive non-blocking - flags = LockFlags.SHARED | LockFlags.NON_BLOCKING - - # First lock file a - portalocker.lock(a, flags) - - # Now see if we can lock using fileno() - portalocker.lock(b.fileno(), flags) - - # Cleanup - a.close() - b.close() - diff --git a/contrib/python/portalocker/py3/tests/ya.make b/contrib/python/portalocker/py3/tests/ya.make deleted file mode 100644 index 342f72a5391..00000000000 --- a/contrib/python/portalocker/py3/tests/ya.make +++ /dev/null @@ -1,23 +0,0 @@ -PY3TEST() - -SUBSCRIBER(g:python-contrib) - -NO_LINT() - -PEERDIR( - contrib/python/portalocker - contrib/python/redis -) - -TEST_SRCS( - __init__.py - conftest.py - temporary_file_lock.py - # Tests intallation. - # test_combined.py - test_redis.py - test_semaphore.py - tests.py -) - -END() |
