File size: 956 Bytes
079c32c |
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 |
import pytest
import numpy as np
from collections import deque
from ding.utils import LockContext, LockContextType, get_rw_file_lock
@pytest.mark.unittest
def test_usage():
lock = LockContext(LockContextType.PROCESS_LOCK)
queue = deque(maxlen=10)
data = np.random.randn(4)
with lock:
queue.append(np.copy(data))
with lock:
output = queue.popleft()
assert (output == data).all()
lock.acquire()
queue.append(np.copy(data))
lock.release()
lock.acquire()
output = queue.popleft()
lock.release()
assert (output == data).all()
@pytest.mark.unittest
def test_get_rw_file_lock():
path = 'tmp.npy'
# TODO real read-write case
read_lock = get_rw_file_lock(path, 'read')
write_lock = get_rw_file_lock(path, 'write')
with write_lock:
np.save(path, np.random.randint(0, 1, size=(3, 4)))
with read_lock:
data = np.load(path)
assert data.shape == (3, 4)
|