Блог

Главная - Вопросы по программированию - Тест: убедитесь, что консультативная блокировка Postgres работает правильно

Тест: убедитесь, что консультативная блокировка Postgres работает правильно

#python #postgresql #sqlalchemy #locking #pytest

Вопрос:

Я пишу веб-приложение, которое использует SQLAlchemy и Postgres, и для ограничения одновременного доступа к некоторым данным я использую блокировки ( pg_advisory_xact_lock ). Давайте предположим, что у меня есть такой код

 def func():
    obtain_lock()
    try:
       data = ['abc', 123] # 1
       if data in DB:
          raise Exception('Data is already in DB')
       insert_data()  
    # lock is released automatically

Чтобы убедиться, что замки работают правильно, я:

  • запустите два идентичных сеанса отладки с точками останова в строке (1)
  • смотрите, чтобы один сеанс достиг точки останова, а другой зависал без удара
  • продолжите первый сеанс, убедитесь, что все работает правильно, отпустите блокировку
  • точка останова во втором сеансе достигнута, проверьте, что возникло исключение

Возможно ли воспроизвести такое поведение в Pytest, чтобы иметь что-то подобное?

 # code.py
def func():
    obtain_lock()
    set_breakpoint()
    try:
       data = ['abc', 123] # 1
       if data in DB:
          raise Exception('Data is already in DB')
       insert_data()  
    # lock is released automatically

# test.py
def test():
    t1 = Thread(target=func)
    t2 = Thread(target=func)
    t1.start()
    t2.start()
    
    assert t1.breakpoint.hit and not t2.breakpoint.hit
    t1.breakpoint.resume()
    assert data in DB
    assert t2.breakpoint.hit
    with pytest.raises(Exception):
        t2.breakpoint.resume()

Или, может быть, есть какой-то другой способ автоматизировать тестирование этих блокировок?

Метки: ,