0

我正在编写一个使用 SQLAlchemy 和 Postgres 的 Web 应用程序，并限制对某些数据的同时访问，我正在使用锁 ( 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()

或者也许还有其他方法可以自动测试这些锁？

4

0 回答 0