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_cache_type_is_properly_overridden(git, dvc_repo, erepo):
erepo.dvc.config.set(
Config.SECTION_CACHE, Config.SECTION_CACHE_TYPE, "symlink"
)
erepo.dvc.scm.add([erepo.dvc.config.config_file])
erepo.dvc.scm.commit("set source repo cache type to symlinks")
src = erepo.FOO
dst = erepo.FOO + "_imported"
dvc_repo.imp(erepo.root_dir, src, dst)
assert not System.is_symlink(dst)
assert os.path.exists(dst)
assert os.path.isfile(dst)
assert filecmp.cmp(erepo.FOO, dst, shallow=False)
assert git.git.check_ignore(dst)
os.chdir(self._test_git_dir)
os.mkdir(os.path.join('data', 'dir1'))
os.mkdir(os.path.join('.dvc', 'cache', 'dir1'))
self.file1_cache = os.path.join('.dvc', 'cache', 'dir1', 'file1')
self.file1_data = os.path.join('data', 'dir1', 'file1')
open(self.file1_cache, 'w').write('ddfdff')
System.symlink(self.file1_cache, self.file1_data)
self.file2_cache = os.path.join('.dvc', 'cache', 'file2')
self.file2_data = os.path.join('data', 'file2')
open(self.file2_cache, 'w').write('ddfdff')
System.symlink(self.file2_cache, self.file2_data)
def setUp(self):
BasicEnvironment.init_environment(self, test_dir=os.path.join(os.path.sep, 'tmp', 'ntx_unit_test'))
git = GitWrapperI(git_dir=self._test_git_dir, commit='ad45ba8')
config = ConfigI('data')
self.path_factory = PathFactory(git, config)
self.data_file = os.path.join('data', 'file.txt')
self.cache_file = os.path.join(ConfigI.CONFIG_DIR, ConfigI.CACHE_DIR, 'fsymlinc.txt')
fd = open(self.cache_file, 'w+')
fd.write('some text')
fd.close()
os.chdir('data')
System.symlink(os.path.join('..', self.cache_file), 'file.txt')
os.chdir('..')
pass
old_foo_link = readlink(self.FOO)
self.assertTrue(System.is_symlink(self.DATA))
old_data_link = readlink(self.DATA)
old_cache_dir = self.dvc.cache.local.cache_dir
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 = readlink(self.FOO)
self.assertTrue(System.is_symlink(self.DATA))
new_data_link = readlink(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),
)
def get_dvc_path(self):
pwd = System.get_cwd()
if not pwd.startswith(self.settings.git.git_dir_abs):
raise StateFileError('the file cannot be created outside of a git repository')
return os.path.relpath(pwd, self.settings.git.git_dir_abs)
def reflink(source, link_name):
import platform
from dvc.exceptions import DvcException
source, link_name = fspath(source), fspath(link_name)
system = platform.system()
try:
if system == "Windows":
ret = System._reflink_windows(source, link_name)
elif system == "Darwin":
ret = System._reflink_darwin(source, link_name)
elif system == "Linux":
ret = System._reflink_linux(source, link_name)
else:
ret = -1
except IOError:
ret = -1
if ret != 0:
raise DvcException("reflink is not supported")
else:
line.append(" ")
assert len(line) == (smaxcol + 1)
screen.print_at("".join(line), 0, y)
screen.refresh()
# NOTE: get_event() doesn't block by itself,
# so we have to do the blocking ourselves.
#
# NOTE: using this workaround while waiting for PR [1]
# to get merged and released. After that need to adjust
# asciimatics version requirements.
#
# [1] https://github.com/peterbrittain/asciimatics/pull/188
System.wait_for_input(self.TIMEOUT)
event = screen.get_event()
if not isinstance(event, KeyboardEvent):
continue
k = event.key_code
if k == screen.KEY_DOWN or k == ord("s"):
offset_y += 1
elif k == screen.KEY_PAGE_DOWN or k == ord("S"):
offset_y += smaxrow
elif k == screen.KEY_UP or k == ord("w"):
offset_y -= 1
elif k == screen.KEY_PAGE_UP or k == ord("W"):
offset_y -= smaxrow
elif k == screen.KEY_RIGHT or k == ord("d"):
offset_x += 1
def inode(path):
path = fspath(path)
if System.is_unix():
import ctypes
inode = os.lstat(path).st_ino
# NOTE: See https://bugs.python.org/issue29619 and
# https://stackoverflow.com/questions/34643289/
# pythons-os-stat-is-returning-wrong-inode-value
inode = ctypes.c_ulong(inode).value
else:
# getdirinfo from ntfsutils works on both files and dirs
info = System._getdirinfo(path)
inode = abs(
hash(
(
info.dwVolumeSerialNumber,
info.nFileIndexHigh,
info.nFileIndexLow,
)
)
)
assert inode >= 0
assert inode < 2 ** 64
return inode
def copy(self, from_info, to_info):
tmp_info = to_info.parent / tmp_fname(to_info.name)
try:
System.copy(from_info, tmp_info)
os.rename(fspath_py35(tmp_info), fspath_py35(to_info))
except Exception:
self.remove(tmp_info)
raise