Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def scm(tmp_dir, request):
# Use dvc.scm if available
if "dvc" in request.fixturenames:
dvc = request.getfixturevalue("dvc")
tmp_dir.scm = dvc.scm
yield dvc.scm
else:
from dvc.scm.git import Git
_git_init()
try:
scm = tmp_dir.scm = Git(fspath(tmp_dir))
yield scm
finally:
scm.close()
drives = [
s[0].upper()
for s in win32api.GetLogicalDriveStrings().split("\000")
if len(s) > 0
]
new_drive_name = [
letter for letter in string.ascii_uppercase if letter not in drives
][0]
new_drive = "{}:".format(new_drive_name)
target_path = tmp_path_factory.mktemp("tmp_windows_drive")
set_up_result = windll.kernel32.DefineDosDeviceW(
0, new_drive, fspath(target_path)
)
if set_up_result == 0:
raise RuntimeError("Failed to mount windows drive!")
# NOTE: new_drive has form of `A:` and joining it with some relative
# path might result in non-existing path (A:path\\to)
yield os.path.join(new_drive, os.sep)
tear_down_result = windll.kernel32.DefineDosDeviceW(
DDD_REMOVE_DEFINITION, new_drive, fspath(target_path)
)
if tear_down_result == 0:
raise RuntimeError("Could not unmount windows drive!")
tmp_dir.dvc_gen("file", "text")
tmp_dir.dvc_gen({"dir": {"file": "lorem", "subdir/file": "ipsum"}})
dvc.destroy()
# Remove all the files related to DVC
for path in [".dvc", "file.dvc", "dir.dvc"]:
assert not (tmp_dir / path).exists()
# Leave the rest of the files
for path in ["file", "dir/file", "dir/subdir/file"]:
assert (tmp_dir / path).is_file()
# Make sure that data was unprotected after `destroy`
for path in ["file", "dir", "dir/file", "dir/subdir", "dir/subdir/file"]:
assert not System.is_symlink(fspath(tmp_dir / path))
def stages():
return set(stage.relpath for stage in Repo(fspath(tmp_dir)).stages)
def _coerce_filenames(filenames):
if isinstance(filenames, (basestring, pathlib.PurePath)):
filenames = [filenames]
return lmap(fspath, filenames)
def test_daemon_analytics(mock_send, tmp_path):
report = fspath(tmp_path)
assert 0 == main(["daemon", "analytics", report])
mock_send.assert_called_with(report)
def relpath(path, start=os.curdir):
path = fspath(path)
start = os.path.abspath(fspath(start))
# Windows path on different drive than curdir doesn't have relpath
if os.name == "nt" and not os.path.commonprefix(
[start, os.path.abspath(path)]
):
return path
return os.path.relpath(path, start)
def symlink(source, link_name):
import ctypes
from dvc.exceptions import DvcException
source, link_name = fspath(source), fspath(link_name)
if System.is_unix():
try:
os.symlink(source, link_name)
return
except Exception as exc:
msg = "failed to symlink '{}' -> '{}': {}"
raise DvcException(msg.format(source, link_name, str(exc)))
flags = 0
if source is not None and os.path.isdir(source):
flags = 1
func = ctypes.windll.kernel32.CreateSymbolicLinkW
func.argtypes = (ctypes.c_wchar_p, ctypes.c_wchar_p, ctypes.c_uint32)
func.restype = ctypes.c_ubyte
def relpath(path, start=os.curdir):
path = fspath(path)
start = os.path.abspath(fspath(start))
# Windows path on different drive than curdir doesn't have relpath
if os.name == "nt" and not os.path.commonprefix(
[start, os.path.abspath(path)]
):
return path
return os.path.relpath(path, start)