summaryrefslogtreecommitdiffstats
path: root/contrib/python/portalocker/py3/tests
diff options
context:
space:
mode:
authoralevitskii <[email protected]>2025-09-02 06:57:05 +0300
committeralevitskii <[email protected]>2025-09-02 07:19:27 +0300
commit7e399723cf6d967e20c8f9d2ee975426636242c5 (patch)
treeabd5283daba11c07afc8fc16f02aec2c19e2272a /contrib/python/portalocker/py3/tests
parent14e9c865541d5abe545cb496c0143e4905b00c78 (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__.py0
-rw-r--r--contrib/python/portalocker/py3/tests/conftest.py23
-rw-r--r--contrib/python/portalocker/py3/tests/temporary_file_lock.py14
-rw-r--r--contrib/python/portalocker/py3/tests/test_combined.py15
-rw-r--r--contrib/python/portalocker/py3/tests/test_redis.py90
-rw-r--r--contrib/python/portalocker/py3/tests/test_semaphore.py22
-rw-r--r--contrib/python/portalocker/py3/tests/tests.py357
-rw-r--r--contrib/python/portalocker/py3/tests/ya.make23
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()