Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test(self):
ret = main(["config", "cache.type", "symlink"])
self.assertEqual(ret, 0)
ret = main(["add", self.FOO])
self.assertEqual(ret, 0)
shutil.rmtree(self.dvc.cache.local.cache_dir)
self.assertTrue(System.is_symlink(self.FOO))
ret = main(["remove", self.FOO + ".dvc"])
self.assertEqual(ret, 0)
self.assertFalse(os.path.lexists(self.FOO))
def test_readding_dir_should_not_unprotect_all(tmp_dir, dvc, mocker):
tmp_dir.gen("dir/data", "data")
dvc.cache.local.cache_types = ["symlink"]
dvc.cache.local.protected = True
dvc.add("dir")
tmp_dir.gen("dir/new_file", "new_file_content")
unprotect_spy = spy(RemoteLOCAL.unprotect)
mocker.patch.object(RemoteLOCAL, "unprotect", unprotect_spy)
dvc.add("dir")
assert not unprotect_spy.mock.called
assert System.is_symlink(os.path.join("dir", "new_file"))
foo_stage, = dvc_repo.add(repo_dir.FOO)
data_dir_stage, = dvc_repo.add(repo_dir.DATA_DIR)
dvc_repo.destroy()
assert not os.path.exists(dvc_repo.dvc_dir)
assert not os.path.exists(foo_stage.path)
assert not os.path.exists(data_dir_stage.path)
assert os.path.isfile(repo_dir.FOO)
assert os.path.isdir(repo_dir.DATA_DIR)
assert os.path.isfile(repo_dir.DATA)
assert os.path.isdir(repo_dir.DATA_SUB_DIR)
assert os.path.isfile(repo_dir.DATA_SUB)
assert not System.is_symlink(repo_dir.FOO)
assert not System.is_symlink(repo_dir.DATA_DIR)
assert not System.is_symlink(repo_dir.DATA)
assert not System.is_symlink(repo_dir.DATA_SUB)
dvc_repo.destroy()
assert not os.path.exists(dvc_repo.dvc_dir)
assert not os.path.exists(foo_stage.path)
assert not os.path.exists(data_dir_stage.path)
assert os.path.isfile(repo_dir.FOO)
assert os.path.isdir(repo_dir.DATA_DIR)
assert os.path.isfile(repo_dir.DATA)
assert os.path.isdir(repo_dir.DATA_SUB_DIR)
assert os.path.isfile(repo_dir.DATA_SUB)
assert not System.is_symlink(repo_dir.FOO)
assert not System.is_symlink(repo_dir.DATA_DIR)
assert not System.is_symlink(repo_dir.DATA)
assert not System.is_symlink(repo_dir.DATA_SUB)
("symlink", "copy", lambda path: not System.is_symlink(path)),
("copy", "hardlink", System.is_hardlink),
def _unprotect_file(self, path):
import stat
import uuid
from dvc.system import System
from dvc.utils import copyfile, move, remove
if System.is_symlink(path) or System.is_hardlink(path):
logger.debug("Unprotecting '{}'".format(path))
tmp = os.path.join(os.path.dirname(path), "." + str(uuid.uuid4()))
move(path, tmp)
copyfile(tmp, path)
remove(tmp)
else:
logger.debug(
"Skipping copying for '{}', since it is not "
"a symlink or a hardlink.".format(path)
)
os.chmod(path, os.stat(path).st_mode | stat.S_IWRITE)
def contains_symlink_up_to(path, base_path):
base_path = fspath(base_path)
path = fspath(path)
if base_path not in path:
raise BasePathNotInCheckedPathException(path, base_path)
if path == base_path:
return False
if System.is_symlink(path):
return True
if os.path.dirname(path) == path:
return False
return contains_symlink_up_to(os.path.dirname(path), base_path)
def _unprotect_file(path):
if System.is_symlink(path) or System.is_hardlink(path):
logger.debug("Unprotecting '{}'".format(path))
tmp = os.path.join(os.path.dirname(path), "." + str(uuid.uuid4()))
move(path, tmp)
copyfile(tmp, path)
remove(tmp)
else:
logger.debug(
"Skipping copying for '{}', since it is not "
"a symlink or a hardlink.".format(path)
)
os.chmod(path, os.stat(path).st_mode | stat.S_IWRITE)