Stone_Red/dvc
1
0
Fork 0
mirror of https://github.com/Stone-Red-Code/dvc.git synced 2026-11-02 07:18:25 +01:00
Code Issues Packages Projects Releases Wiki Activity

tests: remove single_stage tests from repro/run tests (#9553)

Browse Source
This commit is contained in:
skshetry
2023-06-07 09:25:08 +05:45
committed by GitHub
parent 0c709b6ec1
commit 6c58d438b3
8 changed files with 1371 additions and 2171 deletions
Download Patch File Download Diff File Expand all files Collapse all files
dvc/repo/run.py
+2
View File
@@ -21,6 +21,8 @@ def run(
force: bool = True,
**kwargs,
) -> Union["Stage", "PipelineStage"]:
assert not kwargs.get("single_stage")
assert not kwargs.get("fname")
kwargs.update({"force": force, "params": parse_params(kwargs.get("params", []))})
stage = self.stage.create(**kwargs)
tests/func/test_lockfile.py
+1 -1
View File
@@ -5,7 +5,7 @@ from dvc.dvcfile import LOCK_FILE
from dvc.stage.utils import split_params_deps
from dvc.utils.fs import remove
from dvc.utils.serialize import dumps_yaml, parse_yaml_for_update
from tests.func.test_run_multistage import supported_params
from tests.func.test_run import supported_params
FS_STRUCTURE = {
"foo": "bar\nfoobar",
tests/func/test_repro.py
+650 -671
View File
File diff suppressed because it is too large Load Diff
tests/func/test_repro_multistage.py
-511
View File
@@ -1,511 +0,0 @@
import os
from copy import deepcopy
from textwrap import dedent
import pytest
from funcy import lsplit
from dvc.cli import main
from dvc.dvcfile import LOCK_FILE, PROJECT_FILE
from dvc.exceptions import CyclicGraphError, ReproductionError
from dvc.stage import PipelineStage
from dvc.stage.cache import RunCacheNotSupported
from dvc.stage.exceptions import StageNotFound
from dvc.utils.fs import remove
def test_non_existing_stage_name(tmp_dir, dvc, run_copy):
tmp_dir.gen("file1", "file1")
run_copy("file1", "file2", name="copy-file1-file2")
with pytest.raises(StageNotFound):
dvc.freeze(":copy-file1-file3")
assert main(["freeze", ":copy-file1-file3"]) != 0
def test_repro_frozen(tmp_dir, dvc, run_copy):
(data_stage,) = tmp_dir.dvc_gen("data", "foo")
stage0 = run_copy("data", "stage0", name="copy-data-stage0")
run_copy("stage0", "stage1", name="copy-data-stage1")
run_copy("stage1", "stage2", name="copy-data-stage2")
dvc.freeze("copy-data-stage1")
tmp_dir.gen("data", "bar")
stages = dvc.reproduce()
assert stages == [data_stage, stage0]
def test_downstream(M, tmp_dir, dvc):
# The dependency graph should look like this:
#
# E
# / \
# D F
# / \ \
# B C G
# \ /
# A
#
assert main(["stage", "add", "--run", "-n", "A-gen", "-o", "A", "echo A>A"]) == 0
assert (
main(["stage", "add", "--run", "-n", "B-gen", "-d", "A", "-o", "B", "echo B>B"])
== 0
)
assert (
main(
[
"stage",
"add",
"--run",
"-n",
"C-gen",
"-d",
"A",
"-o",
"C",
"echo C>C",
]
)
== 0
)
assert (
main(
[
"stage",
"add",
"--run",
"-n",
"D-gen",
"-d",
"B",
"-d",
"C",
"-o",
"D",
"echo D>D",
]
)
== 0
)
assert main(["stage", "add", "--run", "-n", "G-gen", "-o", "G", "echo G>G"]) == 0
assert (
main(["stage", "add", "--run", "-n", "F-gen", "-d", "G", "-o", "F", "echo F>F"])
== 0
)
assert (
main(
[
"stage",
"add",
"--run",
"-n",
"E-gen",
"-d",
"D",
"-d",
"F",
"-o",
"E",
"echo E>E",
]
)
== 0
)
# We want the evaluation to move from B to E
#
# E
# /
# D
# /
# B
#
evaluation = dvc.reproduce(PROJECT_FILE + ":B-gen", downstream=True, force=True)
assert len(evaluation) == 3
assert all(isinstance(stage, PipelineStage) for stage in evaluation)
assert all(stage.relpath == PROJECT_FILE for stage in evaluation)
assert [stage.name for stage in evaluation] == ["B-gen", "D-gen", "E-gen"]
# B, C should be run (in any order) before D
# See https://github.com/iterative/dvc/issues/3602
evaluation = dvc.reproduce(PROJECT_FILE + ":A-gen", downstream=True, force=True)
assert len(evaluation) == 5
assert all(isinstance(stage, PipelineStage) for stage in evaluation)
assert all(stage.relpath == PROJECT_FILE for stage in evaluation)
assert [stage.name for stage in evaluation] == [
"A-gen",
M.any_of("B-gen", "C-gen"),
M.any_of("B-gen", "C-gen"),
"D-gen",
"E-gen",
]
def test_repro_when_cmd_changes(tmp_dir, dvc, run_copy, mocker):
from dvc.dvcfile import ProjectFile
tmp_dir.gen("foo", "foo")
stage = run_copy("foo", "bar", name="copy-file")
target = "copy-file"
assert not dvc.reproduce(target)
from dvc.stage.run import cmd_run
m = mocker.patch("dvc.stage.run.cmd_run", wraps=cmd_run)
stage.cmd = " ".join(stage.cmd.split()) # change cmd spacing by two
ProjectFile(dvc, PROJECT_FILE)._dump_pipeline_file(stage)
assert dvc.status([target]) == {target: ["changed command"]}
assert dvc.reproduce(target)[0] == stage
m.assert_called_once_with(stage, dry=False, run_env=None)
def test_repro_when_new_deps_is_added_in_dvcfile(tmp_dir, dvc, run_copy, copy_script):
from dvc.dvcfile import load_file
tmp_dir.gen({"foo": "foo", "bar": "bar"})
stage = dvc.run(
cmd="python copy.py {} {}".format("foo", "foobar"),
outs=["foobar"],
deps=["foo"],
name="copy-file",
)
target = PROJECT_FILE + ":copy-file"
assert not dvc.reproduce(target)
dvcfile = load_file(dvc, stage.path)
data, _ = dvcfile._load()
data["stages"]["copy-file"]["deps"] += ["copy.py"]
(tmp_dir / stage.path).dump(data)
assert dvc.reproduce(target)[0] == stage
def test_repro_when_new_outs_is_added_in_dvcfile(tmp_dir, dvc, copy_script):
from dvc.dvcfile import load_file
tmp_dir.gen({"foo": "foo", "bar": "bar"})
stage = dvc.run(
cmd="python copy.py {} {}".format("foo", "foobar"),
outs=[], # scenario where user forgot to add
deps=["foo"],
name="copy-file",
)
target = ":copy-file"
assert not dvc.reproduce(target)
dvcfile = load_file(dvc, stage.path)
data, _ = dvcfile._load()
data["stages"]["copy-file"]["outs"] = ["foobar"]
(tmp_dir / stage.path).dump(data)
assert dvc.reproduce(target)[0] == stage
def test_repro_when_new_deps_is_moved(tmp_dir, dvc, copy_script):
from dvc.dvcfile import load_file
tmp_dir.gen({"foo": "foo", "bar": "foo"})
stage = dvc.run(
cmd="python copy.py {} {}".format("foo", "foobar"),
outs=["foobar"],
deps=["foo"],
name="copy-file",
)
target = ":copy-file"
assert not dvc.reproduce(target)
# hardcode values in source code, ignore sys.argv
tmp_dir.gen(
"copy.py",
"""
import shutil
shutil.copyfile('bar', 'foobar')
""",
)
from shutil import move
move("foo", "bar")
dvcfile = load_file(dvc, stage.path)
data, _ = dvcfile._load()
data["stages"]["copy-file"]["deps"] = ["bar"]
(tmp_dir / stage.path).dump(data)
assert dvc.reproduce(target)[0] == stage
def test_repro_when_new_out_overlaps_others_stage_outs(tmp_dir, dvc):
from dvc.exceptions import OverlappingOutputPathsError
tmp_dir.gen({"dir": {"file1": "file1"}, "foo": "foo"})
dvc.add("dir")
(tmp_dir / PROJECT_FILE).dump(
{
"stages": {
"run-copy": {
"cmd": "python copy {} {}".format("foo", "dir/foo"),
"deps": ["foo"],
"outs": ["dir/foo"],
}
}
},
)
with pytest.raises(OverlappingOutputPathsError):
dvc.reproduce(":run-copy")
def test_repro_when_new_deps_added_does_not_exist(tmp_dir, dvc, copy_script):
tmp_dir.gen("foo", "foo")
(tmp_dir / PROJECT_FILE).dump(
{
"stages": {
"run-copy": {
"cmd": "python copy.py {} {}".format("foo", "foobar"),
"deps": ["foo", "bar"],
"outs": ["foobar"],
}
}
},
)
with pytest.raises(ReproductionError):
dvc.reproduce(":run-copy")
def test_repro_when_new_outs_added_does_not_exist(tmp_dir, dvc, copy_script):
tmp_dir.gen("foo", "foo")
(tmp_dir / PROJECT_FILE).dump(
{
"stages": {
"run-copy": {
"cmd": "python copy.py {} {}".format("foo", "foobar"),
"deps": ["foo"],
"outs": ["foobar", "bar"],
}
}
},
)
with pytest.raises(ReproductionError):
dvc.reproduce(":run-copy")
def test_repro_when_lockfile_gets_deleted(tmp_dir, dvc, copy_script):
tmp_dir.gen("foo", "foo")
(tmp_dir / PROJECT_FILE).dump(
{
"stages": {
"run-copy": {
"cmd": "python copy.py {} {}".format("foo", "foobar"),
"deps": ["foo"],
"outs": ["foobar"],
}
}
},
)
assert dvc.reproduce(":run-copy")
assert os.path.exists(LOCK_FILE)
assert not dvc.reproduce(":run-copy")
os.unlink(LOCK_FILE)
stages = dvc.reproduce(":run-copy")
assert stages
assert stages[0].relpath == PROJECT_FILE
assert stages[0].name == "run-copy"
def test_cyclic_graph_error(tmp_dir, dvc, run_copy):
tmp_dir.gen("foo", "foo")
run_copy("foo", "bar", name="copy-foo-bar")
run_copy("bar", "baz", name="copy-bar-baz")
run_copy("baz", "foobar", name="copy-baz-foobar")
data = (tmp_dir / PROJECT_FILE).parse()
data["stages"]["copy-baz-foo"] = {
"cmd": "echo baz > foo",
"deps": ["baz"],
"outs": ["foo"],
}
(tmp_dir / PROJECT_FILE).dump(data)
with pytest.raises(CyclicGraphError):
dvc.reproduce(":copy-baz-foo")
def test_repro_multiple_params(tmp_dir, dvc):
from dvc.stage.utils import split_params_deps
from tests.func.test_run_multistage import supported_params
(tmp_dir / "params2.yaml").dump(supported_params)
(tmp_dir / "params.yaml").dump(supported_params)
(tmp_dir / "foo").write_text("foo")
stage = dvc.run(
name="read_params",
deps=["foo"],
outs=["bar"],
params=[
"params2.yaml:lists,floats,name",
"answer,floats,nested.nested1",
],
cmd="cat params2.yaml params.yaml > bar",
)
params, deps = split_params_deps(stage)
assert len(params) == 2
assert len(deps) == 1
assert len(stage.outs) == 1
lockfile = stage.dvcfile._lockfile
assert lockfile.load()["stages"]["read_params"]["params"] == {
"params2.yaml": {
"lists": [42, 42.0, "42"],
"floats": 42.0,
"name": "Answer",
},
"params.yaml": {
"answer": 42,
"floats": 42.0,
"nested.nested1": {"nested2": "42", "nested2-2": 41.99999},
},
}
data, _ = stage.dvcfile._load()
params = data["stages"]["read_params"]["params"]
custom, defaults = lsplit(lambda v: isinstance(v, dict), params)
assert set(custom[0]["params2.yaml"]) == {"name", "lists", "floats"}
assert set(defaults) == {"answer", "floats", "nested.nested1"}
assert not dvc.reproduce(stage.addressing)
params = deepcopy(supported_params)
params["answer"] = 43
(tmp_dir / "params.yaml").dump(params)
assert dvc.reproduce(stage.addressing) == [stage]
@pytest.mark.parametrize("multiline", [True, False])
def test_repro_list_of_commands_in_order(tmp_dir, dvc, multiline):
cmd = ["echo foo>foo", "echo bar>bar"]
if multiline:
cmd = "\n".join(cmd)
(tmp_dir / "dvc.yaml").dump({"stages": {"multi": {"cmd": cmd}}})
(tmp_dir / "dvc.yaml").write_text(
dedent(
"""\
stages:
multi:
cmd:
- echo foo>foo
- echo bar>bar
"""
)
)
dvc.reproduce(targets=["multi"])
assert (tmp_dir / "foo").read_text() == "foo\n"
assert (tmp_dir / "bar").read_text() == "bar\n"
@pytest.mark.parametrize("multiline", [True, False])
def test_repro_list_of_commands_raise_and_stops_after_failure(tmp_dir, dvc, multiline):
cmd = ["echo foo>foo", "failed_command", "echo baz>bar"]
if multiline:
cmd = "\n".join(cmd)
(tmp_dir / "dvc.yaml").dump({"stages": {"multi": {"cmd": cmd}}})
with pytest.raises(ReproductionError):
dvc.reproduce(targets=["multi"])
assert (tmp_dir / "foo").read_text() == "foo\n"
assert not (tmp_dir / "bar").exists()
def test_repro_pulls_mising_data_source(tmp_dir, dvc, mocker, local_remote):
(foo,) = tmp_dir.dvc_gen("foo", "foo")
dvc.push()
dvc.stage.add(name="copy-foo", cmd="cp foo bar", deps=["foo"], outs=["bar"])
remove("foo")
remove(foo.outs[0].cache_path)
assert dvc.reproduce(pull=True)
def test_repro_pulls_mising_import(tmp_dir, dvc, mocker, erepo_dir, local_remote):
with erepo_dir.chdir():
erepo_dir.dvc_gen("foo", "foo", commit="first")
foo_import = dvc.imp(os.fspath(erepo_dir), "foo")
dvc.push()
dvc.stage.add(name="copy-foo", cmd="cp foo bar", deps=["foo"], outs=["bar"])
remove("foo")
remove(foo_import.outs[0].cache_path)
assert dvc.reproduce(pull=True)
def test_repro_allow_missing(tmp_dir, dvc):
tmp_dir.gen("fixed", "fixed")
dvc.stage.add(name="create-foo", cmd="echo foo > foo", deps=["fixed"], outs=["foo"])
dvc.stage.add(name="copy-foo", cmd="cp foo bar", deps=["foo"], outs=["bar"])
(create_foo, copy_foo) = dvc.reproduce()
remove("foo")
remove(create_foo.outs[0].cache_path)
remove(dvc.stage_cache.cache_dir)
ret = dvc.reproduce(allow_missing=True)
# both stages are skipped
assert not ret
def test_repro_allow_missing_and_pull(tmp_dir, dvc, mocker, local_remote):
tmp_dir.gen("fixed", "fixed")
dvc.stage.add(name="create-foo", cmd="echo foo > foo", deps=["fixed"], outs=["foo"])
dvc.stage.add(name="copy-foo", cmd="cp foo bar", deps=["foo"], outs=["bar"])
(create_foo,) = dvc.reproduce("create-foo")
dvc.push()
remove("foo")
remove(create_foo.outs[0].cache_path)
remove(dvc.stage_cache.cache_dir)
ret = dvc.reproduce(pull=True, allow_missing=True)
# create-foo is skipped ; copy-foo pulls missing dep
assert len(ret) == 1
def test_repro_pulls_continue_without_run_cache(tmp_dir, dvc, mocker, local_remote):
(foo,) = tmp_dir.dvc_gen("foo", "foo")
dvc.push()
mocker.patch.object(
dvc.stage_cache, "pull", side_effect=RunCacheNotSupported("foo")
)
dvc.stage.add(name="copy-foo", cmd="cp foo bar", deps=["foo"], outs=["bar"])
remove("foo")
remove(foo.outs[0].cache_path)
assert dvc.reproduce(pull=True)
def test_repro_skip_pull_if_no_run_cache_is_passed(tmp_dir, dvc, mocker, local_remote):
(foo,) = tmp_dir.dvc_gen("foo", "foo")
dvc.push()
spy_pull = mocker.spy(dvc.stage_cache, "pull")
dvc.stage.add(name="copy-foo", cmd="cp foo bar", deps=["foo"], outs=["bar"])
remove("foo")
remove(foo.outs[0].cache_path)
assert dvc.reproduce(pull=True, run_cache=False)
assert not spy_pull.called
tests/func/test_run.py
+718
View File
@@ -0,0 +1,718 @@
import logging
import os
import textwrap
import uuid
import pytest
from funcy import get_in
from dvc.cli import main
from dvc.dependency import ParamsDependency
from dvc.dependency.base import DependencyDoesNotExistError
from dvc.dvcfile import LOCK_FILE, PROJECT_FILE, load_file
from dvc.exceptions import (
ArgumentDuplicationError,
CircularDependencyError,
CyclicGraphError,
InvalidArgumentError,
OutputDuplicationError,
OverlappingOutputPathsError,
)
from dvc.stage import PipelineStage
from dvc.stage.exceptions import (
DuplicateStageName,
InvalidStageName,
StagePathNotDirectoryError,
StagePathNotFoundError,
StagePathOutsideError,
)
from dvc.utils.serialize import load_yaml
def test_run(tmp_dir, dvc, copy_script):
tmp_dir.dvc_gen("foo", "foo")
assert not os.path.exists(PROJECT_FILE)
stage = dvc.run(
cmd="python copy.py foo bar",
deps=["foo", "copy.py"],
outs=["bar"],
name="copy-foo-to-bar",
)
assert isinstance(stage, PipelineStage)
assert stage.name == "copy-foo-to-bar"
assert os.path.exists(PROJECT_FILE)
assert os.path.exists(LOCK_FILE)
assert stage.cmd == "python copy.py foo bar"
assert len(stage.deps) == 2
assert len(stage.outs) == 1
with pytest.raises(OutputDuplicationError):
dvc.run(
cmd="python copy.py foo bar",
deps=["foo", "copy.py"],
outs=["bar"],
name="duplicate",
)
def test_run_empty(dvc):
dvc.run(
cmd="echo hello world",
deps=[],
outs=[],
outs_no_cache=[],
name="empty",
)
def test_run_missing_dep(dvc):
with pytest.raises(DependencyDoesNotExistError):
dvc.run(
cmd="command",
deps=["non-existing-dep"],
outs=[],
outs_no_cache=[],
name="missing-dep",
)
def test_run_no_exec(tmp_dir, dvc, scm, run_copy):
tmp_dir.dvc_gen("foo", "foo")
assert not os.path.exists(PROJECT_FILE)
stage = run_copy("foo", "bar", name="copy-foo-to-bar", no_exec=True)
assert isinstance(stage, PipelineStage)
assert stage.name == "copy-foo-to-bar"
assert os.path.exists(PROJECT_FILE)
assert not os.path.exists(LOCK_FILE)
assert not os.path.exists("bar")
data, _ = stage.dvcfile._load()
assert data["stages"]["copy-foo-to-bar"] == {
"cmd": "python copy.py foo bar",
"deps": ["copy.py", "foo"],
"outs": ["bar"],
}
with open(".gitignore", encoding="utf-8") as fobj:
assert fobj.read() == "/foo\n/bar\n"
def test_run_repeat(tmp_dir, dvc, run_copy):
from dvc.dvcfile import PROJECT_FILE, load_file
from dvc.stage import PipelineStage
tmp_dir.dvc_gen("foo", "foo")
run_copy("foo", "foo1", name="copy-foo-foo1")
run_copy("foo1", "foo2", name="copy-foo1-foo2")
run_copy("foo2", "foo3", name="copy-foo2-foo3")
stages = list(load_file(dvc, PROJECT_FILE).stages.values())
assert len(stages) == 3
assert all(isinstance(stage, PipelineStage) for stage in stages)
assert {stage.name for stage in stages} == {
"copy-foo-foo1",
"copy-foo1-foo2",
"copy-foo2-foo3",
}
def test_run_cached(tmp_dir, dvc, run_copy, mocker):
from dvc.stage.run import subprocess
tmp_dir.dvc_gen("foo", "foo")
run_copy("foo", "foo2", name="copy-foo1-foo2")
spy = mocker.spy(subprocess, "Popen")
run_copy("foo", "foo2", name="copy-foo1-foo2")
assert not spy.called
def test_dump_on_non_cached_outputs(tmp_dir, dvc):
tmp_dir.dvc_gen("foo")
dvc.run(
cmd="cp foo foo1",
deps=["foo"],
name="copy-foo1-foo2",
outs_no_cache=["foo1"],
)
def test_with_wdir(tmp_dir, dvc):
tmp_dir.gen({"dir": {"foo": "foo", "bar": "bar"}})
stage = dvc.run(
cmd="cp foo foo1",
deps=["foo"],
name="copy-foo1-foo2",
outs=["foo1"],
wdir="dir",
)
data, _ = load_file(dvc, stage.path)._load()
assert data["stages"]["copy-foo1-foo2"]["wdir"] == "dir"
def test_always_changed(tmp_dir, dvc):
from dvc.dvcfile import load_file
tmp_dir.gen({"foo": "foo", "bar": "bar"})
stage = dvc.run(
cmd="cp foo foo1",
deps=["foo"],
name="copy-foo1-foo2",
outs=["foo1"],
always_changed=True,
)
data, _ = load_file(dvc, stage.path)._load()
assert data["stages"]["copy-foo1-foo2"]["always_changed"]
def test_graph(tmp_dir, dvc):
from dvc.exceptions import CyclicGraphError
tmp_dir.gen({"foo": "foo", "bar": "bar"})
dvc.run(deps=["foo"], outs=["bar"], cmd="echo foo > bar", name="1")
dvc.run(deps=["bar"], outs=["baz"], cmd="echo bar > baz", name="2")
with pytest.raises(CyclicGraphError):
dvc.run(deps=["baz"], outs=["foo"], cmd="echo baz > foo", name="3")
class TestRunCircularDependency:
def test(self, dvc):
with pytest.raises(CircularDependencyError):
dvc.run(
cmd="command",
deps=["foo"],
outs=["foo"],
name="circular-dependency",
)
def test_outs_no_cache(self, dvc):
with pytest.raises(CircularDependencyError):
dvc.run(
cmd="command",
deps=["foo"],
outs_no_cache=["foo"],
name="circular-dependency",
)
def test_non_normalized_paths(self, dvc):
with pytest.raises(CircularDependencyError):
dvc.run(
cmd="command",
deps=["./foo"],
outs=["foo"],
name="circular-dependency",
)
def test_graph(self, tmp_dir, dvc):
tmp_dir.gen("foo", "foo")
dvc.run(
deps=["foo"],
outs=["bar.txt"],
cmd="echo bar > bar.txt",
name="gen-bar-txt",
)
dvc.run(
deps=["bar.txt"],
outs=["baz.txt"],
cmd="echo baz > baz.txt",
name="gen-baz-txt",
)
with pytest.raises(CyclicGraphError):
dvc.run(
deps=["baz.txt"],
outs=["foo"],
cmd="echo baz > foo",
name="gen-foo",
)
class TestRunDuplicatedArguments:
def test(self, dvc):
with pytest.raises(ArgumentDuplicationError):
dvc.run(
cmd="command",
deps=[],
outs=["foo", "foo"],
name="circular-dependency",
)
def test_outs_no_cache(self, dvc):
with pytest.raises(ArgumentDuplicationError):
dvc.run(
cmd="command",
outs=["foo"],
outs_no_cache=["foo"],
name="circular-dependency",
)
def test_non_normalized_paths(self, dvc):
with pytest.raises(ArgumentDuplicationError):
dvc.run(
cmd="command",
deps=[],
outs=["foo", "./foo"],
name="circular-dependency",
)
class TestRunBadWdir:
def test(self, make_tmp_dir, dvc):
with pytest.raises(StagePathOutsideError):
dvc.run(
cmd="command",
wdir=make_tmp_dir("tmp"),
name="bad-wdir",
)
def test_same_prefix(self, tmp_dir, dvc):
path = f"{tmp_dir}-{uuid.uuid4()}"
os.mkdir(path)
with pytest.raises(StagePathOutsideError):
dvc.run(cmd="command", wdir=path, name="bad-wdir")
def test_not_found(self, tmp_dir, dvc):
path = os.path.join(tmp_dir, str(uuid.uuid4()))
with pytest.raises(StagePathNotFoundError):
dvc.run(cmd="command", wdir=path, name="bad-wdir")
def test_not_dir(self, tmp_dir, dvc):
path = tmp_dir / str(uuid.uuid4())
path.mkdir()
path = path / str(uuid.uuid4())
path.touch()
with pytest.raises(StagePathNotDirectoryError):
dvc.run(
cmd="command",
wdir=os.fspath(path),
name="bad-wdir",
)
class TestCmdRunWorkingDirectory:
def test_default_wdir_is_not_written(self, tmp_dir, dvc):
dvc.run(
cmd="echo test > foo",
outs=["foo"],
wdir=".",
name="echo-foo",
)
d = load_yaml("dvc.yaml")
assert "wdir" not in get_in(d, ["stages", "echo-foo"])
dvc.run(cmd="echo test > bar", outs=["bar"], name="echo-bar")
d = load_yaml("dvc.yaml")
assert "wdir" not in get_in(d, ["stages", "echo-bar"])
def test_fname_changes_path_and_wdir(self, tmp_dir, dvc):
dirpath = tmp_dir / "dir"
dirpath.mkdir()
with dirpath.chdir():
stage = dvc.run(
cmd="echo test > foo",
outs=["foo"],
wdir=os.fspath(tmp_dir),
name="echo",
)
assert stage.wdir == os.path.realpath(tmp_dir)
# Check that it is dumped properly
d = load_yaml("dir/dvc.yaml")
assert get_in(d, ["stages", "echo", "wdir"]) == ".."
def test_run_dump(tmp_dir, dvc, run_head):
from dvc.dvcfile import load_file
tmp_dir.gen(
{
"dir": {
"foo": "foo\nfoo",
"bar": "bar\nbar",
"foobar": "foobar\foobar",
}
}
)
dvc.run(
cmd="cp foo foo2",
deps=["foo"],
name="copy-foo-foo2",
wdir="dir",
outs_persist=["foo2"],
always_changed=True,
)
data = load_file(dvc, PROJECT_FILE)._load()[0]
assert data == {
"stages": {
"copy-foo-foo2": {
"cmd": "cp foo foo2",
"deps": ["foo"],
"outs": [{"foo2": {"persist": True}}],
"always_changed": True,
"wdir": "dir",
}
}
}
run_head(
"foo",
"bar",
"foobar",
name="head-files",
outs=["bar-1"],
outs_persist=["foo-1"],
metrics_no_cache=["foobar-1"],
wdir="dir",
)
assert load_file(dvc, PROJECT_FILE)._load()[0] == {
"stages": {
"head-files": {
"cmd": "python {} foo bar foobar".format(
(tmp_dir / "head.py").resolve()
),
"wdir": "dir",
"deps": ["bar", "foo", "foobar"],
"outs": ["bar-1", {"foo-1": {"persist": True}}],
"metrics": [{"foobar-1": {"cache": False}}],
},
**data["stages"],
}
}
@pytest.mark.parametrize("char", ["@:", "#", "$", ":", "/", "\\", ".", ";", ","])
def test_run_with_invalid_stage_name(run_copy, char):
with pytest.raises(InvalidStageName):
run_copy("foo", "bar", name=f"copy_name-{char}")
def test_run_with_name_having_hyphen_underscore(tmp_dir, dvc, run_copy):
tmp_dir.dvc_gen("foo", "foo")
run_copy("foo", "bar", name="copy-foo_bar")
def test_run_already_exists(tmp_dir, dvc, run_copy):
tmp_dir.dvc_gen("foo", "foo")
run_copy("foo", "bar", name="copy")
with pytest.raises(DuplicateStageName):
run_copy("bar", "foobar", name="copy", force=False)
run_copy("bar", "foobar", name="copy", force=True)
supported_params = {
"name": "Answer",
"answer": 42,
"floats": 42.0,
"lists": [42, 42.0, "42"],
"nested": {"nested1": {"nested2": "42", "nested2-2": 41.99999}},
}
def test_run_params_default(tmp_dir, dvc):
(tmp_dir / "params.yaml").dump(supported_params)
stage = dvc.run(
name="read_params",
params=["nested.nested1.nested2"],
cmd="cat params.yaml",
)
assert isinstance(stage.deps[0], ParamsDependency)
assert stage.deps[0].params == ["nested.nested1.nested2"]
lockfile = stage.dvcfile._lockfile
assert lockfile.load()["stages"]["read_params"]["params"] == {
"params.yaml": {"nested.nested1.nested2": "42"}
}
data, _ = stage.dvcfile._load()
assert data["stages"]["read_params"]["params"] == ["nested.nested1.nested2"]
def test_run_params_custom_file(tmp_dir, dvc):
(tmp_dir / "params2.yaml").dump(supported_params)
stage = dvc.run(
name="read_params",
params=["params2.yaml:lists"],
cmd="cat params2.yaml",
)
isinstance(stage.deps[0], ParamsDependency)
assert stage.deps[0].params == ["lists"]
lockfile = stage.dvcfile._lockfile
assert lockfile.load()["stages"]["read_params"]["params"] == {
"params2.yaml": {"lists": [42, 42.0, "42"]}
}
data, _ = stage.dvcfile._load()
assert data["stages"]["read_params"]["params"] == [{"params2.yaml": ["lists"]}]
def test_run_params_no_exec(tmp_dir, dvc):
(tmp_dir / "params2.yaml").dump(supported_params)
stage = dvc.run(
name="read_params",
params=["params2.yaml:lists"],
cmd="cat params2.yaml",
no_exec=True,
)
isinstance(stage.deps[0], ParamsDependency)
assert stage.deps[0].params == ["lists"]
assert not stage.dvcfile._lockfile.exists()
data, _ = stage.dvcfile._load()
assert data["stages"]["read_params"]["params"] == [{"params2.yaml": ["lists"]}]
@pytest.mark.parametrize(
"kwargs",
[
{"outs": ["foo"], "deps": ["bar"]},
{"outs": ["foo"], "deps": ["bar"], "name": "copy-foo-bar"},
],
)
def test_run_without_cmd(tmp_dir, dvc, kwargs):
with pytest.raises(InvalidArgumentError) as exc:
dvc.run(**kwargs)
assert str(exc.value) == "command is not specified"
def test_run_overwrite_order(tmp_dir, dvc, run_copy):
tmp_dir.gen({"foo": "foo", "foo1": "foo1"})
run_copy("foo", "bar", name="copy-foo-bar")
run_copy("bar", "foobar", name="copy-bar-foobar")
run_copy("foo1", "bar1", name="copy-foo-bar", force=True)
data = (tmp_dir / PROJECT_FILE).parse()
assert list(data["stages"].keys()) == ["copy-foo-bar", "copy-bar-foobar"]
def test_run_overwrite_preserves_meta_and_comment(tmp_dir, dvc, run_copy):
tmp_dir.gen({"foo": "foo", "foo1": "foo1"})
text = textwrap.dedent(
"""\
stages:
copy-foo-bar:
cmd: python copy.py {src} {dest}
deps:
- copy.py
- {src}
outs:
# comments are preserved
- {dest}
meta:
name: meta is preserved too
"""
)
(tmp_dir / PROJECT_FILE).write_text(text.format(src="foo", dest="bar"))
assert dvc.reproduce(PROJECT_FILE)
assert run_copy("foo1", "bar1", name="copy-foo-bar", force=True)
assert (tmp_dir / PROJECT_FILE).read_text() == text.format(src="foo1", dest="bar1")
def test_run_external_outputs(
tmp_dir,
dvc,
local_workspace,
):
hash_name = "md5"
foo_hash = "acbd18db4cc2f85cedef654fccc4a4d8"
bar_hash = "37b51d194a7513e45b56f6524f2d51f2"
local_workspace.gen("foo", "foo")
dvc.run(
name="mystage",
cmd="mycmd",
deps=["remote://workspace/foo"],
outs=["remote://workspace/bar"],
no_exec=True,
)
dvc_yaml = (
"stages:\n"
" mystage:\n"
" cmd: mycmd\n"
" deps:\n"
" - remote://workspace/foo\n"
" outs:\n"
" - remote://workspace/bar\n"
)
assert (tmp_dir / "dvc.yaml").read_text() == dvc_yaml
assert not (tmp_dir / "dvc.lock").exists()
local_workspace.gen("bar", "bar")
dvc.commit("dvc.yaml", force=True)
assert (tmp_dir / "dvc.yaml").read_text() == dvc_yaml
assert (tmp_dir / "dvc.lock").read_text() == (
"schema: '2.0'\n"
"stages:\n"
" mystage:\n"
" cmd: mycmd\n"
" deps:\n"
" - path: remote://workspace/foo\n"
f" {hash_name}: {foo_hash}\n"
" size: 3\n"
" outs:\n"
" - path: remote://workspace/bar\n"
f" {hash_name}: {bar_hash}\n"
" size: 3\n"
)
assert (local_workspace / "foo").read_text() == "foo"
assert (local_workspace / "bar").read_text() == "bar"
assert (
local_workspace / "cache" / bar_hash[:2] / bar_hash[2:]
).read_text() == "bar"
def test_rerun_callback(dvc):
def run_callback(force=False):
return dvc.run(cmd="echo content > out", force=force, name="echo")
assert run_callback() is not None
with pytest.raises(DuplicateStageName):
assert run_callback() is not None
assert run_callback(force=True) is not None
def test_rerun_changed_dep(tmp_dir, run_copy):
tmp_dir.gen("foo", "foo content")
assert run_copy("foo", "out", name="copy") is not None
tmp_dir.gen("foo", "changed content")
with pytest.raises(DuplicateStageName):
run_copy("foo", "out", force=False, name="copy")
assert run_copy("foo", "out", force=True, name="copy")
def test_run_remove_outs(tmp_dir, dvc, append_foo_script):
tmp_dir.gen("foo", "foo")
dvc.run(
deps=["append_foo.py"],
outs=["foo"],
cmd="python append_foo.py foo",
name="append-foo",
)
@pytest.mark.parametrize("metrics_type", ["metrics", "metrics_no_cache"])
def test_metrics_dir(tmp_dir, dvc, caplog, run_copy_metrics, metrics_type):
copyargs = {metrics_type: ["dir_metric"]}
tmp_dir.gen({"dir": {"file": "content"}})
with caplog.at_level(logging.DEBUG, "dvc"):
run_copy_metrics("dir", "dir_metric", name="copy-metrics", **copyargs)
assert "directory 'dir_metric' cannot be used as metrics." in caplog.messages
def test_rerun_deterministic(tmp_dir, run_copy, mocker):
from dvc.stage.run import subprocess
tmp_dir.gen("foo", "foo content")
spy = mocker.spy(subprocess, "Popen")
run_copy("foo", "out", name="copy")
assert spy.called
spy.reset_mock()
run_copy("foo", "out", name="copy")
assert not spy.called
def test_rerun_deterministic_ignore_cache(tmp_dir, run_copy, mocker):
from dvc.stage.run import subprocess
tmp_dir.gen("foo", "foo content")
spy = mocker.spy(subprocess, "Popen")
run_copy("foo", "out", name="copy")
assert spy.called
spy.reset_mock()
run_copy("foo", "out", run_cache=False, name="copy")
assert spy.called
def test_rerun_changed_stage(tmp_dir, run_copy):
tmp_dir.gen("foo", "foo content")
assert run_copy("foo", "out", name="copy") is not None
tmp_dir.gen("bar", "bar content")
with pytest.raises(DuplicateStageName):
run_copy("bar", "out", force=False, name="copy")
def test_rerun_changed_out(tmp_dir, run_copy):
tmp_dir.gen("foo", "foo content")
assert run_copy("foo", "out", name="copy") is not None
tmp_dir.gen("out", "modification")
with pytest.raises(DuplicateStageName):
run_copy("foo", "out", force=False, name="copy")
def test_should_raise_on_overlapping_output_paths(tmp_dir, dvc, append_foo_script):
tmp_dir.gen("data", {"foo": "foo", "bar": "bar"})
ret = main(["add", "data"])
assert ret == 0
foo_file = os.path.join("data", "foo")
with pytest.raises(OverlappingOutputPathsError) as err:
dvc.run(
outs=["data/foo"],
cmd=f"python append_foo.py {foo_file}",
name="append-foo",
)
error_output = str(err.value)
assert "The output paths:\n" in error_output
assert "\n'data'('data.dvc')\n" in error_output
assert f"\n'{foo_file}'('append-foo')\n" in error_output
assert (
"overlap and are thus in the same tracked directory.\n"
"To keep reproducibility, outputs should be in separate "
"tracked directories or tracked individually." in error_output
)
def test_should_not_checkout_upon_corrupted_local_hardlink_cache(
mocker, tmp_dir, dvc, copy_script
):
tmp_dir.gen("foo", "foo")
dvc.cache.local.cache_types = ["hardlink"]
stage = dvc.run(
deps=["foo"],
outs=["bar"],
cmd="python copy.py foo bar",
name="copy",
)
os.chmod("bar", 0o644)
with open("bar", "w", encoding="utf-8") as fd:
fd.write("corrupting the output cache")
spy_checkout = mocker.spy(stage.outs[0], "checkout")
from dvc.stage import run as stage_run
spy_run = mocker.spy(stage_run, "cmd_run")
with dvc.lock:
stage.run()
spy_run.assert_called_once()
spy_checkout.assert_not_called()
tests/func/test_run_multistage.py
-384
View File
@@ -1,384 +0,0 @@
import os
import textwrap
import pytest
from dvc.exceptions import InvalidArgumentError
from dvc.stage.exceptions import DuplicateStageName, InvalidStageName
def test_run_with_name(tmp_dir, dvc, run_copy):
from dvc.dvcfile import LOCK_FILE, PROJECT_FILE
from dvc.stage import PipelineStage
tmp_dir.dvc_gen("foo", "foo")
assert not os.path.exists(PROJECT_FILE)
stage = run_copy("foo", "bar", name="copy-foo-to-bar")
assert isinstance(stage, PipelineStage)
assert stage.name == "copy-foo-to-bar"
assert os.path.exists(PROJECT_FILE)
assert os.path.exists(LOCK_FILE)
def test_run_no_exec(tmp_dir, dvc, run_copy):
from dvc.dvcfile import LOCK_FILE, PROJECT_FILE
from dvc.stage import PipelineStage
tmp_dir.dvc_gen("foo", "foo")
assert not os.path.exists(PROJECT_FILE)
stage = run_copy("foo", "bar", name="copy-foo-to-bar", no_exec=True)
assert isinstance(stage, PipelineStage)
assert stage.name == "copy-foo-to-bar"
assert os.path.exists(PROJECT_FILE)
assert not os.path.exists(LOCK_FILE)
data, _ = stage.dvcfile._load()
assert data["stages"]["copy-foo-to-bar"] == {
"cmd": "python copy.py foo bar",
"deps": ["copy.py", "foo"],
"outs": ["bar"],
}
def test_run_multi_stage_repeat(tmp_dir, dvc, run_copy):
from dvc.dvcfile import PROJECT_FILE, load_file
from dvc.stage import PipelineStage
tmp_dir.dvc_gen("foo", "foo")
run_copy("foo", "foo1", name="copy-foo-foo1")
run_copy("foo1", "foo2", name="copy-foo1-foo2")
run_copy("foo2", "foo3", name="copy-foo2-foo3")
stages = list(load_file(dvc, PROJECT_FILE).stages.values())
assert len(stages) == 3
assert all(isinstance(stage, PipelineStage) for stage in stages)
assert {stage.name for stage in stages} == {
"copy-foo-foo1",
"copy-foo1-foo2",
"copy-foo2-foo3",
}
def test_multi_stage_run_cached(tmp_dir, dvc, run_copy, mocker):
from dvc.stage.run import subprocess
tmp_dir.dvc_gen("foo", "foo")
run_copy("foo", "foo2", name="copy-foo1-foo2")
spy = mocker.spy(subprocess, "Popen")
run_copy("foo", "foo2", name="copy-foo1-foo2")
assert not spy.called
def test_multistage_dump_on_non_cached_outputs(tmp_dir, dvc):
tmp_dir.dvc_gen("foo")
dvc.run(
cmd="cp foo foo1",
deps=["foo"],
name="copy-foo1-foo2",
outs_no_cache=["foo1"],
)
def test_multistage_with_wdir(tmp_dir, dvc):
from dvc.dvcfile import load_file
tmp_dir.gen({"dir": {"foo": "foo", "bar": "bar"}})
stage = dvc.run(
cmd="cp foo foo1",
deps=["foo"],
name="copy-foo1-foo2",
outs=["foo1"],
wdir="dir",
)
data, _ = load_file(dvc, stage.path)._load()
assert data["stages"]["copy-foo1-foo2"]["wdir"] == "dir"
def test_multistage_always_changed(tmp_dir, dvc):
from dvc.dvcfile import load_file
tmp_dir.gen({"foo": "foo", "bar": "bar"})
stage = dvc.run(
cmd="cp foo foo1",
deps=["foo"],
name="copy-foo1-foo2",
outs=["foo1"],
always_changed=True,
)
data, _ = load_file(dvc, stage.path)._load()
assert data["stages"]["copy-foo1-foo2"]["always_changed"]
def test_graph(tmp_dir, dvc):
from dvc.exceptions import CyclicGraphError
tmp_dir.gen({"foo": "foo", "bar": "bar"})
dvc.run(deps=["foo"], outs=["bar"], cmd="echo foo > bar", name="1")
dvc.run(deps=["bar"], outs=["baz"], cmd="echo bar > baz", name="2")
with pytest.raises(CyclicGraphError):
dvc.run(deps=["baz"], outs=["foo"], cmd="echo baz > foo", name="3")
def test_run_dump_on_multistage(tmp_dir, dvc, run_head):
from dvc.dvcfile import PROJECT_FILE, load_file
tmp_dir.gen(
{
"dir": {
"foo": "foo\nfoo",
"bar": "bar\nbar",
"foobar": "foobar\foobar",
}
}
)
dvc.run(
cmd="cp foo foo2",
deps=["foo"],
name="copy-foo-foo2",
wdir="dir",
outs_persist=["foo2"],
always_changed=True,
)
data = load_file(dvc, PROJECT_FILE)._load()[0]
assert data == {
"stages": {
"copy-foo-foo2": {
"cmd": "cp foo foo2",
"deps": ["foo"],
"outs": [{"foo2": {"persist": True}}],
"always_changed": True,
"wdir": "dir",
}
}
}
run_head(
"foo",
"bar",
"foobar",
name="head-files",
outs=["bar-1"],
outs_persist=["foo-1"],
metrics_no_cache=["foobar-1"],
wdir="dir",
)
assert load_file(dvc, PROJECT_FILE)._load()[0] == {
"stages": {
"head-files": {
"cmd": "python {} foo bar foobar".format(
(tmp_dir / "head.py").resolve()
),
"wdir": "dir",
"deps": ["bar", "foo", "foobar"],
"outs": ["bar-1", {"foo-1": {"persist": True}}],
"metrics": [{"foobar-1": {"cache": False}}],
},
**data["stages"],
}
}
@pytest.mark.parametrize("char", ["@:", "#", "$", ":", "/", "\\", ".", ";", ","])
def test_run_with_invalid_stage_name(run_copy, char):
with pytest.raises(InvalidStageName):
run_copy("foo", "bar", name=f"copy_name-{char}")
def test_run_with_name_having_hyphen_underscore(tmp_dir, dvc, run_copy):
tmp_dir.dvc_gen("foo", "foo")
run_copy("foo", "bar", name="copy-foo_bar")
def test_run_already_exists(tmp_dir, dvc, run_copy):
tmp_dir.dvc_gen("foo", "foo")
run_copy("foo", "bar", name="copy")
with pytest.raises(DuplicateStageName):
run_copy("bar", "foobar", name="copy", force=False)
run_copy("bar", "foobar", name="copy", force=True)
supported_params = {
"name": "Answer",
"answer": 42,
"floats": 42.0,
"lists": [42, 42.0, "42"],
"nested": {"nested1": {"nested2": "42", "nested2-2": 41.99999}},
}
def test_run_params_default(tmp_dir, dvc):
from dvc.dependency import ParamsDependency
(tmp_dir / "params.yaml").dump(supported_params)
stage = dvc.run(
name="read_params",
params=["nested.nested1.nested2"],
cmd="cat params.yaml",
)
assert isinstance(stage.deps[0], ParamsDependency)
assert stage.deps[0].params == ["nested.nested1.nested2"]
lockfile = stage.dvcfile._lockfile
assert lockfile.load()["stages"]["read_params"]["params"] == {
"params.yaml": {"nested.nested1.nested2": "42"}
}
data, _ = stage.dvcfile._load()
assert data["stages"]["read_params"]["params"] == ["nested.nested1.nested2"]
def test_run_params_custom_file(tmp_dir, dvc):
from dvc.dependency import ParamsDependency
(tmp_dir / "params2.yaml").dump(supported_params)
stage = dvc.run(
name="read_params",
params=["params2.yaml:lists"],
cmd="cat params2.yaml",
)
isinstance(stage.deps[0], ParamsDependency)
assert stage.deps[0].params == ["lists"]
lockfile = stage.dvcfile._lockfile
assert lockfile.load()["stages"]["read_params"]["params"] == {
"params2.yaml": {"lists": [42, 42.0, "42"]}
}
data, _ = stage.dvcfile._load()
assert data["stages"]["read_params"]["params"] == [{"params2.yaml": ["lists"]}]
def test_run_params_no_exec(tmp_dir, dvc):
from dvc.dependency import ParamsDependency
(tmp_dir / "params2.yaml").dump(supported_params)
stage = dvc.run(
name="read_params",
params=["params2.yaml:lists"],
cmd="cat params2.yaml",
no_exec=True,
)
isinstance(stage.deps[0], ParamsDependency)
assert stage.deps[0].params == ["lists"]
assert not stage.dvcfile._lockfile.exists()
data, _ = stage.dvcfile._load()
assert data["stages"]["read_params"]["params"] == [{"params2.yaml": ["lists"]}]
@pytest.mark.parametrize(
"kwargs",
[
{"outs": ["foo"], "deps": ["bar"]},
{"outs": ["foo"], "deps": ["bar"], "name": "copy-foo-bar"},
],
)
def test_run_without_cmd(tmp_dir, dvc, kwargs):
with pytest.raises(InvalidArgumentError) as exc:
dvc.run(**kwargs)
assert str(exc.value) == "command is not specified"
def test_run_overwrite_order(tmp_dir, dvc, run_copy):
from dvc.dvcfile import PROJECT_FILE
tmp_dir.gen({"foo": "foo", "foo1": "foo1"})
run_copy("foo", "bar", name="copy-foo-bar")
run_copy("bar", "foobar", name="copy-bar-foobar")
run_copy("foo1", "bar1", name="copy-foo-bar", force=True)
data = (tmp_dir / PROJECT_FILE).parse()
assert list(data["stages"].keys()) == ["copy-foo-bar", "copy-bar-foobar"]
def test_run_overwrite_preserves_meta_and_comment(tmp_dir, dvc, run_copy):
from dvc.dvcfile import PROJECT_FILE
tmp_dir.gen({"foo": "foo", "foo1": "foo1"})
text = textwrap.dedent(
"""\
stages:
copy-foo-bar:
cmd: python copy.py {src} {dest}
deps:
- copy.py
- {src}
outs:
# comments are preserved
- {dest}
meta:
name: meta is preserved too
"""
)
(tmp_dir / PROJECT_FILE).write_text(text.format(src="foo", dest="bar"))
assert dvc.reproduce(PROJECT_FILE)
assert run_copy("foo1", "bar1", name="copy-foo-bar", force=True)
assert (tmp_dir / PROJECT_FILE).read_text() == text.format(src="foo1", dest="bar1")
def test_run_external_outputs(
tmp_dir,
dvc,
local_workspace,
):
hash_name = "md5"
foo_hash = "acbd18db4cc2f85cedef654fccc4a4d8"
bar_hash = "37b51d194a7513e45b56f6524f2d51f2"
local_workspace.gen("foo", "foo")
dvc.run(
name="mystage",
cmd="mycmd",
deps=["remote://workspace/foo"],
outs=["remote://workspace/bar"],
no_exec=True,
)
dvc_yaml = (
"stages:\n"
" mystage:\n"
" cmd: mycmd\n"
" deps:\n"
" - remote://workspace/foo\n"
" outs:\n"
" - remote://workspace/bar\n"
)
assert (tmp_dir / "dvc.yaml").read_text() == dvc_yaml
assert not (tmp_dir / "dvc.lock").exists()
local_workspace.gen("bar", "bar")
dvc.commit("dvc.yaml", force=True)
assert (tmp_dir / "dvc.yaml").read_text() == dvc_yaml
assert (tmp_dir / "dvc.lock").read_text() == (
"schema: '2.0'\n"
"stages:\n"
" mystage:\n"
" cmd: mycmd\n"
" deps:\n"
" - path: remote://workspace/foo\n"
f" {hash_name}: {foo_hash}\n"
" size: 3\n"
" outs:\n"
" - path: remote://workspace/bar\n"
f" {hash_name}: {bar_hash}\n"
" size: 3\n"
)
assert (local_workspace / "foo").read_text() == "foo"
assert (local_workspace / "bar").read_text() == "bar"
assert (
local_workspace / "cache" / bar_hash[:2] / bar_hash[2:]
).read_text() == "bar"
tests/func/test_run_single_stage.py
-592
View File
@@ -1,592 +0,0 @@
import filecmp
import logging
import os
import textwrap
import uuid
from pathlib import Path
import pytest
from dvc.cli import main
from dvc.dependency.base import DependencyIsStageFileError
from dvc.dvcfile import DVC_FILE_SUFFIX
from dvc.exceptions import (
ArgumentDuplicationError,
CircularDependencyError,
CyclicGraphError,
OutputDuplicationError,
OverlappingOutputPathsError,
StagePathAsOutputError,
)
from dvc.output import OutputIsStageFileError
from dvc.stage import Stage
from dvc.stage.exceptions import (
StageFileAlreadyExistsError,
StageFileBadNameError,
StagePathNotDirectoryError,
StagePathNotFoundError,
StagePathOutsideError,
)
from dvc.utils.serialize import load_yaml
from dvc_data.hashfile.hash import file_md5
def test_run(tmp_dir, copy_script, dvc):
tmp_dir.gen("foo", "foo")
cmd = "python copy.py foo out"
deps = ["foo", "copy.py"]
outs = [os.path.join(tmp_dir, "out")]
outs_no_cache = []
fname = "out.dvc"
dvc.add("foo")
stage = dvc.run(
cmd=cmd,
deps=deps,
outs=outs,
outs_no_cache=outs_no_cache,
fname=fname,
single_stage=True,
)
assert filecmp.cmp("foo", "out", shallow=False)
assert os.path.isfile(stage.path)
assert stage.cmd == cmd
assert len(stage.deps) == len(deps)
assert len(stage.outs) == len(outs + outs_no_cache)
assert stage.outs[0].fspath == outs[0]
assert stage.outs[0].hash_info.value == file_md5("foo")
assert stage.path, fname
with pytest.raises(OutputDuplicationError):
dvc.run(
cmd=cmd,
deps=deps,
outs=outs,
outs_no_cache=outs_no_cache,
fname="duplicate" + fname,
single_stage=True,
)
def test_run_empty(dvc):
dvc.run(
cmd="echo hello world",
deps=[],
outs=[],
outs_no_cache=[],
fname="empty.dvc",
single_stage=True,
)
def test_run_missing_dep(dvc):
from dvc.dependency.base import DependencyDoesNotExistError
with pytest.raises(DependencyDoesNotExistError):
dvc.run(
cmd="command",
deps=["non-existing-dep"],
outs=[],
outs_no_cache=[],
fname="empty.dvc",
single_stage=True,
)
def test_run_noexec(tmp_dir, dvc, scm):
tmp_dir.gen("foo", "foo")
dvc.run(
cmd="cp foo bar",
deps=["foo"],
outs=["bar"],
no_exec=True,
single_stage=True,
)
assert not os.path.exists("bar")
with open(".gitignore", encoding="utf-8") as fobj:
assert fobj.read() == "/bar\n"
class TestRunCircularDependency:
def test(self, dvc):
with pytest.raises(CircularDependencyError):
dvc.run(
cmd="command",
deps=["foo"],
outs=["foo"],
fname="circular-dependency.dvc",
single_stage=True,
)
def test_outs_no_cache(self, dvc):
with pytest.raises(CircularDependencyError):
dvc.run(
cmd="command",
deps=["foo"],
outs_no_cache=["foo"],
fname="circular-dependency.dvc",
single_stage=True,
)
def test_non_normalized_paths(self, dvc):
with pytest.raises(CircularDependencyError):
dvc.run(
cmd="command",
deps=["./foo"],
outs=["foo"],
fname="circular-dependency.dvc",
single_stage=True,
)
def test_graph(self, tmp_dir, dvc):
tmp_dir.gen("foo", "foo")
dvc.run(
deps=["foo"],
outs=["bar.txt"],
cmd="echo bar > bar.txt",
single_stage=True,
)
dvc.run(
deps=["bar.txt"],
outs=["baz.txt"],
cmd="echo baz > baz.txt",
single_stage=True,
)
with pytest.raises(CyclicGraphError):
dvc.run(
deps=["baz.txt"],
outs=["foo"],
cmd="echo baz > foo",
single_stage=True,
)
class TestRunDuplicatedArguments:
def test(self, dvc):
with pytest.raises(ArgumentDuplicationError):
dvc.run(
cmd="command",
deps=[],
outs=["foo", "foo"],
fname="circular-dependency.dvc",
single_stage=True,
)
def test_outs_no_cache(self, dvc):
with pytest.raises(ArgumentDuplicationError):
dvc.run(
cmd="command",
outs=["foo"],
outs_no_cache=["foo"],
fname="circular-dependency.dvc",
single_stage=True,
)
def test_non_normalized_paths(self, dvc):
with pytest.raises(ArgumentDuplicationError):
dvc.run(
cmd="command",
deps=[],
outs=["foo", "./foo"],
fname="circular-dependency.dvc",
single_stage=True,
)
class TestRunStageInsideOutput:
def test_cwd(self, tmp_dir, dvc):
tmp_dir.gen("data", {"foo": "foo", "bar": "bar"})
dvc.run(
cmd="mkdir data",
deps=[],
outs=["data"],
single_stage=True,
)
with pytest.raises(StagePathAsOutputError):
dvc.run(
cmd="command",
fname=os.path.join("data", "inside-cwd.dvc"),
single_stage=True,
)
def test_file_name(self, tmp_dir, dvc):
tmp_dir.gen("data", {"foo": "foo", "bar": "bar"})
dvc.run(
cmd="mkdir data",
deps=[],
outs=["data"],
single_stage=True,
)
with pytest.raises(StagePathAsOutputError):
dvc.run(
cmd="command",
outs=["foo"],
fname=os.path.join("data", "inside-cwd.dvc"),
single_stage=True,
)
class TestRunBadCwd:
def test(self, make_tmp_dir, dvc):
with pytest.raises(StagePathOutsideError):
dvc.run(
cmd="command",
wdir=make_tmp_dir("tmp"),
fname="bad-cwd.dvc",
single_stage=True,
)
def test_same_prefix(self, tmp_dir, dvc):
path = f"{tmp_dir}-{uuid.uuid4()}"
os.mkdir(path)
with pytest.raises(StagePathOutsideError):
dvc.run(cmd="command", wdir=path, fname="bad-cwd.dvc", single_stage=True)
class TestRunBadWdir:
def test(self, make_tmp_dir, dvc):
with pytest.raises(StagePathOutsideError):
dvc.run(
cmd="command",
wdir=make_tmp_dir("tmp"),
fname="bad-wdir.dvc",
single_stage=True,
)
def test_same_prefix(self, tmp_dir, dvc):
path = f"{tmp_dir}-{uuid.uuid4()}"
os.mkdir(path)
with pytest.raises(StagePathOutsideError):
dvc.run(cmd="command", wdir=path, fname="bad-wdir.dvc", single_stage=True)
def test_not_found(self, tmp_dir, dvc):
path = os.path.join(tmp_dir, str(uuid.uuid4()))
with pytest.raises(StagePathNotFoundError):
dvc.run(cmd="command", wdir=path, fname="bad-wdir.dvc", single_stage=True)
def test_not_dir(self, tmp_dir, dvc):
path = tmp_dir / str(uuid.uuid4())
path.mkdir()
path = path / str(uuid.uuid4())
path.touch()
with pytest.raises(StagePathNotDirectoryError):
dvc.run(
cmd="command",
wdir=os.fspath(path),
fname="bad-wdir.dvc",
single_stage=True,
)
class TestRunBadName:
def test(self, make_tmp_dir, dvc):
with pytest.raises(StagePathOutsideError):
dvc.run(
cmd="command",
fname=os.path.join(make_tmp_dir("tmp"), "foo.dvc"),
single_stage=True,
)
def test_same_prefix(self, tmp_dir, dvc):
path = f"{tmp_dir}-{uuid.uuid4()}"
os.mkdir(path)
with pytest.raises(StagePathOutsideError):
dvc.run(
cmd="command",
fname=os.path.join(path, "foo.dvc"),
single_stage=True,
)
def test_not_found(self, tmp_dir, dvc):
path = os.path.join(tmp_dir, str(uuid.uuid4()))
with pytest.raises(StagePathNotFoundError):
dvc.run(
cmd="command",
fname=os.path.join(path, "foo.dvc"),
single_stage=True,
)
def test_run_remove_outs(tmp_dir, dvc, append_foo_script):
dvc.run(
deps=["append_foo.py"],
outs=["foo"],
cmd="python append_foo.py foo",
single_stage=True,
)
class TestCmdRunWorkingDirectory:
def test_default_wdir_is_not_written(self, tmp_dir, dvc):
stage = dvc.run(
cmd="echo test > foo",
outs=["foo"],
wdir=".",
single_stage=True,
)
d = load_yaml(stage.relpath)
assert Stage.PARAM_WDIR not in d.keys()
stage = dvc.run(cmd="echo test > bar", outs=["bar"], single_stage=True)
d = load_yaml(stage.relpath)
assert Stage.PARAM_WDIR not in d.keys()
def test_fname_changes_path_and_wdir(self, tmp_dir, dvc):
dname = "dir"
os.mkdir(os.path.join(tmp_dir, dname))
foo = os.path.join(dname, "foo")
fname = os.path.join(dname, "stage" + DVC_FILE_SUFFIX)
stage = dvc.run(
cmd=f"echo test > {foo}",
outs=[foo],
fname=fname,
single_stage=True,
)
assert stage.wdir == os.path.realpath(tmp_dir)
assert stage.path == os.path.join(os.path.realpath(tmp_dir), fname)
# Check that it is dumped properly (relative to fname)
d = load_yaml(stage.relpath)
assert d[Stage.PARAM_WDIR] == ".."
def test_rerun_deterministic(tmp_dir, run_copy, mocker):
from dvc.stage.run import subprocess
tmp_dir.gen("foo", "foo content")
spy = mocker.spy(subprocess, "Popen")
run_copy("foo", "out", single_stage=True)
assert spy.called
spy.reset_mock()
run_copy("foo", "out", single_stage=True)
assert not spy.called
def test_rerun_deterministic_ignore_cache(tmp_dir, run_copy, mocker):
from dvc.stage.run import subprocess
tmp_dir.gen("foo", "foo content")
spy = mocker.spy(subprocess, "Popen")
run_copy("foo", "out", single_stage=True)
assert spy.called
spy.reset_mock()
run_copy("foo", "out", run_cache=False, single_stage=True)
assert spy.called
def test_rerun_callback(dvc):
def run_callback(force=False):
return dvc.run(
cmd="echo content > out", force=force, fname="echo.dvc", single_stage=True
)
assert run_callback() is not None
with pytest.raises(StageFileAlreadyExistsError):
assert run_callback() is not None
assert run_callback(force=True) is not None
def test_rerun_changed_dep(tmp_dir, run_copy):
tmp_dir.gen("foo", "foo content")
assert run_copy("foo", "out", single_stage=True) is not None
tmp_dir.gen("foo", "changed content")
with pytest.raises(StageFileAlreadyExistsError):
run_copy("foo", "out", force=False, single_stage=True)
assert run_copy("foo", "out", force=True, single_stage=True)
def test_rerun_changed_stage(tmp_dir, run_copy):
tmp_dir.gen("foo", "foo content")
assert run_copy("foo", "out", single_stage=True) is not None
tmp_dir.gen("bar", "bar content")
with pytest.raises(StageFileAlreadyExistsError):
run_copy("bar", "out", force=False, single_stage=True)
def test_rerun_changed_out(tmp_dir, run_copy):
tmp_dir.gen("foo", "foo content")
assert run_copy("foo", "out", single_stage=True) is not None
Path("out").write_text("modification", encoding="utf-8")
with pytest.raises(StageFileAlreadyExistsError):
run_copy("foo", "out", force=False, single_stage=True)
def test_should_raise_on_overlapping_output_paths(tmp_dir, dvc, append_foo_script):
tmp_dir.gen("data", {"foo": "foo", "bar": "bar"})
ret = main(["add", "data"])
assert ret == 0
foo_file = os.path.join("data", "foo")
with pytest.raises(OverlappingOutputPathsError) as err:
dvc.run(
outs=["data/foo"],
cmd=f"python append_foo.py {foo_file}",
single_stage=True,
)
error_output = str(err.value)
assert "The output paths:\n" in error_output
assert "\n'data'('data.dvc')\n" in error_output
assert f"\n'{foo_file}'('foo.dvc')\n" in error_output
assert (
"overlap and are thus in the same tracked directory.\n"
"To keep reproducibility, outputs should be in separate "
"tracked directories or tracked individually." in error_output
)
def test_should_not_checkout_upon_corrupted_local_hardlink_cache(
mocker, tmp_dir, dvc, copy_script
):
tmp_dir.gen("foo", "foo")
dvc.cache.local.cache_types = ["hardlink"]
stage = dvc.run(
deps=["foo"],
outs=["bar"],
cmd="python copy.py foo bar",
single_stage=True,
)
os.chmod("bar", 0o644)
with open("bar", "w", encoding="utf-8") as fd:
fd.write("corrupting the output cache")
spy_checkout = mocker.spy(stage.outs[0], "checkout")
from dvc.stage import run as stage_run
spy_run = mocker.spy(stage_run, "cmd_run")
with dvc.lock:
stage.run()
spy_run.assert_called_once()
spy_checkout.assert_not_called()
def test_bad_stage_fname(tmp_dir, dvc, run_copy):
tmp_dir.dvc_gen("foo", "foo content")
with pytest.raises(StageFileBadNameError):
# fname should end with .dvc
run_copy("foo", "foo_copy", fname="out_stage", single_stage=True)
# Check that command hasn't been run
assert not (tmp_dir / "foo_copy").exists()
def test_should_raise_on_stage_dependency(run_copy):
with pytest.raises(DependencyIsStageFileError):
run_copy("name.dvc", "stage_copy", single_stage=True)
def test_should_raise_on_stage_output(tmp_dir, dvc, run_copy):
tmp_dir.dvc_gen("foo", "foo content")
with pytest.raises(OutputIsStageFileError):
run_copy("foo", "name.dvc", single_stage=True)
@pytest.mark.parametrize("metrics_type", ["metrics", "metrics_no_cache"])
def test_metrics_dir(tmp_dir, dvc, caplog, run_copy_metrics, metrics_type):
copyargs = {metrics_type: ["dir_metric"]}
tmp_dir.gen({"dir": {"file": "content"}})
with caplog.at_level(logging.DEBUG, "dvc"):
run_copy_metrics("dir", "dir_metric", **copyargs)
assert "directory 'dir_metric' cannot be used as metrics." in caplog.messages
def test_run_force_preserves_comments_and_meta(tmp_dir, dvc, run_copy):
tmp_dir.gen({"foo": "foo", "foo1": "foo1"})
text = textwrap.dedent(
"""\
desc: top desc
cmd: python copy.py foo bar
deps:
- path: copy.py
- path: foo
outs:
# comment preserved
- path: bar
desc: out desc
type: mytype
labels:
- label1
- label2
meta:
key: value
meta:
name: copy-foo-bar
"""
)
(tmp_dir / "bar.dvc").write_text(text)
dvc.reproduce("bar.dvc")
# CRLF on windows makes the generated file bigger in size
code_size = 176 if os.name == "nt" else 167
assert (tmp_dir / "bar.dvc").read_text() == textwrap.dedent(
f"""\
desc: top desc
cmd: python copy.py foo bar
deps:
- path: copy.py
md5: a618d3a2c3d5a35f0aa4707951d986f5
size: {code_size}
- path: foo
md5: acbd18db4cc2f85cedef654fccc4a4d8
size: 3
outs:
# comment preserved
- path: bar
desc: out desc
type: mytype
labels:
- label1
- label2
meta:
key: value
md5: acbd18db4cc2f85cedef654fccc4a4d8
size: 3
meta:
name: copy-foo-bar
md5: 262f10a31f4b218b7b450b3511c2413f
"""
)
run_copy("foo1", "bar1", single_stage=True, force=True, fname="bar.dvc")
assert (tmp_dir / "bar.dvc").read_text() == textwrap.dedent(
f"""\
desc: top desc
cmd: python copy.py foo1 bar1
deps:
- path: foo1
md5: 299a0be4a5a79e6a59fdd251b19d78bb
size: 4
- path: copy.py
md5: a618d3a2c3d5a35f0aa4707951d986f5
size: {code_size}
outs:
# comment preserved
- path: bar1
md5: 299a0be4a5a79e6a59fdd251b19d78bb
size: 4
meta:
name: copy-foo-bar
md5: 61a1506995d7550a366b80d2301530b7
"""
)
tests/unit/repo/test_run.py
-12
View File
@@ -1,12 +0,0 @@
import pytest
from dvc.exceptions import InvalidArgumentError
def test_file(tmp_dir, dvc):
msg = (
"`--file` is currently incompatible with `-n|--name` "
"and requires `--single-stage`"
)
with pytest.raises(InvalidArgumentError, match=msg):
dvc.run(fname="path/dvc.yaml", name="my", cmd="mycmd")