mirror of
https://github.com/Stone-Red-Code/dvc.git
synced 2026-11-01 23:08:43 +01:00
migrate test_checkout.py to pytest
This commit is contained in:
Saugat Pachhai (सौगात)
committed by
skshetry
skshetry
parent
1cad8ea7fd
commit
f9b0de883f
+305
-409
@@ -1,16 +1,13 @@
|
||||
import collections
|
||||
import filecmp
|
||||
import logging
|
||||
import os
|
||||
import shutil
|
||||
import stat
|
||||
import textwrap
|
||||
from unittest.mock import patch
|
||||
|
||||
import pytest
|
||||
|
||||
from dvc.cli import main
|
||||
from dvc.dvcfile import DVC_FILE_SUFFIX, PIPELINE_FILE, Dvcfile
|
||||
from dvc.dvcfile import PIPELINE_FILE, Dvcfile
|
||||
from dvc.exceptions import (
|
||||
CheckoutError,
|
||||
CheckoutErrorSuggestGit,
|
||||
@@ -24,8 +21,7 @@ from dvc.stage.exceptions import StageFileDoesNotExistError
|
||||
from dvc.utils import relpath
|
||||
from dvc.utils.fs import remove
|
||||
from dvc.utils.serialize import dump_yaml, load_yaml
|
||||
from tests.basic_env import TestDvc, TestDvcGit
|
||||
from tests.func.test_repro import TestRepro
|
||||
from tests.utils import get_gitignore_content
|
||||
|
||||
logger = logging.getLogger("dvc")
|
||||
|
||||
@@ -36,465 +32,365 @@ def walk_files(directory):
|
||||
yield os.path.join(root, f)
|
||||
|
||||
|
||||
class TestCheckout(TestRepro):
|
||||
def setUp(self):
|
||||
super().setUp()
|
||||
def test_checkout(tmp_dir, dvc, copy_script):
|
||||
tmp_dir.dvc_gen({"foo": "foo", "data": {"file": "file"}})
|
||||
dvc.run(
|
||||
fname="file1.dvc",
|
||||
outs=["file1"],
|
||||
deps=["foo", "copy.py"],
|
||||
cmd="python copy.py foo file1",
|
||||
single_stage=True,
|
||||
)
|
||||
remove(tmp_dir / "foo")
|
||||
remove("data")
|
||||
|
||||
stages = self.dvc.add(self.DATA_DIR)
|
||||
self.assertEqual(len(stages), 1)
|
||||
self.data_dir_stage = stages[0]
|
||||
self.assertTrue(self.data_dir_stage is not None)
|
||||
|
||||
self.orig = "orig"
|
||||
shutil.copy(self.FOO, self.orig)
|
||||
os.unlink(self.FOO)
|
||||
|
||||
self.orig_dir = "orig_dir"
|
||||
shutil.copytree(self.DATA_DIR, self.orig_dir)
|
||||
shutil.rmtree(self.DATA_DIR)
|
||||
|
||||
def test(self):
|
||||
self.dvc.checkout(force=True)
|
||||
self._test_checkout()
|
||||
|
||||
def _test_checkout(self):
|
||||
self.assertTrue(os.path.isfile(self.FOO))
|
||||
self.assertTrue(filecmp.cmp(self.FOO, self.orig, shallow=False))
|
||||
dvc.checkout(force=True)
|
||||
assert (tmp_dir / "foo").read_text() == "foo"
|
||||
assert (tmp_dir / "data").read_text() == {"file": "file"}
|
||||
|
||||
|
||||
class TestCheckoutSingleStage(TestCheckout):
|
||||
def test(self):
|
||||
ret = main(["checkout", "--force", self.foo_stage.path])
|
||||
self.assertEqual(ret, 0)
|
||||
def test_checkout_cli(tmp_dir, dvc, copy_script):
|
||||
tmp_dir.dvc_gen({"foo": "foo", "data": {"file": "file"}})
|
||||
dvc.run(
|
||||
fname="file1.dvc",
|
||||
outs=["file1"],
|
||||
deps=["foo", "copy.py"],
|
||||
cmd="python copy.py foo file1",
|
||||
single_stage=True,
|
||||
)
|
||||
remove(tmp_dir / "foo")
|
||||
remove("data")
|
||||
|
||||
ret = main(["checkout", "--force", self.data_dir_stage.path])
|
||||
self.assertEqual(ret, 0)
|
||||
assert main(["checkout", "--force"]) == 0
|
||||
assert (tmp_dir / "foo").read_text() == "foo"
|
||||
assert (tmp_dir / "data").read_text() == {"file": "file"}
|
||||
|
||||
self._test_checkout()
|
||||
remove(tmp_dir / "foo")
|
||||
remove("data")
|
||||
|
||||
assert main(["checkout", "--force", "foo.dvc"]) == 0
|
||||
assert main(["checkout", "--force", "data.dvc"]) == 0
|
||||
assert (tmp_dir / "foo").read_text() == "foo"
|
||||
assert (tmp_dir / "data").read_text() == {"file": "file"}
|
||||
|
||||
|
||||
class TestCheckoutCorruptedCacheFile(TestRepro):
|
||||
def test(self):
|
||||
cache = self.foo_stage.outs[0].cache_path
|
||||
def test_checkout_corrupted_cache_file(tmp_dir, dvc):
|
||||
(foo_stage,) = tmp_dir.dvc_gen("foo", "foo")
|
||||
cache = foo_stage.outs[0].cache_path
|
||||
|
||||
os.chmod(cache, 0o644)
|
||||
with open(cache, "a", encoding="utf-8") as fd:
|
||||
fd.write("1")
|
||||
os.chmod(cache, 0o644)
|
||||
with open(cache, "a", encoding="utf-8") as fd:
|
||||
fd.write("1")
|
||||
|
||||
with pytest.raises(CheckoutError):
|
||||
self.dvc.checkout(force=True)
|
||||
with pytest.raises(CheckoutError):
|
||||
dvc.checkout(force=True)
|
||||
|
||||
self.assertFalse(os.path.isfile(self.FOO))
|
||||
self.assertFalse(os.path.isfile(cache))
|
||||
assert not os.path.isfile("foo")
|
||||
assert not os.path.isfile(cache)
|
||||
|
||||
|
||||
class TestCheckoutCorruptedCacheDir(TestDvc):
|
||||
def test(self):
|
||||
from dvc_data.hashfile import load
|
||||
def test_checkout_corrupted_cache_dir(tmp_dir, dvc):
|
||||
from dvc_data.hashfile import load
|
||||
|
||||
# NOTE: using 'copy' so that cache and link don't have same inode
|
||||
ret = main(["config", "cache.type", "copy"])
|
||||
self.assertEqual(ret, 0)
|
||||
tmp_dir.gen("data", {"foo": "foo", "bar": "bar"})
|
||||
# NOTE: using 'copy' so that cache and link don't have same inode
|
||||
ret = main(["config", "cache.type", "copy"])
|
||||
assert ret == 0
|
||||
|
||||
self.dvc.config.load()
|
||||
dvc.config.load()
|
||||
|
||||
stages = self.dvc.add(self.DATA_DIR)
|
||||
self.assertEqual(len(stages), 1)
|
||||
self.assertEqual(len(stages[0].outs), 1)
|
||||
out = stages[0].outs[0]
|
||||
stages = dvc.add("data")
|
||||
assert len(stages) == 1
|
||||
assert len(stages[0].outs) == 1
|
||||
out = stages[0].outs[0]
|
||||
|
||||
# NOTE: modifying cache file for one of the files inside the directory
|
||||
# to check if dvc will detect that the cache is corrupted.
|
||||
obj = load(self.dvc.odb.local, out.hash_info)
|
||||
_, _, entry_oid = list(obj)[0]
|
||||
cache = self.dvc.odb.local.oid_to_path(entry_oid.value)
|
||||
# NOTE: modifying cache file for one of the files inside the directory
|
||||
# to check if dvc will detect that the cache is corrupted.
|
||||
obj = load(dvc.odb.local, out.hash_info)
|
||||
_, _, entry_oid = list(obj)[0]
|
||||
cache = dvc.odb.local.oid_to_path(entry_oid.value)
|
||||
|
||||
os.chmod(cache, 0o644)
|
||||
with open(cache, "w+", encoding="utf-8") as fobj:
|
||||
fobj.write("1")
|
||||
os.chmod(cache, 0o644)
|
||||
with open(cache, "w+", encoding="utf-8") as fobj:
|
||||
fobj.write("1")
|
||||
|
||||
with pytest.raises(CheckoutError):
|
||||
self.dvc.checkout(force=True)
|
||||
with pytest.raises(CheckoutError):
|
||||
dvc.checkout(force=True)
|
||||
|
||||
self.assertFalse(os.path.exists(cache))
|
||||
assert not os.path.exists(cache)
|
||||
|
||||
|
||||
class TestCmdCheckout(TestCheckout):
|
||||
def test(self):
|
||||
ret = main(["checkout", "--force"])
|
||||
self.assertEqual(ret, 0)
|
||||
self._test_checkout()
|
||||
def test_remove_files_when_checkout(tmp_dir, dvc, scm):
|
||||
# add the file into a separate branch
|
||||
scm.checkout("branch", True)
|
||||
ret = main(["checkout", "--force"])
|
||||
assert ret == 0
|
||||
tmp_dir.dvc_gen("file_in_a_branch", "random text", commit="add file")
|
||||
|
||||
# Checkout back in master
|
||||
scm.checkout("master")
|
||||
assert os.path.exists("file_in_a_branch")
|
||||
|
||||
# Make sure `dvc checkout` removes the file
|
||||
# self.dvc.checkout()
|
||||
ret = main(["checkout", "--force"])
|
||||
assert ret == 0
|
||||
assert not os.path.exists("file_in_a_branch")
|
||||
|
||||
|
||||
class CheckoutBase(TestDvcGit):
|
||||
GIT_IGNORE = ".gitignore"
|
||||
|
||||
def commit_data_file(self, fname, content="random text"):
|
||||
with open(fname, "w", encoding="utf-8") as fd:
|
||||
fd.write(content)
|
||||
stages = self.dvc.add(fname)
|
||||
self.assertEqual(len(stages), 1)
|
||||
self.assertTrue(stages[0] is not None)
|
||||
self.dvc.scm.add([fname + ".dvc", ".gitignore"])
|
||||
self.dvc.scm.commit("adding " + fname)
|
||||
|
||||
def read_ignored(self):
|
||||
with open(self.GIT_IGNORE, encoding="utf-8") as f:
|
||||
return [s.strip("\n") for s in f.readlines()]
|
||||
|
||||
def outs_info(self, stage):
|
||||
FileInfo = collections.namedtuple("FileInfo", "path inode")
|
||||
|
||||
paths = [
|
||||
path
|
||||
for output in stage["outs"]
|
||||
for path in self.dvc.fs.find(output["path"])
|
||||
]
|
||||
|
||||
return [
|
||||
FileInfo(path=path, inode=system.inode(path)) for path in paths
|
||||
]
|
||||
|
||||
|
||||
class TestRemoveFilesWhenCheckout(CheckoutBase):
|
||||
def test(self):
|
||||
fname = "file_in_a_branch"
|
||||
branch_master = "master"
|
||||
branch_1 = "b1"
|
||||
|
||||
self.dvc.scm.add(self.dvc.scm.untracked_files())
|
||||
self.dvc.scm.commit("add all files")
|
||||
|
||||
# add the file into a separate branch
|
||||
self.dvc.scm.checkout(branch_1, True)
|
||||
ret = main(["checkout", "--force"])
|
||||
self.assertEqual(ret, 0)
|
||||
self.commit_data_file(fname)
|
||||
|
||||
# Checkout back in master
|
||||
self.dvc.scm.checkout(branch_master)
|
||||
self.assertTrue(os.path.exists(fname))
|
||||
|
||||
# Make sure `dvc checkout` removes the file
|
||||
# self.dvc.checkout()
|
||||
ret = main(["checkout", "--force"])
|
||||
self.assertEqual(ret, 0)
|
||||
self.assertFalse(os.path.exists(fname))
|
||||
|
||||
|
||||
class TestCheckoutCleanWorkingDir(CheckoutBase):
|
||||
@patch("dvc.prompt.confirm", return_value=True)
|
||||
def test(self, mock_prompt):
|
||||
mock_prompt.return_value = True
|
||||
|
||||
stages = self.dvc.add(self.DATA_DIR)
|
||||
stage = stages[0]
|
||||
|
||||
working_dir_change = os.path.join(self.DATA_DIR, "not_cached.txt")
|
||||
with open(working_dir_change, "w", encoding="utf-8") as f:
|
||||
f.write("not_cached")
|
||||
|
||||
ret = main(["checkout", stage.relpath])
|
||||
self.assertEqual(ret, 0)
|
||||
self.assertFalse(os.path.exists(working_dir_change))
|
||||
|
||||
@patch("dvc.prompt.confirm", return_value=False)
|
||||
def test_force(self, mock_prompt):
|
||||
mock_prompt.return_value = False
|
||||
|
||||
stages = self.dvc.add(self.DATA_DIR)
|
||||
self.assertEqual(len(stages), 1)
|
||||
stage = stages[0]
|
||||
|
||||
working_dir_change = os.path.join(self.DATA_DIR, "not_cached.txt")
|
||||
with open(working_dir_change, "w", encoding="utf-8") as f:
|
||||
f.write("not_cached")
|
||||
|
||||
ret = main(["checkout", stage.relpath])
|
||||
self.assertNotEqual(ret, 0)
|
||||
class TestCheckoutCleanWorkingDir:
|
||||
def test(self, mocker, tmp_dir, dvc): # pylint: disable=unused-argument
|
||||
mock_prompt = mocker.patch("dvc.prompt.confirm", return_value=True)
|
||||
(stage,) = tmp_dir.dvc_gen("data", {"foo": "foo"})
|
||||
|
||||
# change working directory
|
||||
(tmp_dir / "data").gen("not_cached.txt", "not_cached")
|
||||
assert main(["checkout", stage.relpath]) == 0
|
||||
mock_prompt.assert_called()
|
||||
self.assertNotEqual(ret, 0)
|
||||
self.assertRaises(DvcException)
|
||||
assert not (tmp_dir / "data" / "not_cached.txt").exists()
|
||||
|
||||
def test_force(self, mocker, tmp_dir, dvc):
|
||||
mock_prompt = mocker.patch("dvc.prompt.confirm", return_value=False)
|
||||
(stage,) = tmp_dir.dvc_gen("data", {"foo": "foo"})
|
||||
|
||||
# change working directory
|
||||
(tmp_dir / "data").gen("not_cached.txt", "not_cached")
|
||||
assert main(["checkout", stage.relpath]) != 0
|
||||
mock_prompt.assert_called()
|
||||
assert (tmp_dir / "data" / "not_cached.txt").exists()
|
||||
|
||||
|
||||
def test_checkout_selective_remove(tmp_dir, dvc):
|
||||
# Use copy to test for changes in the inodes
|
||||
dvc.odb.local.cache_types = ["copy"]
|
||||
tmp_dir.dvc_gen({"data": {"foo": "foo", "bar": "bar"}})
|
||||
|
||||
foo_inode = system.inode(os.path.join("data", "foo"))
|
||||
bar_inode = system.inode(os.path.join("data", "bar"))
|
||||
# move instead of remove, to lock inode assigned to stage_files[0].path
|
||||
# if we were to use remove, we might end up with same inode assigned to
|
||||
# newly checked out file
|
||||
shutil.move(os.path.join("data", "foo"), "random_name")
|
||||
|
||||
assert main(["checkout", "--force", "data.dvc"]) == 0
|
||||
assert (tmp_dir / "data").read_text() == {"foo": "foo", "bar": "bar"}
|
||||
assert system.inode(os.path.join("data", "foo")) != foo_inode
|
||||
assert system.inode(os.path.join("data", "bar")) == bar_inode
|
||||
|
||||
|
||||
def test_gitignore_basic(tmp_dir, dvc, scm):
|
||||
tmp_dir.gen("foo", "foo")
|
||||
assert not os.path.exists(scm.GITIGNORE)
|
||||
|
||||
tmp_dir.dvc_gen("file1", "random text1", commit="add file1")
|
||||
tmp_dir.dvc_gen("file2", "random text2", commit="add file2")
|
||||
dvc.run(
|
||||
single_stage=True,
|
||||
cmd="cp foo file3",
|
||||
deps=["foo"],
|
||||
outs_no_cache=["file3"],
|
||||
)
|
||||
assert get_gitignore_content() == ["/file1", "/file2"]
|
||||
|
||||
|
||||
def test_gitignore_when_checkout(tmp_dir, dvc, scm):
|
||||
tmp_dir.dvc_gen("file_in_a_master", "master", commit="master")
|
||||
|
||||
class TestCheckoutSelectiveRemove(CheckoutBase):
|
||||
def test(self):
|
||||
# Use copy to test for changes in the inodes
|
||||
ret = main(["config", "cache.type", "copy"])
|
||||
self.assertEqual(ret, 0)
|
||||
scm.checkout("branch", True)
|
||||
ret = main(["checkout", "--force"])
|
||||
assert ret == 0
|
||||
tmp_dir.dvc_gen("file_in_a_branch", "branch", commit="branch")
|
||||
|
||||
scm.checkout("master")
|
||||
ret = main(["checkout", "--force"])
|
||||
assert ret == 0
|
||||
|
||||
ignored = get_gitignore_content()
|
||||
|
||||
assert len(ignored) == 1
|
||||
assert "/file_in_a_master" in ignored
|
||||
|
||||
ret = main(["add", self.DATA_DIR])
|
||||
self.assertEqual(0, ret)
|
||||
scm.checkout("branch")
|
||||
ret = main(["checkout", "--force"])
|
||||
assert ret == 0
|
||||
ignored = get_gitignore_content()
|
||||
assert "/file_in_a_branch" in ignored
|
||||
|
||||
|
||||
def test_checkout_missing_md5_in_stage_file(tmp_dir, dvc, copy_script):
|
||||
tmp_dir.dvc_gen({"foo": "foo", "data": {"file": "file"}})
|
||||
stage = dvc.run(
|
||||
fname="file1.dvc",
|
||||
outs=["file1"],
|
||||
deps=["foo", "copy.py"],
|
||||
cmd="python copy.py foo file1",
|
||||
single_stage=True,
|
||||
)
|
||||
d = load_yaml(stage.relpath)
|
||||
del d[Stage.PARAM_OUTS][0][LocalFileSystem.PARAM_CHECKSUM]
|
||||
del d[Stage.PARAM_DEPS][0][LocalFileSystem.PARAM_CHECKSUM]
|
||||
dump_yaml(stage.relpath, d)
|
||||
|
||||
with pytest.raises(CheckoutError):
|
||||
dvc.checkout(force=True)
|
||||
|
||||
|
||||
def test_checkout_empty_dir(tmp_dir, dvc):
|
||||
empty_dir = tmp_dir / "empty_dir"
|
||||
empty_dir.mkdir()
|
||||
(stage,) = dvc.add("empty_dir")
|
||||
|
||||
stage.outs[0].remove()
|
||||
assert not empty_dir.exists()
|
||||
|
||||
stats = dvc.checkout(force=True)
|
||||
assert stats["added"] == [os.path.join("empty_dir", "")]
|
||||
assert empty_dir.is_dir()
|
||||
assert not list(empty_dir.iterdir())
|
||||
|
||||
|
||||
def test_checkout_not_cached_file(tmp_dir, dvc):
|
||||
tmp_dir.dvc_gen("foo", "foo")
|
||||
dvc.run(
|
||||
cmd="cp foo bar",
|
||||
deps=["foo"],
|
||||
outs_no_cache=["bar"],
|
||||
single_stage=True,
|
||||
)
|
||||
stats = dvc.checkout(force=True)
|
||||
assert not any(stats.values())
|
||||
|
||||
|
||||
def test_checkout_with_deps_cli(tmp_dir, dvc, copy_script):
|
||||
tmp_dir.dvc_gen({"foo": "foo", "data": {"file": "file"}})
|
||||
dvc.run(
|
||||
fname="file1.dvc",
|
||||
outs=["file1"],
|
||||
deps=["foo", "copy.py"],
|
||||
cmd="python copy.py foo file1",
|
||||
single_stage=True,
|
||||
)
|
||||
remove("foo")
|
||||
remove("file1")
|
||||
|
||||
assert not os.path.exists("foo")
|
||||
assert not os.path.exists("file1")
|
||||
|
||||
stage_path = self.DATA_DIR + DVC_FILE_SUFFIX
|
||||
stage = load_yaml(stage_path)
|
||||
staged_files = self.outs_info(stage)
|
||||
ret = main(["checkout", "--force", "file1.dvc", "--with-deps"])
|
||||
assert ret == 0
|
||||
|
||||
assert os.path.exists("foo")
|
||||
assert os.path.exists("file1")
|
||||
|
||||
# move instead of remove, to lock inode assigned to stage_files[0].path
|
||||
# if we were to use remove, we might end up with same inode assigned to
|
||||
# newly checked out file
|
||||
shutil.move(staged_files[0].path, "random_name")
|
||||
|
||||
ret = main(["checkout", "--force", stage_path])
|
||||
self.assertEqual(ret, 0)
|
||||
def test_checkout_directory(tmp_dir, dvc):
|
||||
(stage,) = tmp_dir.dvc_gen({"data": {"foo": "foo", "bar": "bar"}})
|
||||
|
||||
remove("data")
|
||||
assert not os.path.exists("data")
|
||||
|
||||
checkedout_files = self.outs_info(stage)
|
||||
ret = main(["checkout", stage.path])
|
||||
assert ret == 0
|
||||
|
||||
self.assertEqual(len(staged_files), len(checkedout_files))
|
||||
self.assertEqual(staged_files[0].path, checkedout_files[0].path)
|
||||
self.assertNotEqual(staged_files[0].inode, checkedout_files[0].inode)
|
||||
self.assertEqual(staged_files[1].inode, checkedout_files[1].inode)
|
||||
assert os.path.exists("data")
|
||||
|
||||
|
||||
class TestGitIgnoreBasic(CheckoutBase):
|
||||
def test(self):
|
||||
fname1 = "file_1"
|
||||
fname2 = "file_2"
|
||||
fname3 = "file_3"
|
||||
def test_checkout_hook(mocker, tmp_dir, dvc):
|
||||
"""Test that dvc checkout handles EOFError gracefully, which is what
|
||||
it will experience when running in a git hook.
|
||||
"""
|
||||
tmp_dir.dvc_gen({"data": {"foo": "foo"}})
|
||||
mocker.patch("sys.stdout.isatty", return_value=True)
|
||||
mocker.patch("dvc.prompt.input", side_effect=EOFError)
|
||||
|
||||
self.dvc.scm.add(self.dvc.scm.untracked_files())
|
||||
self.dvc.scm.commit("add all files")
|
||||
(tmp_dir / "data").gen("test", "test")
|
||||
with pytest.raises(ConfirmRemoveError):
|
||||
dvc.checkout()
|
||||
|
||||
self.assertFalse(os.path.exists(self.GIT_IGNORE))
|
||||
|
||||
self.commit_data_file(fname1)
|
||||
self.commit_data_file(fname2)
|
||||
self.dvc.run(
|
||||
single_stage=True,
|
||||
cmd=f"python {self.CODE} {self.FOO} {fname3}",
|
||||
deps=[self.CODE, self.FOO],
|
||||
outs_no_cache=[fname3],
|
||||
)
|
||||
def test_checkout_suggest_git(tmp_dir, dvc, scm):
|
||||
# pylint: disable=no-member
|
||||
tmp_dir.dvc_gen("foo", "foo")
|
||||
try:
|
||||
dvc.checkout(targets="gitbranch")
|
||||
except DvcException as exc:
|
||||
assert isinstance(exc, CheckoutErrorSuggestGit)
|
||||
assert isinstance(exc.__cause__, NoOutputOrStageError)
|
||||
assert isinstance(exc.__cause__.__cause__, StageFileDoesNotExistError)
|
||||
|
||||
try:
|
||||
dvc.checkout(targets="foo")
|
||||
except DvcException as exc:
|
||||
assert isinstance(exc, CheckoutErrorSuggestGit)
|
||||
assert isinstance(exc.__cause__, NoOutputOrStageError)
|
||||
assert exc.__cause__.__cause__ is None
|
||||
|
||||
try:
|
||||
dvc.checkout(targets="looks-like-dvcfile.dvc")
|
||||
except DvcException as exc:
|
||||
assert isinstance(exc, CheckoutErrorSuggestGit)
|
||||
assert isinstance(exc.__cause__, StageFileDoesNotExistError)
|
||||
assert exc.__cause__.__cause__ is None
|
||||
|
||||
self.assertTrue(os.path.exists(self.GIT_IGNORE))
|
||||
|
||||
ignored = self.read_ignored()
|
||||
def test_checkout_target_recursive_should_not_remove_other_used_files(
|
||||
tmp_dir, dvc
|
||||
):
|
||||
tmp_dir.dvc_gen({"foo": "foo", "bar": "bar", "data": {"file": "file"}})
|
||||
assert main(["checkout", "-R", "data"]) == 0
|
||||
assert (tmp_dir / "foo").exists()
|
||||
assert (tmp_dir / "bar").exists()
|
||||
|
||||
self.assertEqual(len(ignored), 2)
|
||||
|
||||
self.assertIn("/" + fname1, ignored)
|
||||
self.assertIn("/" + fname2, ignored)
|
||||
def test_checkout_recursive_not_directory(tmp_dir, dvc):
|
||||
tmp_dir.gen("foo", "foo")
|
||||
ret = main(["add", "foo"])
|
||||
assert ret == 0
|
||||
|
||||
stats = dvc.checkout(targets=["foo" + ".dvc"], recursive=True)
|
||||
assert stats == {"added": [], "modified": [], "deleted": []}
|
||||
|
||||
class TestGitIgnoreWhenCheckout(CheckoutBase):
|
||||
def test(self):
|
||||
fname_master = "file_in_a_master"
|
||||
branch_master = "master"
|
||||
fname_branch = "file_in_a_branch"
|
||||
branch_1 = "b1"
|
||||
|
||||
self.dvc.scm.add(self.dvc.scm.untracked_files())
|
||||
self.dvc.scm.commit("add all files")
|
||||
self.commit_data_file(fname_master)
|
||||
def test_checkout_moved_cache_dir_with_symlinks(tmp_dir, dvc):
|
||||
tmp_dir.gen({"foo": "foo", "data": {"file": "file"}})
|
||||
ret = main(["config", "cache.type", "symlink"])
|
||||
assert ret == 0
|
||||
|
||||
self.dvc.scm.checkout(branch_1, True)
|
||||
ret = main(["checkout", "--force"])
|
||||
self.assertEqual(ret, 0)
|
||||
self.commit_data_file(fname_branch)
|
||||
ret = main(["add", "foo"])
|
||||
assert ret == 0
|
||||
|
||||
self.dvc.scm.checkout(branch_master)
|
||||
ret = main(["checkout", "--force"])
|
||||
self.assertEqual(ret, 0)
|
||||
ret = main(["add", "data"])
|
||||
assert ret == 0
|
||||
|
||||
ignored = self.read_ignored()
|
||||
assert system.is_symlink("foo")
|
||||
old_foo_link = os.path.realpath("foo")
|
||||
|
||||
self.assertEqual(len(ignored), 1)
|
||||
self.assertIn("/" + fname_master, ignored)
|
||||
assert system.is_symlink(os.path.join("data", "file"))
|
||||
old_data_link = os.path.realpath(os.path.join("data", "file"))
|
||||
|
||||
self.dvc.scm.checkout(branch_1)
|
||||
ret = main(["checkout", "--force"])
|
||||
self.assertEqual(ret, 0)
|
||||
ignored = self.read_ignored()
|
||||
self.assertIn("/" + fname_branch, ignored)
|
||||
old_cache_dir = dvc.odb.local.path
|
||||
new_cache_dir = old_cache_dir + "_new"
|
||||
os.rename(old_cache_dir, new_cache_dir)
|
||||
|
||||
ret = main(["cache", "dir", new_cache_dir])
|
||||
assert ret == 0
|
||||
|
||||
class TestCheckoutMissingMd5InStageFile(TestRepro):
|
||||
def test(self):
|
||||
d = load_yaml(self.file1_stage)
|
||||
del d[Stage.PARAM_OUTS][0][LocalFileSystem.PARAM_CHECKSUM]
|
||||
del d[Stage.PARAM_DEPS][0][LocalFileSystem.PARAM_CHECKSUM]
|
||||
dump_yaml(self.file1_stage, d)
|
||||
ret = main(["checkout", "-f"])
|
||||
assert ret == 0
|
||||
|
||||
with pytest.raises(CheckoutError):
|
||||
self.dvc.checkout(force=True)
|
||||
assert system.is_symlink("foo")
|
||||
new_foo_link = os.path.realpath("foo")
|
||||
|
||||
assert system.is_symlink(os.path.join("data", "file"))
|
||||
new_data_link = os.path.realpath(os.path.join("data", "file"))
|
||||
|
||||
class TestCheckoutEmptyDir(TestDvc):
|
||||
def test(self):
|
||||
dname = "empty_dir"
|
||||
os.mkdir(dname)
|
||||
assert relpath(old_foo_link, old_cache_dir) == relpath(
|
||||
new_foo_link, new_cache_dir
|
||||
)
|
||||
|
||||
stages = self.dvc.add(dname)
|
||||
self.assertEqual(len(stages), 1)
|
||||
stage = stages[0]
|
||||
self.assertTrue(stage is not None)
|
||||
self.assertEqual(len(stage.outs), 1)
|
||||
|
||||
stage.outs[0].remove()
|
||||
self.assertFalse(os.path.exists(dname))
|
||||
|
||||
stats = self.dvc.checkout(force=True)
|
||||
assert stats["added"] == [dname + os.sep]
|
||||
|
||||
self.assertTrue(os.path.isdir(dname))
|
||||
self.assertEqual(len(os.listdir(dname)), 0)
|
||||
|
||||
|
||||
class TestCheckoutNotCachedFile(TestDvc):
|
||||
def test(self):
|
||||
cmd = "python {} {} {}".format(self.CODE, self.FOO, "out")
|
||||
|
||||
self.dvc.add(self.FOO)
|
||||
stage = self.dvc.run(
|
||||
cmd=cmd,
|
||||
deps=[self.FOO, self.CODE],
|
||||
outs_no_cache=["out"],
|
||||
single_stage=True,
|
||||
)
|
||||
self.assertTrue(stage is not None)
|
||||
|
||||
stats = self.dvc.checkout(force=True)
|
||||
assert not any(stats.values())
|
||||
|
||||
|
||||
class TestCheckoutWithDeps(TestRepro):
|
||||
def test(self):
|
||||
os.unlink(self.FOO)
|
||||
os.unlink(self.file1)
|
||||
|
||||
self.assertFalse(os.path.exists(self.FOO))
|
||||
self.assertFalse(os.path.exists(self.file1))
|
||||
|
||||
ret = main(["checkout", "--force", self.file1_stage, "--with-deps"])
|
||||
self.assertEqual(ret, 0)
|
||||
|
||||
self.assertTrue(os.path.exists(self.FOO))
|
||||
self.assertTrue(os.path.exists(self.file1))
|
||||
|
||||
|
||||
class TestCheckoutDirectory(TestRepro):
|
||||
def test(self):
|
||||
stage = self.dvc.add(self.DATA_DIR)[0]
|
||||
|
||||
shutil.rmtree(self.DATA_DIR)
|
||||
self.assertFalse(os.path.exists(self.DATA_DIR))
|
||||
|
||||
ret = main(["checkout", stage.path])
|
||||
self.assertEqual(ret, 0)
|
||||
|
||||
self.assertTrue(os.path.exists(self.DATA_DIR))
|
||||
|
||||
|
||||
class TestCheckoutHook(TestDvc):
|
||||
@patch("sys.stdout.isatty", return_value=True)
|
||||
@patch("dvc.prompt.input", side_effect=EOFError)
|
||||
def test(self, _mock_input, _mock_isatty):
|
||||
"""Test that dvc checkout handles EOFError gracefully, which is what
|
||||
it will experience when running in a git hook.
|
||||
"""
|
||||
stages = self.dvc.add(self.DATA_DIR)
|
||||
self.assertEqual(len(stages), 1)
|
||||
stage = stages[0]
|
||||
self.assertNotEqual(stage, None)
|
||||
|
||||
self.create(os.path.join(self.DATA_DIR, "test"), "test")
|
||||
|
||||
with self.assertRaises(ConfirmRemoveError):
|
||||
self.dvc.checkout()
|
||||
|
||||
|
||||
class TestCheckoutSuggestGit(TestRepro):
|
||||
def test(self):
|
||||
# pylint: disable=no-member
|
||||
|
||||
try:
|
||||
self.dvc.checkout(targets="gitbranch")
|
||||
except DvcException as exc:
|
||||
self.assertIsInstance(exc, CheckoutErrorSuggestGit)
|
||||
self.assertIsInstance(exc.__cause__, NoOutputOrStageError)
|
||||
self.assertIsInstance(
|
||||
exc.__cause__.__cause__, StageFileDoesNotExistError
|
||||
)
|
||||
|
||||
try:
|
||||
self.dvc.checkout(targets=self.FOO)
|
||||
except DvcException as exc:
|
||||
self.assertIsInstance(exc, CheckoutErrorSuggestGit)
|
||||
self.assertIsInstance(exc.__cause__, NoOutputOrStageError)
|
||||
self.assertIsNone(exc.__cause__.__cause__)
|
||||
|
||||
try:
|
||||
self.dvc.checkout(targets="looks-like-dvcfile.dvc")
|
||||
except DvcException as exc:
|
||||
self.assertIsInstance(exc, CheckoutErrorSuggestGit)
|
||||
self.assertIsInstance(exc.__cause__, StageFileDoesNotExistError)
|
||||
self.assertIsNone(exc.__cause__.__cause__)
|
||||
|
||||
|
||||
class TestCheckoutTargetRecursiveShouldNotRemoveOtherUsedFiles(TestDvc):
|
||||
def test(self):
|
||||
ret = main(["add", self.DATA_DIR, self.FOO, self.BAR])
|
||||
self.assertEqual(0, ret)
|
||||
|
||||
ret = main(["checkout", "-R", self.DATA_DIR])
|
||||
self.assertEqual(0, ret)
|
||||
|
||||
self.assertTrue(os.path.exists(self.FOO))
|
||||
self.assertTrue(os.path.exists(self.BAR))
|
||||
|
||||
|
||||
class TestCheckoutRecursiveNotDirectory(TestDvc):
|
||||
def test(self):
|
||||
ret = main(["add", self.FOO])
|
||||
self.assertEqual(0, ret)
|
||||
|
||||
stats = self.dvc.checkout(targets=[self.FOO + ".dvc"], recursive=True)
|
||||
assert stats == {"added": [], "modified": [], "deleted": []}
|
||||
|
||||
|
||||
class TestCheckoutMovedCacheDirWithSymlinks(TestDvc):
|
||||
def test(self):
|
||||
ret = main(["config", "cache.type", "symlink"])
|
||||
self.assertEqual(ret, 0)
|
||||
|
||||
ret = main(["add", self.FOO])
|
||||
self.assertEqual(ret, 0)
|
||||
|
||||
ret = main(["add", self.DATA_DIR])
|
||||
self.assertEqual(ret, 0)
|
||||
|
||||
self.assertTrue(system.is_symlink(self.FOO))
|
||||
old_foo_link = os.path.realpath(self.FOO)
|
||||
|
||||
self.assertTrue(system.is_symlink(self.DATA))
|
||||
old_data_link = os.path.realpath(self.DATA)
|
||||
|
||||
old_cache_dir = self.dvc.odb.local.path
|
||||
new_cache_dir = old_cache_dir + "_new"
|
||||
os.rename(old_cache_dir, new_cache_dir)
|
||||
|
||||
ret = main(["cache", "dir", new_cache_dir])
|
||||
self.assertEqual(ret, 0)
|
||||
|
||||
ret = main(["checkout", "-f"])
|
||||
self.assertEqual(ret, 0)
|
||||
|
||||
self.assertTrue(system.is_symlink(self.FOO))
|
||||
new_foo_link = os.path.realpath(self.FOO)
|
||||
|
||||
self.assertTrue(system.is_symlink(self.DATA))
|
||||
new_data_link = os.path.realpath(self.DATA)
|
||||
|
||||
self.assertEqual(
|
||||
relpath(old_foo_link, old_cache_dir),
|
||||
relpath(new_foo_link, new_cache_dir),
|
||||
)
|
||||
|
||||
self.assertEqual(
|
||||
relpath(old_data_link, old_cache_dir),
|
||||
relpath(new_data_link, new_cache_dir),
|
||||
)
|
||||
assert relpath(old_data_link, old_cache_dir) == relpath(
|
||||
new_data_link, new_cache_dir
|
||||
)
|
||||
|
||||
|
||||
def test_checkout_no_checksum(tmp_dir, dvc):
|
||||
|
||||
Reference in New Issue
Block a user