mirror of
https://github.com/Stone-Red-Code/dvc.git
synced 2026-11-02 07:18:25 +01:00
54 lines
1.6 KiB
Python
54 lines
1.6 KiB
Python
import pytest
|
|
|
|
from dvc.cli import main
|
|
from dvc.exceptions import DvcException
|
|
from dvc.lock import Lock, LockError
|
|
|
|
|
|
def test_with(tmp_dir, dvc, mocker):
|
|
# patching to speedup tests
|
|
mocker.patch("dvc.lock.DEFAULT_TIMEOUT", 0.01)
|
|
|
|
lockfile = tmp_dir / dvc.tmp_dir / "lock"
|
|
with Lock(lockfile):
|
|
with pytest.raises(LockError), Lock(lockfile):
|
|
pass
|
|
|
|
|
|
def test_unlock_lock_failed(tmp_dir, dvc, mocker):
|
|
# patching to speedup tests
|
|
mocker.patch("dvc.lock.DEFAULT_TIMEOUT", 0.01)
|
|
|
|
lockfile = tmp_dir / dvc.tmp_dir / "lock"
|
|
lock = Lock(lockfile)
|
|
lock_ext = Lock(lockfile)
|
|
|
|
# It's a common scenario now to have lock unlocked and locked back (e.g. in
|
|
# repro of a stage) in with. We should see LockError exception here.
|
|
with lock:
|
|
lock.unlock()
|
|
lock_ext.lock() # imitate an exernal process had time to lock it
|
|
with pytest.raises(LockError):
|
|
lock.lock()
|
|
|
|
|
|
def test_unlock_unlocked_raises():
|
|
lock = Lock("lock")
|
|
with pytest.raises(DvcException, match="Unlock called on an unlocked lock"):
|
|
lock.unlock()
|
|
|
|
|
|
def test_cli(tmp_dir, dvc, mocker, caplog):
|
|
# patching to speedup tests
|
|
mocker.patch("dvc.lock.DEFAULT_TIMEOUT", 0.01)
|
|
|
|
expected_error_msg = (
|
|
"Unable to acquire lock. Most likely another DVC process is "
|
|
"running or was terminated abruptly. Check the page "
|
|
"<https://dvc.org/doc/user-guide/troubleshooting#lock-issue> "
|
|
"for other possible reasons and to learn how to resolve this."
|
|
)
|
|
with Lock(tmp_dir / dvc.tmp_dir / "lock"):
|
|
assert main(["add", "foo"]) == 1
|
|
assert expected_error_msg in caplog.text
|