模拟的 lockf 最终被调用了两次，而它应该被调用一次
Mocked `lockf` ends up getting called twice when it should be called once
我正在编写一个测试来检查我的程序的锁定功能。
对于第一个测试，我试图断言 lockf 函数是否只被调用一次，因为后续的锁定调用而不释放锁会导致错误（这正是我想要的）。
所以，我模拟了 lockf 函数，并断言它被调用了一次。
但是 lockf 函数应该被调用一次却被调用了两次，（第二次应该会出错）。
我在这里做错了什么？我该如何解决这个问题？
我的猜测是 t1 线程在 t2 锁定它之前完成并释放了锁，因此没有错误，lockf 最终被调用了两次。
代码
测试
import logging
from argparse import Namespace
from threading import Thread
from unittest.mock import patch
import pytest
import pacstall.__main__
class TestLockingCommands:
@pytest.mark.parametrize("command", ["install", "remove", "upgrade", "repo"])
def test_if_lock_is_called(self, command: str, monkeypatch: pytest.MonkeyPatch):
logging.getLogger(__name__).setLevel(logging.DEBUG)
with patch("fcntl.lockf") as mock_lockf:
t1 = Thread(
target=pacstall.__main__.lock,
args=(Namespace(command=command), logging.getLogger(__name__)),
)
t2 = Thread(
target=pacstall.__main__.lock,
args=(Namespace(command=command), logging.getLogger(__name__)),
)
t1.start()
t2.start()
t1.join()
t2.join()
mock_lockf.assert_called_once()
生产代码
import fcntl
import sys
from argparse import Namespace
from asyncio import run
from getpass import getuser
from logging import Logger, getLogger
from time import sleep
from rich import print as rprint
from rich.traceback import install
from pacstall.api import logger
from pacstall.api.error_codes import ErrorCodes
from pacstall.cmds import download
from pacstall.parser import parse_arguments
def lock(args: Namespace, log: Logger) -> None:
if args.command in ["install", "remove", "upgrade", "repo"]:
lock_file = open("/var/lock/pacstall.lock", "w")
while True:
try:
fcntl.lockf(lock_file, fcntl.LOCK_EX | fcntl.LOCK_NB)
log.debug("Lock acquired")
break
except OSError:
log.warn("Pacstall is already running another instance")
sleep(1)
测试输出
========================================================= test session starts ==========================================================
platform linux -- Python 3.8.10, pytest-6.2.5, py-1.11.0, pluggy-1.0.0
rootdir: /media/pop-os/SBASAK/code/pacstall, configfile: pyproject.toml
plugins: anyio-3.5.0
collected 4 items
tests/test_locking.py FFFF [100%]
=============================================================== FAILURES ===============================================================
_________________________________________ TestLockingCommands.test_if_lock_is_called[install] __________________________________________
self = <tests.test_locking.TestLockingCommands object at 0x7f32944893a0>, command = 'install'
monkeypatch = <_pytest.monkeypatch.MonkeyPatch object at 0x7f32944d7e50>
@pytest.mark.parametrize("command", ["install", "remove", "upgrade", "repo"])
def test_if_lock_is_called(self, command: str, monkeypatch: pytest.MonkeyPatch):
logging.getLogger(__name__).setLevel(logging.DEBUG)
with patch("fcntl.lockf") as mock_lockf:
t1 = Thread(
target=pacstall.__main__.lock,
args=(Namespace(command=command), logging.getLogger(__name__)),
)
t2 = Thread(
target=pacstall.__main__.lock,
args=(Namespace(command=command), logging.getLogger(__name__)),
)
t1.start()
t2.start()
t1.join()
t2.join()
> mock_lockf.assert_called_once()
tests/test_locking.py:50:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
self = <MagicMock name='lockf' id='139855212529552'>
def assert_called_once(self):
"""assert that the mock was called only once.
"""
if not self.call_count == 1:
msg = ("Expected '%s' to have been called once. Called %s times.%s"
% (self._mock_name or 'mock',
self.call_count,
self._calls_repr()))
> raise AssertionError(msg)
E AssertionError: Expected 'lockf' to have been called once. Called 2 times.
E Calls: [call(<_io.TextIOWrapper name='/var/lock/pacstall.lock' mode='w' encoding='UTF-8'>, 6),
E call(<_io.TextIOWrapper name='/var/lock/pacstall.lock' mode='w' encoding='UTF-8'>, 6)].
/usr/lib/python3.8/unittest/mock.py:892: AssertionError
---------------------------------------------------------- Captured log call -----------------------------------------------------------
DEBUG tests.test_locking:__main__.py:50 Lock acquired
DEBUG tests.test_locking:__main__.py:50 Lock acquired
__________________________________________ TestLockingCommands.test_if_lock_is_called[remove] __________________________________________
self = <tests.test_locking.TestLockingCommands object at 0x7f329318f040>, command = 'remove'
monkeypatch = <_pytest.monkeypatch.MonkeyPatch object at 0x7f329318f370>
@pytest.mark.parametrize("command", ["install", "remove", "upgrade", "repo"])
def test_if_lock_is_called(self, command: str, monkeypatch: pytest.MonkeyPatch):
logging.getLogger(__name__).setLevel(logging.DEBUG)
with patch("fcntl.lockf") as mock_lockf:
t1 = Thread(
target=pacstall.__main__.lock,
args=(Namespace(command=command), logging.getLogger(__name__)),
)
t2 = Thread(
target=pacstall.__main__.lock,
args=(Namespace(command=command), logging.getLogger(__name__)),
)
t1.start()
t2.start()
t1.join()
t2.join()
> mock_lockf.assert_called_once()
tests/test_locking.py:50:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
self = <MagicMock name='lockf' id='139855192978336'>
def assert_called_once(self):
"""assert that the mock was called only once.
"""
if not self.call_count == 1:
msg = ("Expected '%s' to have been called once. Called %s times.%s"
% (self._mock_name or 'mock',
self.call_count,
self._calls_repr()))
> raise AssertionError(msg)
E AssertionError: Expected 'lockf' to have been called once. Called 2 times.
E Calls: [call(<_io.TextIOWrapper name='/var/lock/pacstall.lock' mode='w' encoding='UTF-8'>, 6),
E call(<_io.TextIOWrapper name='/var/lock/pacstall.lock' mode='w' encoding='UTF-8'>, 6)].
/usr/lib/python3.8/unittest/mock.py:892: AssertionError
---------------------------------------------------------- Captured log call -----------------------------------------------------------
DEBUG tests.test_locking:__main__.py:50 Lock acquired
DEBUG tests.test_locking:__main__.py:50 Lock acquired
_________________________________________ TestLockingCommands.test_if_lock_is_called[upgrade] __________________________________________
self = <tests.test_locking.TestLockingCommands object at 0x7f329316f4c0>, command = 'upgrade'
monkeypatch = <_pytest.monkeypatch.MonkeyPatch object at 0x7f329316f310>
@pytest.mark.parametrize("command", ["install", "remove", "upgrade", "repo"])
def test_if_lock_is_called(self, command: str, monkeypatch: pytest.MonkeyPatch):
logging.getLogger(__name__).setLevel(logging.DEBUG)
with patch("fcntl.lockf") as mock_lockf:
t1 = Thread(
target=pacstall.__main__.lock,
args=(Namespace(command=command), logging.getLogger(__name__)),
)
t2 = Thread(
target=pacstall.__main__.lock,
args=(Namespace(command=command), logging.getLogger(__name__)),
)
t1.start()
t2.start()
t1.join()
t2.join()
> mock_lockf.assert_called_once()
tests/test_locking.py:50:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
self = <MagicMock name='lockf' id='139855192848512'>
def assert_called_once(self):
"""assert that the mock was called only once.
"""
if not self.call_count == 1:
msg = ("Expected '%s' to have been called once. Called %s times.%s"
% (self._mock_name or 'mock',
self.call_count,
self._calls_repr()))
> raise AssertionError(msg)
E AssertionError: Expected 'lockf' to have been called once. Called 2 times.
E Calls: [call(<_io.TextIOWrapper name='/var/lock/pacstall.lock' mode='w' encoding='UTF-8'>, 6),
E call(<_io.TextIOWrapper name='/var/lock/pacstall.lock' mode='w' encoding='UTF-8'>, 6)].
/usr/lib/python3.8/unittest/mock.py:892: AssertionError
---------------------------------------------------------- Captured log call -----------------------------------------------------------
DEBUG tests.test_locking:__main__.py:50 Lock acquired
DEBUG tests.test_locking:__main__.py:50 Lock acquired
___________________________________________ TestLockingCommands.test_if_lock_is_called[repo] ___________________________________________
self = <tests.test_locking.TestLockingCommands object at 0x7f3293171df0>, command = 'repo'
monkeypatch = <_pytest.monkeypatch.MonkeyPatch object at 0x7f3293171130>
@pytest.mark.parametrize("command", ["install", "remove", "upgrade", "repo"])
def test_if_lock_is_called(self, command: str, monkeypatch: pytest.MonkeyPatch):
logging.getLogger(__name__).setLevel(logging.DEBUG)
with patch("fcntl.lockf") as mock_lockf:
t1 = Thread(
target=pacstall.__main__.lock,
args=(Namespace(command=command), logging.getLogger(__name__)),
)
t2 = Thread(
target=pacstall.__main__.lock,
args=(Namespace(command=command), logging.getLogger(__name__)),
)
t1.start()
t2.start()
t1.join()
t2.join()
> mock_lockf.assert_called_once()
tests/test_locking.py:50:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
self = <MagicMock name='lockf' id='139855192855024'>
def assert_called_once(self):
"""assert that the mock was called only once.
"""
if not self.call_count == 1:
msg = ("Expected '%s' to have been called once. Called %s times.%s"
% (self._mock_name or 'mock',
self.call_count,
self._calls_repr()))
> raise AssertionError(msg)
E AssertionError: Expected 'lockf' to have been called once. Called 2 times.
E Calls: [call(<_io.TextIOWrapper name='/var/lock/pacstall.lock' mode='w' encoding='UTF-8'>, 6),
E call(<_io.TextIOWrapper name='/var/lock/pacstall.lock' mode='w' encoding='UTF-8'>, 6)].
/usr/lib/python3.8/unittest/mock.py:892: AssertionError
---------------------------------------------------------- Captured log call -----------------------------------------------------------
DEBUG tests.test_locking:__main__.py:50 Lock acquired
DEBUG tests.test_locking:__main__.py:50 Lock acquired
======================================================= short test summary info ========================================================
FAILED tests/test_locking.py::TestLockingCommands::test_if_lock_is_called[install] - AssertionError: Expected 'lockf' to have been ca...
FAILED tests/test_locking.py::TestLockingCommands::test_if_lock_is_called[remove] - AssertionError: Expected 'lockf' to have been cal...
FAILED tests/test_locking.py::TestLockingCommands::test_if_lock_is_called[upgrade] - AssertionError: Expected 'lockf' to have been ca...
FAILED tests/test_locking.py::TestLockingCommands::test_if_lock_is_called[repo] - AssertionError: Expected 'lockf' to have been calle...
========================================================== 4 failed in 1.24s ===========================================================
您的代码取决于 lockf 引发 OSError 以修改其控制流的行为 -- 但您模拟了 lockf，因此它不会引发 OSError。实际上，您的 patch 已经修改了代码的行为，因此锁定将始终“成功”，这破坏了您正在尝试测试的内容。
您可以将 side_effect 添加到模拟的 lockf 中，以便它更准确地模拟真实的 lockf，但检查其 call_count 仍然没有用，因为它仍然会被 调用 两次。
测试锁定是否有效的更有用的方法，IMO，将单独留下 lockf，并将 Mock 对象作为 log（而不是真正的记录器）传递)，因此您可以检查模拟记录器的调用并验证 log.debug("Lock acquired") 和 log.warn("Pacstall is already running another instance") 都发生了。
大致如下：
log = Mock()
threads = [
Thread(
target=pacstall.__main__.lock,
args=(Namespace(command=command), log),
) for _ in range(2) # or however many
]
for t in threads:
t.start()
for t in threads:
t.join()
assert log.debug.call_count == 1 # exactly one successful lock
assert log.warn.call_count > 0 # all other locks should fail
我正在编写一个测试来检查我的程序的锁定功能。
对于第一个测试，我试图断言 lockf 函数是否只被调用一次，因为后续的锁定调用而不释放锁会导致错误（这正是我想要的）。
所以，我模拟了 lockf 函数，并断言它被调用了一次。
但是 lockf 函数应该被调用一次却被调用了两次，（第二次应该会出错）。
我在这里做错了什么？我该如何解决这个问题？
我的猜测是 t1 线程在 t2 锁定它之前完成并释放了锁，因此没有错误，lockf 最终被调用了两次。
代码
测试
import logging
from argparse import Namespace
from threading import Thread
from unittest.mock import patch
import pytest
import pacstall.__main__
class TestLockingCommands:
@pytest.mark.parametrize("command", ["install", "remove", "upgrade", "repo"])
def test_if_lock_is_called(self, command: str, monkeypatch: pytest.MonkeyPatch):
logging.getLogger(__name__).setLevel(logging.DEBUG)
with patch("fcntl.lockf") as mock_lockf:
t1 = Thread(
target=pacstall.__main__.lock,
args=(Namespace(command=command), logging.getLogger(__name__)),
)
t2 = Thread(
target=pacstall.__main__.lock,
args=(Namespace(command=command), logging.getLogger(__name__)),
)
t1.start()
t2.start()
t1.join()
t2.join()
mock_lockf.assert_called_once()
生产代码
import fcntl
import sys
from argparse import Namespace
from asyncio import run
from getpass import getuser
from logging import Logger, getLogger
from time import sleep
from rich import print as rprint
from rich.traceback import install
from pacstall.api import logger
from pacstall.api.error_codes import ErrorCodes
from pacstall.cmds import download
from pacstall.parser import parse_arguments
def lock(args: Namespace, log: Logger) -> None:
if args.command in ["install", "remove", "upgrade", "repo"]:
lock_file = open("/var/lock/pacstall.lock", "w")
while True:
try:
fcntl.lockf(lock_file, fcntl.LOCK_EX | fcntl.LOCK_NB)
log.debug("Lock acquired")
break
except OSError:
log.warn("Pacstall is already running another instance")
sleep(1)
测试输出
========================================================= test session starts ==========================================================
platform linux -- Python 3.8.10, pytest-6.2.5, py-1.11.0, pluggy-1.0.0
rootdir: /media/pop-os/SBASAK/code/pacstall, configfile: pyproject.toml
plugins: anyio-3.5.0
collected 4 items
tests/test_locking.py FFFF [100%]
=============================================================== FAILURES ===============================================================
_________________________________________ TestLockingCommands.test_if_lock_is_called[install] __________________________________________
self = <tests.test_locking.TestLockingCommands object at 0x7f32944893a0>, command = 'install'
monkeypatch = <_pytest.monkeypatch.MonkeyPatch object at 0x7f32944d7e50>
@pytest.mark.parametrize("command", ["install", "remove", "upgrade", "repo"])
def test_if_lock_is_called(self, command: str, monkeypatch: pytest.MonkeyPatch):
logging.getLogger(__name__).setLevel(logging.DEBUG)
with patch("fcntl.lockf") as mock_lockf:
t1 = Thread(
target=pacstall.__main__.lock,
args=(Namespace(command=command), logging.getLogger(__name__)),
)
t2 = Thread(
target=pacstall.__main__.lock,
args=(Namespace(command=command), logging.getLogger(__name__)),
)
t1.start()
t2.start()
t1.join()
t2.join()
> mock_lockf.assert_called_once()
tests/test_locking.py:50:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
self = <MagicMock name='lockf' id='139855212529552'>
def assert_called_once(self):
"""assert that the mock was called only once.
"""
if not self.call_count == 1:
msg = ("Expected '%s' to have been called once. Called %s times.%s"
% (self._mock_name or 'mock',
self.call_count,
self._calls_repr()))
> raise AssertionError(msg)
E AssertionError: Expected 'lockf' to have been called once. Called 2 times.
E Calls: [call(<_io.TextIOWrapper name='/var/lock/pacstall.lock' mode='w' encoding='UTF-8'>, 6),
E call(<_io.TextIOWrapper name='/var/lock/pacstall.lock' mode='w' encoding='UTF-8'>, 6)].
/usr/lib/python3.8/unittest/mock.py:892: AssertionError
---------------------------------------------------------- Captured log call -----------------------------------------------------------
DEBUG tests.test_locking:__main__.py:50 Lock acquired
DEBUG tests.test_locking:__main__.py:50 Lock acquired
__________________________________________ TestLockingCommands.test_if_lock_is_called[remove] __________________________________________
self = <tests.test_locking.TestLockingCommands object at 0x7f329318f040>, command = 'remove'
monkeypatch = <_pytest.monkeypatch.MonkeyPatch object at 0x7f329318f370>
@pytest.mark.parametrize("command", ["install", "remove", "upgrade", "repo"])
def test_if_lock_is_called(self, command: str, monkeypatch: pytest.MonkeyPatch):
logging.getLogger(__name__).setLevel(logging.DEBUG)
with patch("fcntl.lockf") as mock_lockf:
t1 = Thread(
target=pacstall.__main__.lock,
args=(Namespace(command=command), logging.getLogger(__name__)),
)
t2 = Thread(
target=pacstall.__main__.lock,
args=(Namespace(command=command), logging.getLogger(__name__)),
)
t1.start()
t2.start()
t1.join()
t2.join()
> mock_lockf.assert_called_once()
tests/test_locking.py:50:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
self = <MagicMock name='lockf' id='139855192978336'>
def assert_called_once(self):
"""assert that the mock was called only once.
"""
if not self.call_count == 1:
msg = ("Expected '%s' to have been called once. Called %s times.%s"
% (self._mock_name or 'mock',
self.call_count,
self._calls_repr()))
> raise AssertionError(msg)
E AssertionError: Expected 'lockf' to have been called once. Called 2 times.
E Calls: [call(<_io.TextIOWrapper name='/var/lock/pacstall.lock' mode='w' encoding='UTF-8'>, 6),
E call(<_io.TextIOWrapper name='/var/lock/pacstall.lock' mode='w' encoding='UTF-8'>, 6)].
/usr/lib/python3.8/unittest/mock.py:892: AssertionError
---------------------------------------------------------- Captured log call -----------------------------------------------------------
DEBUG tests.test_locking:__main__.py:50 Lock acquired
DEBUG tests.test_locking:__main__.py:50 Lock acquired
_________________________________________ TestLockingCommands.test_if_lock_is_called[upgrade] __________________________________________
self = <tests.test_locking.TestLockingCommands object at 0x7f329316f4c0>, command = 'upgrade'
monkeypatch = <_pytest.monkeypatch.MonkeyPatch object at 0x7f329316f310>
@pytest.mark.parametrize("command", ["install", "remove", "upgrade", "repo"])
def test_if_lock_is_called(self, command: str, monkeypatch: pytest.MonkeyPatch):
logging.getLogger(__name__).setLevel(logging.DEBUG)
with patch("fcntl.lockf") as mock_lockf:
t1 = Thread(
target=pacstall.__main__.lock,
args=(Namespace(command=command), logging.getLogger(__name__)),
)
t2 = Thread(
target=pacstall.__main__.lock,
args=(Namespace(command=command), logging.getLogger(__name__)),
)
t1.start()
t2.start()
t1.join()
t2.join()
> mock_lockf.assert_called_once()
tests/test_locking.py:50:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
self = <MagicMock name='lockf' id='139855192848512'>
def assert_called_once(self):
"""assert that the mock was called only once.
"""
if not self.call_count == 1:
msg = ("Expected '%s' to have been called once. Called %s times.%s"
% (self._mock_name or 'mock',
self.call_count,
self._calls_repr()))
> raise AssertionError(msg)
E AssertionError: Expected 'lockf' to have been called once. Called 2 times.
E Calls: [call(<_io.TextIOWrapper name='/var/lock/pacstall.lock' mode='w' encoding='UTF-8'>, 6),
E call(<_io.TextIOWrapper name='/var/lock/pacstall.lock' mode='w' encoding='UTF-8'>, 6)].
/usr/lib/python3.8/unittest/mock.py:892: AssertionError
---------------------------------------------------------- Captured log call -----------------------------------------------------------
DEBUG tests.test_locking:__main__.py:50 Lock acquired
DEBUG tests.test_locking:__main__.py:50 Lock acquired
___________________________________________ TestLockingCommands.test_if_lock_is_called[repo] ___________________________________________
self = <tests.test_locking.TestLockingCommands object at 0x7f3293171df0>, command = 'repo'
monkeypatch = <_pytest.monkeypatch.MonkeyPatch object at 0x7f3293171130>
@pytest.mark.parametrize("command", ["install", "remove", "upgrade", "repo"])
def test_if_lock_is_called(self, command: str, monkeypatch: pytest.MonkeyPatch):
logging.getLogger(__name__).setLevel(logging.DEBUG)
with patch("fcntl.lockf") as mock_lockf:
t1 = Thread(
target=pacstall.__main__.lock,
args=(Namespace(command=command), logging.getLogger(__name__)),
)
t2 = Thread(
target=pacstall.__main__.lock,
args=(Namespace(command=command), logging.getLogger(__name__)),
)
t1.start()
t2.start()
t1.join()
t2.join()
> mock_lockf.assert_called_once()
tests/test_locking.py:50:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
self = <MagicMock name='lockf' id='139855192855024'>
def assert_called_once(self):
"""assert that the mock was called only once.
"""
if not self.call_count == 1:
msg = ("Expected '%s' to have been called once. Called %s times.%s"
% (self._mock_name or 'mock',
self.call_count,
self._calls_repr()))
> raise AssertionError(msg)
E AssertionError: Expected 'lockf' to have been called once. Called 2 times.
E Calls: [call(<_io.TextIOWrapper name='/var/lock/pacstall.lock' mode='w' encoding='UTF-8'>, 6),
E call(<_io.TextIOWrapper name='/var/lock/pacstall.lock' mode='w' encoding='UTF-8'>, 6)].
/usr/lib/python3.8/unittest/mock.py:892: AssertionError
---------------------------------------------------------- Captured log call -----------------------------------------------------------
DEBUG tests.test_locking:__main__.py:50 Lock acquired
DEBUG tests.test_locking:__main__.py:50 Lock acquired
======================================================= short test summary info ========================================================
FAILED tests/test_locking.py::TestLockingCommands::test_if_lock_is_called[install] - AssertionError: Expected 'lockf' to have been ca...
FAILED tests/test_locking.py::TestLockingCommands::test_if_lock_is_called[remove] - AssertionError: Expected 'lockf' to have been cal...
FAILED tests/test_locking.py::TestLockingCommands::test_if_lock_is_called[upgrade] - AssertionError: Expected 'lockf' to have been ca...
FAILED tests/test_locking.py::TestLockingCommands::test_if_lock_is_called[repo] - AssertionError: Expected 'lockf' to have been calle...
========================================================== 4 failed in 1.24s ===========================================================
您的代码取决于 lockf 引发 OSError 以修改其控制流的行为 -- 但您模拟了 lockf，因此它不会引发 OSError。实际上，您的 patch 已经修改了代码的行为，因此锁定将始终“成功”，这破坏了您正在尝试测试的内容。
您可以将 side_effect 添加到模拟的 lockf 中，以便它更准确地模拟真实的 lockf，但检查其 call_count 仍然没有用，因为它仍然会被 调用 两次。
测试锁定是否有效的更有用的方法，IMO，将单独留下 lockf，并将 Mock 对象作为 log（而不是真正的记录器）传递)，因此您可以检查模拟记录器的调用并验证 log.debug("Lock acquired") 和 log.warn("Pacstall is already running another instance") 都发生了。
大致如下：
log = Mock()
threads = [
Thread(
target=pacstall.__main__.lock,
args=(Namespace(command=command), log),
) for _ in range(2) # or however many
]
for t in threads:
t.start()
for t in threads:
t.join()
assert log.debug.call_count == 1 # exactly one successful lock
assert log.warn.call_count > 0 # all other locks should fail