src/modernmp/test_synchronize.py

# Copyright (C) 2020 The Android Open Source Project
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#      http:#www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
"""Tests for interprocess synchronization"""

import logging
import pytest
from _multiprocessing import SemLock
from .util_test import (
  run_resmanc,
)

from .synchronize import (
  Lock,
  RLock,
  SEMAPHORE,
  ShmLock,
)

log = logging.getLogger(__name__)

def test_lock_basic():
  """Test that we can create a ShmLock and acquire, release it"""
  with run_resmanc():
    lock = ShmLock(SEMAPHORE, 1, 1)
    lock.acquire(block=False)
    lock.release()

def _get_semlock(lock):
  assert isinstance(lock, ShmLock)
  return lock._semlock  # pylint: disable=protected-access

def _rebuild_semlock(handle, kind, maxvalue, name):
  # pylint: disable=protected-access
  return SemLock._rebuild(handle, kind, maxvalue, name)

def test_lock_destructor():
  """Test that we unlink the lock when all references drop"""
  with run_resmanc() as resmanc:
    lock = ShmLock(SEMAPHORE, 1, 1)
    sl = _get_semlock(lock)
    lock_name = lock.name
    assert lock_name == sl.name
    #log.debug("lock name is %r", lock_name)
    _rebuild_semlock(sl.handle, sl.kind, sl.maxvalue, sl.name)
    del lock
    resmanc.getpid()  # Force refcount flush
    with pytest.raises(FileNotFoundError):
      # Lock should have been unlinked on server
      _rebuild_semlock(sl.handle, sl.kind, sl.maxvalue, sl.name)

def test_lock():
  """Test the actual Lock class"""
  with run_resmanc():
    lock = Lock()
    assert repr(lock) == "<Lock(owner=None)>"
    lock.acquire()
    assert repr(lock) == "<Lock(owner=MainProcess)>"
    assert not lock.acquire(block=False)
    lock.release()

def test_rlock():
  """Test the recursive Lock class"""
  with run_resmanc():
    lock = RLock()
    assert repr(lock) == "<RLock(None, 0)>"
    lock.acquire()
    assert repr(lock) == "<RLock(MainProcess, 1)>"
    assert lock.acquire(block=False)
    assert repr(lock) == "<RLock(MainProcess, 2)>"
    lock.release()
    lock.release()