mirror of
https://github.com/Stone-Red-Code/dvc.git
synced 2026-11-08 23:09:48 +01:00
Ruslan Kuprieiev
d5dd0fd804
Since we now use file size to help detect file changes on filesystems with bad mtime resolution, we no longer need these explicit sleep() calls in our tests for mac os. Signed-off-by: Ruslan Kuprieiev <[email protected]>
352 lines
10 KiB
Python
352 lines
10 KiB
Python
import os
|
|
import yaml
|
|
import shutil
|
|
import filecmp
|
|
import collections
|
|
|
|
from dvc.main import main
|
|
from dvc.project import Project
|
|
from dvc.system import System
|
|
from tests.basic_env import TestDvc
|
|
from tests.test_repro import TestRepro
|
|
from dvc.stage import Stage
|
|
from dvc.remote.local import RemoteLOCAL
|
|
from dvc.exceptions import DvcException
|
|
|
|
from mock import patch
|
|
|
|
|
|
class TestCheckout(TestRepro):
|
|
def setUp(self):
|
|
super(TestCheckout, self).setUp()
|
|
|
|
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))
|
|
|
|
|
|
class TestCheckoutSingleStage(TestCheckout):
|
|
def test(self):
|
|
ret = main(['checkout', '--force', self.foo_stage.path])
|
|
self.assertEqual(ret, 0)
|
|
|
|
ret = main(['checkout', '--force', self.data_dir_stage.path])
|
|
self.assertEqual(ret, 0)
|
|
|
|
self._test_checkout()
|
|
|
|
|
|
class TestCheckoutCorruptedCacheFile(TestRepro):
|
|
def test(self):
|
|
cache = self.foo_stage.outs[0].cache
|
|
|
|
with open(cache, 'a') as fd:
|
|
fd.write('1')
|
|
|
|
self.dvc.checkout(force=True)
|
|
|
|
self.assertFalse(os.path.isfile(self.FOO))
|
|
self.assertFalse(os.path.isfile(cache))
|
|
|
|
|
|
class TestCheckoutCorruptedCacheDir(TestDvc):
|
|
def test(self):
|
|
# NOTE: using 'copy' so that cache and link don't have same inode
|
|
ret = main(['config', 'cache.type', 'copy'])
|
|
self.assertEqual(ret, 0)
|
|
|
|
self.dvc = Project('.')
|
|
stages = self.dvc.add(self.DATA_DIR)
|
|
self.assertEqual(len(stages), 1)
|
|
self.assertEqual(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.
|
|
entry = self.dvc.cache.local.load_dir_cache(out.md5)[0]
|
|
md5 = entry[self.dvc.cache.local.PARAM_MD5]
|
|
cache = self.dvc.cache.local.get(md5)
|
|
|
|
with open(cache, 'w+') as fobj:
|
|
fobj.write('1')
|
|
|
|
self.dvc.checkout(force=True)
|
|
|
|
self.assertFalse(os.path.exists(cache))
|
|
|
|
|
|
class TestCmdCheckout(TestCheckout):
|
|
def test(self):
|
|
ret = main(['checkout', '--force'])
|
|
self.assertEqual(ret, 0)
|
|
self._test_checkout()
|
|
|
|
|
|
class CheckoutBase(TestDvc):
|
|
GIT_IGNORE = '.gitignore'
|
|
|
|
def commit_data_file(self, fname, content='random text'):
|
|
with open(fname, 'w') 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):
|
|
return list(map(lambda s: s.strip('\n'),
|
|
open(self.GIT_IGNORE).readlines()))
|
|
|
|
def outs_info(self, stage):
|
|
FileInfo = collections.namedtuple('FileInfo', 'path inode')
|
|
|
|
paths = [
|
|
os.path.join(root, file)
|
|
for output in stage.outs
|
|
for root, _, files in os.walk(output.path)
|
|
for file in files
|
|
]
|
|
|
|
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') 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') as f:
|
|
f.write('not_cached')
|
|
|
|
ret = main(['checkout', stage.relpath])
|
|
self.assertNotEqual(ret, 0)
|
|
|
|
mock_prompt.assert_called()
|
|
self.assertNotEqual(ret, 0)
|
|
self.assertRaises(DvcException)
|
|
|
|
|
|
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)
|
|
|
|
stages = self.dvc.add(self.DATA_DIR)
|
|
self.assertEqual(len(stages), 1)
|
|
stage = stages[0]
|
|
staged_files = self.outs_info(stage)
|
|
|
|
os.remove(staged_files[0].path)
|
|
|
|
ret = main(['checkout', '--force', stage.relpath])
|
|
self.assertEqual(ret, 0)
|
|
|
|
checkedout_files = self.outs_info(stage)
|
|
|
|
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)
|
|
|
|
|
|
class TestGitIgnoreBasic(CheckoutBase):
|
|
def test(self):
|
|
fname1 = 'file_1'
|
|
fname2 = 'file_2'
|
|
fname3 = 'file_3'
|
|
|
|
self.dvc.scm.add(self.dvc.scm.untracked_files())
|
|
self.dvc.scm.commit('add all files')
|
|
|
|
self.assertFalse(os.path.exists(self.GIT_IGNORE))
|
|
|
|
self.commit_data_file(fname1)
|
|
self.commit_data_file(fname2)
|
|
self.dvc.run(cmd='python {} {} {}'.format(self.CODE, self.FOO, fname3),
|
|
deps=[self.CODE, self.FOO],
|
|
outs_no_cache=[fname3])
|
|
|
|
self.assertTrue(os.path.exists(self.GIT_IGNORE))
|
|
|
|
ignored = self.read_ignored()
|
|
|
|
self.assertEqual(len(ignored), 2)
|
|
|
|
self.assertIn('/' + fname1, ignored)
|
|
self.assertIn('/' + fname2, ignored)
|
|
|
|
|
|
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)
|
|
|
|
self.dvc.scm.checkout(branch_1, True)
|
|
ret = main(['checkout', '--force'])
|
|
self.assertEqual(ret, 0)
|
|
self.commit_data_file(fname_branch)
|
|
|
|
self.dvc.scm.checkout(branch_master)
|
|
ret = main(['checkout', '--force'])
|
|
self.assertEqual(ret, 0)
|
|
|
|
ignored = self.read_ignored()
|
|
|
|
self.assertEqual(len(ignored), 1)
|
|
self.assertIn('/' + fname_master, ignored)
|
|
|
|
self.dvc.scm.checkout(branch_1)
|
|
ret = main(['checkout', '--force'])
|
|
self.assertEqual(ret, 0)
|
|
ignored = self.read_ignored()
|
|
self.assertIn('/' + fname_branch, ignored)
|
|
|
|
|
|
class TestCheckoutMissingMd5InStageFile(TestRepro):
|
|
def test(self):
|
|
with open(self.file1_stage, 'r') as fd:
|
|
d = yaml.load(fd)
|
|
|
|
del(d[Stage.PARAM_OUTS][0][RemoteLOCAL.PARAM_MD5])
|
|
del(d[Stage.PARAM_DEPS][0][RemoteLOCAL.PARAM_MD5])
|
|
|
|
with open(self.file1_stage, 'w') as fd:
|
|
yaml.dump(d, fd)
|
|
|
|
self.dvc.checkout(force=True)
|
|
|
|
|
|
class TestCheckoutEmptyDir(TestDvc):
|
|
def test(self):
|
|
dname = 'empty_dir'
|
|
os.mkdir(dname)
|
|
|
|
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))
|
|
|
|
self.dvc.checkout(force=True)
|
|
|
|
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'])
|
|
self.assertTrue(stage is not None)
|
|
|
|
self.dvc.checkout(force=True)
|
|
|
|
|
|
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))
|