'Mocked `lockf` ends up getting called twice when it should be called once
I'm writing a test to check my program's locking functionality.
For the first test I'm trying to assert if the lockf function is getting called only once, as the subsequent locking calls without releasing the lock would result in errors (which is what I want).
So, I have mocked the lockf function, and am asserting that it's getting called once.
But the lockf function ends up getting called twice when it should be called once, (it should error out the second time).
What am I doing wrong here? And how do I fix this?
My guess is that the t1 thread finishes and releases the lock before t2 locks it, so there are no errors and lockf ends up getting called twice.
Code
Test
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()
Production code
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 output
========================================================= 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 ===========================================================
Sources
This article follows the attribution requirements of Stack Overflow and is licensed under CC BY-SA 3.0.
Source: Stack Overflow
| Solution | Source |
|---|