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_should_post_checkout_hook_checkout(self, repo_dir, git, dvc_repo):
assert main(["install"]) == 0
stage_file = repo_dir.FOO + Stage.STAGE_FILE_SUFFIX
dvc_repo.add(repo_dir.FOO)
dvc_repo.scm.add([".gitignore", stage_file])
dvc_repo.scm.commit("add")
os.unlink(repo_dir.FOO)
dvc_repo.scm.checkout("new_branch", create_new=True)
assert os.path.isfile(repo_dir.FOO)
def test(self):
remove_outs_call_counter = spy(dvc.stage.Stage.remove_outs)
with patch.object(
dvc.stage.Stage, "remove_outs", remove_outs_call_counter
):
ret = main(["import-url", self.external_source])
self.assertEqual(0, ret)
self.assertEqual(1, remove_outs_call_counter.mock.call_count)
def test_download_error_pulling_imported_stage(dvc_repo, erepo):
src = erepo.FOO
dst = erepo.FOO + "_imported"
dvc_repo.imp(erepo.root_dir, src, dst)
dst_stage = Stage.load(dvc_repo, "foo_imported.dvc")
dst_cache = dst_stage.outs[0].cache_path
os.remove(dst)
os.remove(dst_cache)
with patch(
"dvc.remote.RemoteLOCAL._download", side_effect=Exception
), pytest.raises(DownloadError):
dvc_repo.pull(["foo_imported.dvc"])
def test_ignored_in_checksum(self):
stage = self.dvc.run(
cmd="echo test > {}".format(self.FOO),
deps=[self.BAR],
outs=[self.FOO],
)
d = stage.dumpd()
self.assertNotIn(Stage.PARAM_WDIR, d.keys())
d = load_stage_file(stage.relpath)
self.assertNotIn(Stage.PARAM_WDIR, d.keys())
with self.dvc.state:
stage = Stage.load(self.dvc, stage.relpath)
self.assertFalse(stage.changed())
def test_should_relink_only_one_file_in_dir(dvc_repo, repo_dir):
dvc_repo.cache.local.cache_types = ["symlink"]
dvc_repo.add(repo_dir.DATA_DIR)
dvc_repo.unprotect(repo_dir.DATA_SUB)
link_spy = spy(System.symlink)
with patch.object(dvc_repo.cache.local, "symlink", link_spy):
dvc_repo.checkout([repo_dir.DATA_DIR + Stage.STAGE_FILE_SUFFIX])
assert link_spy.mock.call_count == 1
def test_always_changed(dvc_repo):
stage = Stage(dvc_repo, "path", always_changed=True)
stage.save()
assert stage.changed()
assert stage.status()["path"] == ["always changed"]
def test_pull_imported_stage(dvc_repo, erepo):
src = erepo.FOO
dst = erepo.FOO + "_imported"
dvc_repo.imp(erepo.root_dir, src, dst)
dst_stage = Stage.load(dvc_repo, "foo_imported.dvc")
dst_cache = dst_stage.outs[0].cache_path
os.remove(dst)
os.remove(dst_cache)
dvc_repo.pull(["foo_imported.dvc"])
assert os.path.isfile(dst)
assert os.path.isfile(dst_cache)
" along with `-f` to specify DVC-file path with working"
" directory.".format(fname=fname)
)
wdir = cwd
elif wdir is None:
wdir = os.curdir
stage = Stage(
repo=repo,
wdir=wdir,
cmd=kwargs.get("cmd", None),
locked=kwargs.get("locked", False),
always_changed=kwargs.get("always_changed", False),
)
Stage._fill_stage_outputs(stage, **kwargs)
stage.deps = dependency.loads_from(
stage, kwargs.get("deps", []), erepo=kwargs.get("erepo", None)
)
stage._check_circular_dependency()
stage._check_duplicated_arguments()
if not fname:
fname = Stage._stage_fname(stage.outs, add)
stage._check_dvc_filename(fname)
# Autodetecting wdir for add, we need to create outs first to do that,
# so we start with wdir = . and remap out paths later.
if add and kwargs.get("wdir") is None and cwd is None:
wdir = os.path.dirname(fname)
def load(repo, fname):
fname, tag = Stage._get_path_tag(fname)
# it raises the proper exceptions by priority:
# 1. when the file doesn't exists
# 2. filename is not a DVC-file
# 3. path doesn't represent a regular file
Stage._check_file_exists(repo, fname)
Stage._check_dvc_filename(fname)
Stage._check_isfile(repo, fname)
with repo.tree.open(fname) as fd:
stage_text = fd.read()
d = parse_stage(stage_text, fname)
Stage.validate(d, fname=relpath(fname))
path = os.path.abspath(fname)
stage = Stage(
repo=repo,
path=path,
wdir=os.path.abspath(
os.path.join(
os.path.dirname(path), d.get(Stage.PARAM_WDIR, ".")
)
),
add = kwargs.get("add", False)
# Backward compatibility for `cwd` option
if wdir is None and cwd is not None:
if fname is not None and os.path.basename(fname) != fname:
raise StageFileBadNameError(
"DVC-file name '{fname}' may not contain subdirectories"
" if `-c|--cwd` (deprecated) is specified. Use `-w|--wdir`"
" along with `-f` to specify DVC-file path with working"
" directory.".format(fname=fname)
)
wdir = cwd
elif wdir is None:
wdir = os.curdir
stage = Stage(
repo=repo,
wdir=wdir,
cmd=kwargs.get("cmd", None),
locked=kwargs.get("locked", False),
always_changed=kwargs.get("always_changed", False),
)
Stage._fill_stage_outputs(stage, **kwargs)
stage.deps = dependency.loads_from(
stage, kwargs.get("deps", []), erepo=kwargs.get("erepo", None)
)
stage._check_circular_dependency()
stage._check_duplicated_arguments()
if not fname: