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

dvc: add a support for multistage dvcfile. (#3584)

Browse Source 
Now, pipeline stage files can house multiple stages, and separate lockfiles
are created which has the checksums, whereas Dvcfile will be clean and human
readable and editable. The *.dvc files will be generated for output files.

It is available via a hidden flag: -n.

Fixes #1871
PR: #3584
This commit is contained in:
Saugat Pachhai
2020-04-17 22:58:36 +03:00
committed by GitHub
parent 11e56dbdf9
commit 128fa7f887
23 changed files with 1734 additions and 292 deletions
Download Patch File Download Diff File Expand all files Collapse all files
dvc/command/pipeline.py
+24 -11
View File
@@ -8,15 +8,26 @@ from dvc.exceptions import DvcException
logger = logging.getLogger(__name__)
def _stage_repr(stage):
from dvc.stage import PipelineStage
return (
"{}:{}".format(stage.relpath, stage.name)
if isinstance(stage, PipelineStage)
else stage.relpath
)
class CmdPipelineShow(CmdBase):
def _show(self, target, commands, outs, locked):
import networkx
from dvc.dvcfile import Dvcfile
from dvc import dvcfile
from dvc.utils import parse_target
stage = Dvcfile(self.repo, target).load()
G = self.repo.graph
path, name = parse_target(target)
stage = dvcfile.Dvcfile(self.repo, path).load_one(name)
G = self.repo.pipeline_graph
stages = networkx.dfs_postorder_nodes(G, stage)
if locked:
stages = [s for s in stages if s.locked]
@@ -29,14 +40,16 @@ class CmdPipelineShow(CmdBase):
for out in stage.outs:
logger.info(str(out))
else:
logger.info(stage.path_in_repo)
logger.info(_stage_repr(stage))
def _build_graph(self, target, commands, outs):
def _build_graph(self, target, commands=False, outs=False):
import networkx
from dvc.dvcfile import Dvcfile
from dvc import dvcfile
from dvc.repo.graph import get_pipeline
from dvc.utils import parse_target
target_stage = Dvcfile(self.repo, target).load()
path, name = parse_target(target)
target_stage = dvcfile.Dvcfile(self.repo, path).load_one(name)
G = get_pipeline(self.repo.pipelines, target_stage)
nodes = set()
@@ -49,7 +62,7 @@ class CmdPipelineShow(CmdBase):
for out in stage.outs:
nodes.add(str(out))
else:
nodes.add(stage.relpath)
nodes.add(_stage_repr(stage))
edges = []
for from_stage, to_stage in networkx.edge_dfs(G, target_stage):
@@ -62,7 +75,7 @@ class CmdPipelineShow(CmdBase):
for to_out in to_stage.outs:
edges.append((str(from_out), str(to_out)))
else:
edges.append((from_stage.relpath, to_stage.relpath))
edges.append((_stage_repr(from_stage), _stage_repr(to_stage)))
return list(nodes), edges, networkx.is_tree(G)
@@ -150,7 +163,7 @@ class CmdPipelineList(CmdBase):
pipelines = self.repo.pipelines
for pipeline in pipelines:
for stage in pipeline:
logger.info(stage.relpath)
logger.info(_stage_repr(stage))
if len(pipeline) != 0:
logger.info("=" * 80)
logger.info("{} pipelines total".format(len(pipelines)))
dvc/command/run.py
+2
View File
@@ -51,6 +51,7 @@ class CmdRun(CmdBase):
outs_persist=self.args.outs_persist,
outs_persist_no_cache=self.args.outs_persist_no_cache,
always_changed=self.args.always_changed,
name=self.args.name,
)
except DvcException:
logger.exception("failed to run command")
@@ -96,6 +97,7 @@ def add_parser(subparsers, parent_parser):
default=[],
help="Declare dependencies for reproducible cmd.",
)
run_parser.add_argument("-n", "--name", help=argparse.SUPPRESS)
run_parser.add_argument(
"-o",
"--outs",
dvc/dvcfile.py
+239 -27
View File
@@ -2,11 +2,14 @@ import os
import re
import logging
from funcy import project
import dvc.prompt as prompt
from voluptuous import MultipleInvalid
from dvc import dependency, output
from dvc.exceptions import DvcException
from dvc.stage.exceptions import (
StageFileBadNameError,
StageFileDoesNotExistError,
@@ -29,6 +32,39 @@ DVC_FILE_SUFFIX = ".dvc"
TAG_REGEX = r"^(?P<path>.*)@(?P<tag>[^\\/@:]*)$"
class MultiStageFileLoadError(DvcException):
def __init__(self):
super().__init__("Cannot load multi-stage file.")
def _serialize_stage(stage):
outs_bucket = {}
for o in stage.outs:
bucket_key = ["metrics"] if o.metric else ["outs"]
if not o.metric and o.persist:
bucket_key += ["persist"]
if not o.use_cache:
bucket_key += ["no_cache"]
key = "_".join(bucket_key)
outs_bucket[key] = outs_bucket.get(key, []) + [o.def_path]
return {
stage.name: {
key: value
for key, value in {
stage.PARAM_CMD: stage.cmd,
stage.PARAM_WDIR: stage.resolve_wdir(),
stage.PARAM_DEPS: [d.def_path for d in stage.deps],
**outs_bucket,
stage.PARAM_LOCKED: stage.locked,
stage.PARAM_ALWAYS_CHANGED: stage.always_changed,
}.items()
if value
}
}
class Dvcfile:
def __init__(self, repo, path):
self.repo = repo
@@ -77,9 +113,61 @@ class Dvcfile:
return s, None
return match.group("path"), match.group("tag")
def dump(self, stage):
@property
def lockfile(self):
return os.path.splitext(self.path)[0] + ".lock"
def dump(self, stage, update_dvcfile=False):
"""Dumps given stage appropriately in the dvcfile."""
self.dump_single_stage(stage)
if not hasattr(stage, "name"):
self.dump_single_stage(stage)
return
self.dump_lockfile(stage)
if update_dvcfile and not stage.is_data_source:
self.dump_multistage_dvcfile(stage)
from .stage import Stage, create_stage
for out in stage.outs:
if not out.use_cache:
continue
s = create_stage(
Stage,
stage.repo,
os.path.join(stage.wdir, out.def_path + DVC_FILE_SUFFIX),
wdir=stage.wdir,
)
s.outs = [out]
s.md5 = s._compute_md5()
Dvcfile(s.repo, s.path).dump_single_stage(s)
def dump_lockfile(self, stage):
from . import lockfile
lockfile.dump(self.repo, self.lockfile, stage)
self.repo.scm.track_file(relpath(self.lockfile))
def dump_multistage_dvcfile(self, stage):
from dvc.utils.stage import parse_stage_for_update, dump_stage_file
from dvc.schema import COMPILED_MULTI_STAGE_SCHEMA
path = self.path
if not os.path.exists(path):
open(path, "w+").close()
with open(path, "r") as fd:
data = parse_stage_for_update(fd.read(), path)
if not self.is_multi_stage(data):
raise MultiStageFileLoadError
# handle this in Stage::dumpd()
data["stages"] = data.get("stages", {})
data["stages"].update(_serialize_stage(stage))
dump_stage_file(path, COMPILED_MULTI_STAGE_SCHEMA(data))
self.repo.scm.track_file(relpath(path))
def dump_single_stage(self, stage):
self.check_dvc_filename(self.path)
@@ -108,10 +196,7 @@ class Dvcfile:
self.repo.scm.track_file(relpath(self.path))
def load(self):
"""Loads single stage."""
from dvc.stage import Stage
def _load(self):
# it raises the proper exceptions by priority:
# 1. when the file doesn't exists
# 2. filename is not a DVC-file
@@ -123,27 +208,71 @@ class Dvcfile:
with self.repo.tree.open(self.path) as fd:
stage_text = fd.read()
d = parse_stage(stage_text, self.path)
return d, stage_text
Dvcfile.validate(d, fname=relpath(self.path))
path = os.path.abspath(self.path)
stage = Stage(
repo=self.repo,
path=path,
wdir=os.path.abspath(
os.path.join(
os.path.dirname(path), d.get(Stage.PARAM_WDIR, ".")
def load_one(self, target=None):
data, raw = self._load()
if not self.is_multi_stage(data):
if target:
logger.warning(
"Ignoring target name '%s' as it's a single stage file.",
target,
)
),
cmd=d.get(Stage.PARAM_CMD),
md5=d.get(Stage.PARAM_MD5),
locked=d.get(Stage.PARAM_LOCKED, False),
tag=self.tag,
always_changed=d.get(Stage.PARAM_ALWAYS_CHANGED, False),
# We store stage text to apply updates to the same structure
stage_text=stage_text,
return self._load_single_stage(data, raw)
if not target:
raise DvcException(
"No target provided for multi-stage file '{}'.".format(
self.path
)
)
if not self.has_stage(name=target, data=data):
raise DvcException(
"Target '{}' does not exist "
"inside '{}' multi-stage file.".format(target, self.path)
)
stages = self._load_multi_stage(
{"stages": {target: self._get_stage_data(target, data)}}
)
assert stages
return stages[0]
@staticmethod
def _get_stage_data(name, data):
return data.get("stages", {}).get(name)
def has_stage(self, name, data=None):
if not data:
data, _ = self._load()
return bool(self._get_stage_data(name, data))
def load(self):
"""Loads single stage."""
data, raw = self._load()
if not self.is_multi_stage(data):
return self._load_single_stage(data, raw)
raise MultiStageFileLoadError
def load_all(self):
data, raw = self._load()
return (
[self._load_single_stage(data, raw)]
if not self.is_multi_stage(data)
else self._load_multi_stage(data)
)
def _load_single_stage(self, d: dict, stage_text: str):
from dvc.stage import Stage, loads_from
path = os.path.abspath(self.path)
wdir = os.path.abspath(
os.path.join(os.path.dirname(path), d.get(Stage.PARAM_WDIR, "."))
)
stage = loads_from(Stage, self.repo, path, wdir, d)
stage._stage_text, stage.tag = stage_text, self.tag
stage.deps = dependency.loadd_from(
stage, d.get(Stage.PARAM_DEPS) or []
)
@@ -151,22 +280,105 @@ class Dvcfile:
return stage
def load_multi(self):
data, _ = self._load()
if self.is_multi_stage(data):
return self._load_multi_stage(data)
raise DvcException(
"Cannot load multiple stages from single stage file."
)
def _load_multi_stage(self, data):
from . import lockfile
from .stage import PipelineStage, Stage, loads_from
stages = []
path = os.path.abspath(self.path)
lock_data = lockfile.load(self.repo, self.lockfile)
for stage_name, d in data.get("stages", {}).items():
lock_stage_data = lock_data.get(stage_name, {})
wdir = os.path.abspath(
os.path.join(
os.path.dirname(path), d.get(Stage.PARAM_WDIR, ".")
)
)
stage = loads_from(PipelineStage, self.repo, path, wdir, d)
stage.name = stage_name
stage.cmd_changed = lock_stage_data.get(Stage.PARAM_CMD) != d.get(
Stage.PARAM_CMD
)
stage._fill_stage_dependencies(**project(d, ["deps"]))
stage._fill_stage_outputs(**d)
stages.append(stage)
for dep in stage.deps:
dep.info[dep.remote.PARAM_CHECKSUM] = lock_stage_data.get(
Stage.PARAM_DEPS, {}
).get(dep.def_path)
if stage.cmd_changed:
continue
for out in stage.outs:
out.info[out.remote.PARAM_CHECKSUM] = lock_stage_data.get(
Stage.PARAM_OUTS, {}
).get(out.def_path)
return stages
@staticmethod
def validate(d, fname=None):
from dvc.stage.schema import SINGLE_STAGE_SCHEMA
def validate_single_stage(d, fname=None):
from dvc.schema import COMPILED_SINGLE_STAGE_SCHEMA
try:
SINGLE_STAGE_SCHEMA(d)
COMPILED_SINGLE_STAGE_SCHEMA(d)
except MultipleInvalid as exc:
raise StageFileFormatError(fname, exc)
@staticmethod
def validate_multi_stage(d, fname=None):
from dvc.schema import COMPILED_MULTI_STAGE_SCHEMA
try:
COMPILED_MULTI_STAGE_SCHEMA(d)
except MultipleInvalid as exc:
raise StageFileFormatError(fname, exc)
@staticmethod
def validate(d, fname=None):
Dvcfile.validate_single_stage(d, fname)
def is_multi_stage(self, d=None):
# TODO: maybe the following heuristics is enough?
if d is None:
d = self._load()[0]
check_multi_stage = d.get("stages") or not d
exc = None
if check_multi_stage:
try:
self.validate_multi_stage(d, self.path)
return True
except StageFileFormatError as _exc:
exc = _exc
try:
self.validate_single_stage(d, self.path)
return False
except StageFileFormatError:
if check_multi_stage:
raise exc
self.validate_multi_stage(d, self.path)
return True
def overwrite_with_prompt(self, force=False):
if not self.exists():
return
msg = (
"'{}' already exists. Do you wish to run the command and "
"overwrite it?".format(self.path)
"overwrite it?".format(relpath(self.path))
)
if not (force or prompt.confirm(msg)):
raise StageFileAlreadyExistsError(self.path)
dvc/lockfile.py
+82
View File
@@ -0,0 +1,82 @@
import json
from typing import TYPE_CHECKING
from collections import OrderedDict
from dvc.exceptions import DvcException
from dvc.schema import COMPILED_LOCKFILE_SCHEMA
from voluptuous import MultipleInvalid
if TYPE_CHECKING:
from dvc.repo import Repo
from dvc.stage import PipelineStage
class LockfileCorruptedError(DvcException):
def __init__(self, path):
super().__init__("Lockfile '{}' is corrupted.".format(path))
def serialize_stage(stage: "PipelineStage") -> OrderedDict:
assert stage.cmd
assert stage.name
deps = OrderedDict(
[
(dep.def_path, dep.remote.get_checksum(dep.path_info),)
for dep in stage.deps
if dep.remote.get_checksum(dep.path_info)
]
)
outs = OrderedDict(
[
(out.def_path, out.remote.get_checksum(out.path_info),)
for out in stage.outs
if out.remote.get_checksum(out.path_info)
]
)
return OrderedDict(
[
(
stage.name,
OrderedDict(
[("cmd", stage.cmd), ("deps", deps,), ("outs", outs)]
),
)
]
)
def exists(repo: "Repo", path: str) -> bool:
return repo.tree.exists(path)
def read(repo: "Repo", path: str) -> dict:
with repo.tree.open(path) as f:
return json.load(f, object_pairs_hook=OrderedDict)
def write(repo: "Repo", path: str, data: dict) -> dict:
with repo.tree.open(path, "w+") as f:
json.dump(data, f)
def load(repo: "Repo", path: str) -> dict:
if not exists(repo, path):
return {}
try:
return COMPILED_LOCKFILE_SCHEMA(read(repo, path))
except MultipleInvalid:
raise LockfileCorruptedError(path)
def dump(repo: "Repo", path: str, stage: "PipelineStage"):
stage_data = serialize_stage(stage)
if not exists(repo, path):
data = stage_data
else:
data = read(repo, path)
data.update(stage_data)
write(repo, path, COMPILED_LOCKFILE_SCHEMA(data))
dvc/repo/__init__.py
+85 -15
View File
@@ -174,7 +174,7 @@ class Repo(object):
self.scm.ignore_list(flist)
def check_modified_graph(self, new_stages):
def check_modified_graph(self, new_stages, old_stages=None):
"""Generate graph including the new stage to check for errors"""
# Building graph might be costly for the ones with many DVC-files,
# so we provide this undocumented hack to skip it. See [1] for
@@ -189,7 +189,29 @@ class Repo(object):
#
# [1] https://github.com/iterative/dvc/issues/2671
if not getattr(self, "_skip_graph_checks", False):
self._collect_graph(self.stages + new_stages)
self._collect_graph((old_stages or self.stages) + new_stages)
def _collect_inside(self, path, graph):
import networkx as nx
stages = nx.dfs_postorder_nodes(graph or self.pipeline_graph)
return [stage for stage in stages if path_isin(stage.path, path)]
def collect_for_pipelines(
self, path=None, name=None, recursive=False, graph=None
):
from ..dvcfile import Dvcfile
if not path:
return list(graph) if graph else self.pipeline_stages
path = os.path.abspath(path)
if recursive and os.path.isdir(path):
return self._collect_inside(path, graph or self.pipeline_graph)
dvcfile = Dvcfile(self, path)
dvcfile.check_file_exists()
return [dvcfile.load_one(name)]
def collect(self, target, with_deps=False, recursive=False, graph=None):
import networkx as nx
@@ -201,8 +223,7 @@ class Repo(object):
target = os.path.abspath(target)
if recursive and os.path.isdir(target):
stages = nx.dfs_postorder_nodes(graph or self.graph)
return [stage for stage in stages if path_isin(stage.path, target)]
return self._collect_inside(target, graph or self.graph)
stage = Dvcfile(self, target).load()
@@ -388,18 +409,17 @@ class Repo(object):
overlapping.extend(outs.values(prefix=dep_key))
G.add_edges_from((stage, out.stage) for out in overlapping)
check_acyclic(G)
return G
@cached_property
def graph(self):
return self._collect_graph()
return self._collect_graph(self.stages)
@cached_property
def pipelines(self):
return get_pipelines(self.graph)
return get_pipelines(self.pipeline_graph)
@cached_property
def stages(self):
@@ -412,9 +432,24 @@ class Repo(object):
NOTE: For large repos, this could be an expensive
operation. Consider using some memoization.
"""
from ..dvcfile import Dvcfile
return self._collect_stages()[0]
stages = []
@cached_property
def pipeline_stages(self):
return self._collect_stages()[1]
@cached_property
def pipeline_graph(self):
return self._collect_graph(self.pipeline_stages)
def _collect_stages(self):
from dvc.dvcfile import Dvcfile
from dvc.stage import PipelineStage
pipeline_stages = []
single_stages = []
output_stages = []
ignored_outs = []
outs = set()
for root, dirs, files in self.tree.walk(self.root_dir):
@@ -422,16 +457,49 @@ class Repo(object):
path = os.path.join(root, fname)
if not Dvcfile.is_valid_filename(path):
continue
stage = Dvcfile(self, path).load()
stages.append(stage)
stgs = Dvcfile(self, path).load_all()
ignored_outs.extend(
out
for stage in stgs
if isinstance(stage, PipelineStage)
for out in stage.outs
)
for out in stage.outs:
if out.scheme == "local":
outs.add(out.fspath)
for stage in stgs:
stages = (
output_stages
if stage.is_data_source
else pipeline_stages
)
stages.append(stage)
if not (
isinstance(stage, PipelineStage)
or stage.is_data_source
):
single_stages.append(stage)
for out in stage.outs:
if out.scheme == "local":
outs.add(out.fspath)
dirs[:] = [d for d in dirs if os.path.join(root, d) not in outs]
return stages
# DVC files are generated by multi-stage for data management.
# We need to ignore those stages for pipelines_stages, but still
# should be collected for output stages.
_output_stages = [
stage
for stage in output_stages
if all(
stage.outs and out.fspath != stage.outs[0].fspath
for out in ignored_outs
)
]
# Old single-stages are used for both outputs and pipelines
# so they go into both buckets: pipeline_stages and output stages.
return (
output_stages + single_stages,
pipeline_stages + _output_stages,
)
def find_outs_by_path(self, path, outs=None, recursive=False, strict=True):
if not outs:
@@ -528,3 +596,5 @@ class Repo(object):
self.__dict__.pop("stages", None)
self.__dict__.pop("pipelines", None)
self.__dict__.pop("dvcignore", None)
self.__dict__.pop("pipeline_graph", None)
self.__dict__.pop("pipeline_stages", None)
dvc/repo/add.py
+2 -2
View File
@@ -115,7 +115,7 @@ def _find_all_targets(repo, target, recursive):
def _create_stages(repo, targets, fname, pbar=None):
from dvc.stage import Stage
from dvc.stage import Stage, create_stage
stages = []
@@ -126,7 +126,7 @@ def _create_stages(repo, targets, fname, pbar=None):
unit="file",
):
path, wdir, out = resolve_paths(repo, out)
stage = Stage.create(repo, fname or path, wdir=wdir, outs=[out])
stage = create_stage(Stage, repo, fname or path, wdir=wdir, outs=[out])
if stage:
Dvcfile(repo, stage.path).overwrite_with_prompt(force=True)
dvc/repo/imp_url.py
+9 -3
View File
@@ -10,7 +10,7 @@ from dvc.utils.fs import path_isin
@scm_context
def imp_url(self, url, out=None, fname=None, erepo=None, locked=True):
from dvc.dvcfile import Dvcfile
from dvc.stage import Stage
from dvc.stage import Stage, create_stage
out = resolve_output(url, out)
path, wdir, out = resolve_paths(self, out)
@@ -19,8 +19,14 @@ def imp_url(self, url, out=None, fname=None, erepo=None, locked=True):
if os.path.exists(url) and path_isin(os.path.abspath(url), self.root_dir):
url = relpath(url, wdir)
stage = Stage.create(
self, fname or path, wdir=wdir, deps=[url], outs=[out], erepo=erepo,
stage = create_stage(
Stage,
self,
fname or path,
wdir=wdir,
deps=[url],
outs=[out],
erepo=erepo,
)
if stage is None:
dvc/repo/lock.py
+6 -4
View File
@@ -3,11 +3,13 @@ from . import locked
@locked
def lock(self, target, unlock=False):
from ..dvcfile import Dvcfile
from .. import dvcfile
from dvc.utils import parse_target
dvcfile = Dvcfile(self, target)
stage = dvcfile.load()
path, target = parse_target(target)
dvcfile = dvcfile.Dvcfile(self, path)
stage = dvcfile.load_one(target)
stage.locked = False if unlock else True
dvcfile.dump(stage)
dvcfile.dump(stage, update_dvcfile=True)
return stage
dvc/repo/reproduce.py
+8 -4
View File
@@ -61,7 +61,8 @@ def reproduce(
all_pipelines=False,
**kwargs
):
from ..dvcfile import Dvcfile
from .. import dvcfile
from dvc.utils import parse_target
if not target and not all_pipelines:
raise InvalidArgumentError(
@@ -72,14 +73,15 @@ def reproduce(
if not interactive:
kwargs["interactive"] = self.config["core"].get("interactive", False)
active_graph = _get_active_graph(self.graph)
active_graph = _get_active_graph(self.pipeline_graph)
active_pipelines = get_pipelines(active_graph)
path, name = parse_target(target)
if pipeline or all_pipelines:
if all_pipelines:
pipelines = active_pipelines
else:
stage = Dvcfile(self, target).load()
stage = dvcfile.Dvcfile(self, path).load_one(name)
pipelines = [get_pipeline(active_pipelines, stage)]
targets = []
@@ -88,7 +90,9 @@ def reproduce(
if pipeline.in_degree(stage) == 0:
targets.append(stage)
else:
targets = self.collect(target, recursive=recursive, graph=active_graph)
targets = self.collect_for_pipelines(
path, name=name, recursive=recursive, graph=active_graph
)
ret = []
for target in targets:
dvc/repo/run.py
+44 -24
View File
@@ -2,41 +2,61 @@ import os
from . import locked
from .scm_context import scm_context
from ..exceptions import DvcException
from ..utils import relpath
from funcy import first, concat
def _get_file_path(**kwargs):
from dvc.dvcfile import DVC_FILE_SUFFIX, DVC_FILE
out = first(
concat(
kwargs.get("outs", []),
kwargs.get("outs_no_cache", []),
kwargs.get("metrics", []),
kwargs.get("metrics_no_cache", []),
kwargs.get("outs_persist", []),
kwargs.get("outs_persist_no_cache", []),
)
)
return (
os.path.basename(os.path.normpath(out)) + DVC_FILE_SUFFIX
if out
else DVC_FILE
)
@locked
@scm_context
def run(self, fname=None, no_exec=False, **kwargs):
from dvc.stage import Stage
from dvc.dvcfile import Dvcfile, DVC_FILE_SUFFIX, DVC_FILE
from dvc.stage import PipelineStage, Stage, create_stage
from dvc.dvcfile import DVC_FILE, Dvcfile
outs = (
kwargs.get("outs", [])
+ kwargs.get("outs_no_cache", [])
+ kwargs.get("metrics", [])
+ kwargs.get("metrics_no_cache", [])
+ kwargs.get("outs_persist", [])
+ kwargs.get("outs_persist_no_cache", [])
)
stage_cls, path = PipelineStage, fname or DVC_FILE
if not kwargs.get("name"):
kwargs.pop("name", None)
stage_cls, path = Stage, fname or _get_file_path(**kwargs)
if outs:
base = os.path.basename(os.path.normpath(outs[0]))
path = base + DVC_FILE_SUFFIX
else:
path = DVC_FILE
stage = Stage.create(self, fname or path, **kwargs)
stage = create_stage(stage_cls, repo=self, path=path, **kwargs)
if stage is None:
return None
dvcfile = Dvcfile(self, stage.path)
dvcfile.overwrite_with_prompt(force=kwargs.get("overwrite", True))
self.check_modified_graph([stage])
dvcfile = Dvcfile(self, path)
if dvcfile.exists() and not dvcfile.is_multi_stage():
if stage_cls == PipelineStage:
raise DvcException(
"'{}' is a single-stage dvcfile. Please use "
"`-f <different-filename>` and try again.`.".format(
relpath(dvcfile.path)
)
)
dvcfile.overwrite_with_prompt(force=kwargs.get("overwrite", True))
self.check_modified_graph([stage], self.pipeline_stages)
if not no_exec:
stage.run(no_commit=kwargs.get("no_commit", False))
dvcfile.dump(stage)
dvcfile.dump(stage, update_dvcfile=True)
return stage
dvc/schema.py
+41
View File
@@ -0,0 +1,41 @@
from dvc.stage.params import StageParams, OutputParams
from dvc import dependency, output
from voluptuous import Any, Schema, Optional, Required
STAGES = "stages"
SINGLE_STAGE_SCHEMA = {
StageParams.PARAM_MD5: output.CHECKSUM_SCHEMA,
StageParams.PARAM_CMD: Any(str, None),
StageParams.PARAM_WDIR: Any(str, None),
StageParams.PARAM_DEPS: Any([dependency.SCHEMA], None),
StageParams.PARAM_OUTS: Any([output.SCHEMA], None),
StageParams.PARAM_LOCKED: bool,
StageParams.PARAM_META: object,
StageParams.PARAM_ALWAYS_CHANGED: bool,
}
LOCK_FILE_STAGE_SCHEMA = {
Required(StageParams.PARAM_CMD): str,
Required(StageParams.PARAM_DEPS): {str: output.CHECKSUM_SCHEMA},
Required(StageParams.PARAM_OUTS): {str: output.CHECKSUM_SCHEMA},
}
LOCKFILE_SCHEMA = {str: LOCK_FILE_STAGE_SCHEMA}
SINGLE_PIPELINE_STAGE_SCHEMA = {
str: {
StageParams.PARAM_CMD: str,
Optional(StageParams.PARAM_WDIR): str,
Optional(StageParams.PARAM_DEPS): [str],
Optional(StageParams.PARAM_LOCKED): bool,
Optional(StageParams.PARAM_META): object,
Optional(StageParams.PARAM_ALWAYS_CHANGED): bool,
**{Optional(p.value): [str] for p in OutputParams},
}
}
MULTI_STAGE_SCHEMA = {STAGES: SINGLE_PIPELINE_STAGE_SCHEMA}
COMPILED_SINGLE_STAGE_SCHEMA = Schema(SINGLE_STAGE_SCHEMA)
COMPILED_MULTI_STAGE_SCHEMA = Schema(MULTI_STAGE_SCHEMA)
COMPILED_LOCKFILE_SCHEMA = Schema(LOCKFILE_SCHEMA)
dvc/stage/__init__.py
+136 -83
View File
@@ -7,6 +7,8 @@ import threading
from itertools import chain
from funcy import project
import dvc.dependency as dependency
import dvc.output as output
import dvc.prompt as prompt
@@ -22,17 +24,66 @@ from .exceptions import (
MissingDep,
MissingDataSource,
)
from . import schema
from . import params
from dvc.utils import dict_md5
from dvc.utils import fix_env
from dvc.utils import relpath
from dvc.utils.fs import path_isin
from .params import OutputParams
logger = logging.getLogger(__name__)
class Stage(schema.StageParams):
def loads_from(cls, repo, path, wdir, data):
kw = {
"repo": repo,
"path": path,
"wdir": wdir,
**project(
data,
[
Stage.PARAM_CMD,
Stage.PARAM_LOCKED,
Stage.PARAM_ALWAYS_CHANGED,
Stage.PARAM_MD5,
"name",
],
),
}
return cls(**kw)
def create_stage(cls, repo, path, **kwargs):
from dvc.dvcfile import Dvcfile
wdir = os.path.abspath(kwargs.get("wdir", None) or os.curdir)
path = os.path.abspath(path)
Dvcfile.check_dvc_filename(path)
cls._check_stage_path(repo, wdir, is_wdir=kwargs.get("wdir"))
cls._check_stage_path(repo, os.path.dirname(path))
stage = loads_from(cls, repo, path, wdir, kwargs)
stage._fill_stage_outputs(**kwargs)
stage._fill_stage_dependencies(**kwargs)
stage._check_circular_dependency()
stage._check_duplicated_arguments()
if stage and stage.dvcfile.exists():
has_persist_outs = any(out.persist for out in stage.outs)
ignore_build_cache = (
kwargs.get("ignore_build_cache", False) or has_persist_outs
)
if has_persist_outs:
logger.warning("Build cache is ignored when persisting outputs.")
if not ignore_build_cache and stage.can_be_skipped:
logger.info("Stage is cached, skipping.")
return None
return stage
class Stage(params.StageParams):
def __init__(
self,
repo,
@@ -46,6 +97,7 @@ class Stage(schema.StageParams):
tag=None,
always_changed=False,
stage_text=None,
dvcfile=None,
):
if deps is None:
deps = []
@@ -63,6 +115,7 @@ class Stage(schema.StageParams):
self.tag = tag
self.always_changed = always_changed
self._stage_text = stage_text
self._dvcfile = dvcfile
@property
def path(self):
@@ -72,6 +125,26 @@ class Stage(schema.StageParams):
def path(self, path):
self._path = path
@property
def dvcfile(self):
if self.path and self._dvcfile and self.path == self._dvcfile.path:
return self._dvcfile
if not self.path:
raise DvcException(
"Stage does not have any path set "
"and is detached from dvcfile."
)
from dvc.dvcfile import Dvcfile
self._dvcfile = Dvcfile(self.repo, self.path)
return self._dvcfile
@dvcfile.setter
def dvcfile(self, dvcfile):
self._dvcfile = dvcfile
def __repr__(self):
return "Stage: '{path}'".format(
path=self.relpath if self.path else "No path"
@@ -82,7 +155,7 @@ class Stage(schema.StageParams):
def __eq__(self, other):
return (
isinstance(other, Stage)
self.__class__ == other.__class__
and self.repo is other.repo
and self.path_in_repo == other.path_in_repo
)
@@ -173,18 +246,19 @@ class Stage(schema.StageParams):
@rwlocked(read=["deps", "outs"])
def changed(self):
if self._changed():
logger.warning("{} changed.".format(self))
return True
logger.debug("{} didn't change.".format(self))
return False
def _changed(self):
# Short-circuit order: stage md5 is fast, deps are expected to change
ret = (
return (
self._changed_md5() or self._changed_deps() or self._changed_outs()
)
if ret:
logger.warning("Stage '{}' changed.".format(self.relpath))
else:
logger.debug("Stage '{}' didn't change.".format(self.relpath))
return ret
@rwlocked(write=["outs"])
def remove_outs(self, ignore_remove=False, force=False):
"""Used mainly for `dvc remove --outs` and :func:`Stage.reproduce`."""
@@ -273,6 +347,9 @@ class Stage(schema.StageParams):
self.is_cached and not self.is_callback and not self.always_changed
)
def reload(self):
return self.dvcfile.load()
@property
def is_cached(self):
"""
@@ -280,9 +357,8 @@ class Stage(schema.StageParams):
"""
from dvc.remote.local import RemoteLOCAL
from dvc.remote.s3 import RemoteS3
from dvc.dvcfile import Dvcfile
old = Dvcfile(self.repo, self.path).load()
old = self.reload()
if old._changed_outs():
return False
@@ -319,66 +395,11 @@ class Stage(schema.StageParams):
return True
@staticmethod
def create(repo, path, **kwargs):
from dvc.dvcfile import Dvcfile
wdir = kwargs.get("wdir", None) or os.curdir
wdir = os.path.abspath(wdir)
path = os.path.abspath(path)
Dvcfile.check_dvc_filename(path)
Stage._check_stage_path(repo, wdir, is_wdir=kwargs.get("wdir"))
Stage._check_stage_path(repo, os.path.dirname(path))
stage = Stage(
repo=repo,
path=path,
wdir=wdir,
cmd=kwargs.get("cmd", None),
locked=kwargs.get("locked", False),
always_changed=kwargs.get("always_changed", False),
)
stage._fill_stage_outputs(**kwargs)
stage._fill_stage_dependencies(**kwargs)
stage._check_circular_dependency()
stage._check_duplicated_arguments()
Dvcfile.check_dvc_filename(path)
dvcfile = Dvcfile(stage.repo, stage.path)
if dvcfile.exists():
has_persist_outs = any(out.persist for out in stage.outs)
ignore_build_cache = (
kwargs.get("ignore_build_cache", False) or has_persist_outs
)
if has_persist_outs:
logger.warning(
"Build cache is ignored when persisting outputs."
)
if not ignore_build_cache and stage.can_be_skipped:
logger.info("Stage is cached, skipping.")
return None
return stage
def _fill_stage_outputs(self, **kwargs):
assert not self.outs
self.outs = []
for key in [
"outs",
"metrics",
"outs_persist",
"outs_no_cache",
"metrics_no_cache",
"outs_persist_no_cache",
]:
for key in (p.value for p in OutputParams):
self.outs += output.loads_from(
self,
kwargs.get(key, []),
@@ -400,18 +421,19 @@ class Stage(schema.StageParams):
if out.is_in_repo:
out.def_path = relpath(out.path_info, wdir)
def dumpd(self):
def resolve_wdir(self):
rel_wdir = relpath(self.wdir, os.path.dirname(self.path))
return (
pathlib.PurePath(rel_wdir).as_posix() if rel_wdir != "." else None
)
wdir = pathlib.PurePath(rel_wdir).as_posix()
wdir = wdir if wdir != "." else None
def dumpd(self):
return {
key: value
for key, value in {
Stage.PARAM_MD5: self.md5,
Stage.PARAM_CMD: self.cmd,
Stage.PARAM_WDIR: wdir,
Stage.PARAM_WDIR: self.resolve_wdir(),
Stage.PARAM_LOCKED: self.locked,
Stage.PARAM_DEPS: [d.dumpd() for d in self.deps],
Stage.PARAM_OUTS: [o.dumpd() for o in self.outs],
@@ -606,7 +628,11 @@ class Stage(schema.StageParams):
)
)
if not dry:
if not force and self._already_cached():
if (
not force
and not self._changed_md5()
and self._already_cached()
):
self.outs[0].checkout()
else:
self.deps[0].download(self.outs[0])
@@ -709,13 +735,9 @@ class Stage(schema.StageParams):
return {}
def _already_cached(self):
return (
not self.changed_md5()
and all(not dep.changed() for dep in self.deps)
and all(
not out.changed_cache() if out.use_cache else not out.changed()
for out in self.outs
)
return all(not dep.changed() for dep in self.deps) and all(
not out.changed_cache() if out.use_cache else not out.changed()
for out in self.outs
)
def get_all_files_number(self, filter_info=None):
@@ -732,3 +754,34 @@ class Stage(schema.StageParams):
cache.update(out.get_used_cache(*args, **kwargs))
return cache
class PipelineStage(Stage):
def __init__(self, name=None, **kwargs):
super().__init__(**kwargs)
self.name = name
self.cmd_changed = False
def __eq__(self, other):
return super().__eq__(other) and self.name == other.name
def __hash__(self):
return hash((self.path_in_repo, self.name))
def __repr__(self):
return "Stage: '{path}:{name}'".format(
path=self.relpath if self.path else "No path", name=self.name
)
def _changed(self):
if self.cmd_changed:
logger.warning("'cmd' of {} has changed.".format(self))
return self.cmd_changed or self._changed_deps() or self._changed_outs()
def reload(self):
return self.dvcfile.load_one(self.name)
@property
def is_cached(self):
return self.dvcfile.has_stage(name=self.name) and super().is_cached
dvc/stage/params.py
+21
View File
@@ -0,0 +1,21 @@
from enum import Enum
class StageParams:
PARAM_MD5 = "md5"
PARAM_CMD = "cmd"
PARAM_WDIR = "wdir"
PARAM_DEPS = "deps"
PARAM_OUTS = "outs"
PARAM_LOCKED = "locked"
PARAM_META = "meta"
PARAM_ALWAYS_CHANGED = "always_changed"
class OutputParams(Enum):
PERSIST = "outs_persist"
PERSIST_NO_CACHE = "outs_persist_no_cache"
METRICS_NO_CACHE = "metrics_no_cache"
METRICS = "metrics"
NO_CACHE = "outs_no_cache"
OUTS = "outs"
dvc/stage/schema.py
-28
View File
@@ -1,28 +0,0 @@
from voluptuous import Any, Schema
from dvc import dependency
from dvc import output
class StageParams:
PARAM_MD5 = "md5"
PARAM_CMD = "cmd"
PARAM_WDIR = "wdir"
PARAM_DEPS = "deps"
PARAM_OUTS = "outs"
PARAM_LOCKED = "locked"
PARAM_META = "meta"
PARAM_ALWAYS_CHANGED = "always_changed"
SCHEMA = {
StageParams.PARAM_MD5: output.CHECKSUM_SCHEMA,
StageParams.PARAM_CMD: Any(str, None),
StageParams.PARAM_WDIR: Any(str, None),
StageParams.PARAM_DEPS: Any([dependency.SCHEMA], None),
StageParams.PARAM_OUTS: Any([output.SCHEMA], None),
StageParams.PARAM_LOCKED: bool,
StageParams.PARAM_META: object,
StageParams.PARAM_ALWAYS_CHANGED: bool,
}
SINGLE_STAGE_SCHEMA = Schema(SCHEMA)
dvc/utils/__init__.py
+14
View File
@@ -23,6 +23,7 @@ logger = logging.getLogger(__name__)
LOCAL_CHUNK_SIZE = 2 ** 20 # 1 MB
LARGE_FILE_SIZE = 2 ** 30 # 1 GB
LARGE_DIR_SIZE = 100
TARGET_REGEX = re.compile(r"^(?P<path>.*):(?P<name>[^\\/@:]*)$")
def dos2unix(data):
@@ -374,3 +375,16 @@ def format_link(link):
return "<{blue}{link}{nc}>".format(
blue=colorama.Fore.CYAN, link=link, nc=colorama.Fore.RESET
)
def parse_target(target, default="Dvcfile"):
if not target:
return None, None
match = TARGET_REGEX.match(target)
if not match:
return target, None
path, name = match.group("path"), match.group("name")
if not path:
logger.warning("Assuming file to be '%s'", default)
return path or default, name
tests/func/test_dvcfile.py
+165
View File
@@ -0,0 +1,165 @@
import pytest
from dvc.dvcfile import Dvcfile
from dvc.exceptions import DvcException
from dvc.stage import Stage
from dvc.stage.exceptions import StageFileDoesNotExistError
def test_run_load_one_for_multistage(tmp_dir, dvc):
tmp_dir.gen("foo", "foo")
stage1 = dvc.run(
cmd="cp foo foo2",
deps=["foo"],
name="copy-foo-foo2",
outs_persist_no_cache=["foo2"],
always_changed=True,
)
stage2 = Dvcfile(dvc, "Dvcfile").load_one("copy-foo-foo2")
assert stage1 == stage2
foo_out = stage2.outs[0]
assert stage2.cmd == "cp foo foo2"
assert stage2.name == "copy-foo-foo2"
assert foo_out.def_path == "foo2"
assert foo_out.persist
assert not foo_out.use_cache
assert stage2.deps[0].def_path == "foo"
assert dvc.reproduce(":copy-foo-foo2")
def test_run_load_one_for_multistage_non_existing(tmp_dir, dvc):
with pytest.raises(StageFileDoesNotExistError):
Dvcfile(dvc, "Dvcfile").load_one("copy-foo-foo2")
def test_run_load_one_for_multistage_non_existing_stage_name(tmp_dir, dvc):
tmp_dir.gen("foo", "foo")
stage = dvc.run(
cmd="cp foo foo2",
deps=["foo"],
name="copy-foo-foo2",
metrics=["foo2"],
always_changed=True,
)
with pytest.raises(DvcException):
# TODO: Better exception
Dvcfile(dvc, stage.path).load_one("random-name")
def test_run_load_one_on_single_stage(tmp_dir, dvc):
tmp_dir.gen("foo", "foo")
stage = dvc.run(
cmd="cp foo foo2", deps=["foo"], metrics=["foo2"], always_changed=True,
)
Dvcfile(dvc, stage.path).load_one("random-name")
Dvcfile(dvc, stage.path).load_one()
def test_has_stage_with_name(tmp_dir, dvc):
tmp_dir.gen("foo", "foo")
dvc.run(
cmd="cp foo foo2",
deps=["foo"],
name="copy-foo-foo2",
metrics=["foo2"],
always_changed=True,
)
dvcfile = Dvcfile(dvc, "Dvcfile")
assert dvcfile.has_stage("copy-foo-foo2")
assert not dvcfile.has_stage("copy")
def test_load_all_multistage(tmp_dir, dvc):
tmp_dir.gen("foo", "foo")
stage1 = dvc.run(
cmd="cp foo foo2",
deps=["foo"],
name="copy-foo-foo2",
metrics=["foo2"],
always_changed=True,
)
stages = Dvcfile(dvc, "Dvcfile").load_all()
assert len(stages) == 1
assert stages[0] == stage1
tmp_dir.gen("bar", "bar")
stage2 = dvc.run(
cmd="cp bar bar2",
deps=["bar"],
name="copy-bar-bar2",
metrics=["bar2"],
always_changed=True,
)
assert set(Dvcfile(dvc, "Dvcfile").load_all()) == {stage2, stage1}
def test_load_all_singlestage(tmp_dir, dvc):
tmp_dir.gen("foo", "foo")
stage1 = dvc.run(
cmd="cp foo foo2", deps=["foo"], metrics=["foo2"], always_changed=True,
)
stages = Dvcfile(dvc, "foo2.dvc").load_all()
assert len(stages) == 1
assert stages == [stage1]
def test_load_singlestage(tmp_dir, dvc):
tmp_dir.gen("foo", "foo")
stage1 = dvc.run(
cmd="cp foo foo2", deps=["foo"], metrics=["foo2"], always_changed=True,
)
assert Dvcfile(dvc, "foo2.dvc").load() == stage1
def test_load_multistage(tmp_dir, dvc):
from dvc.dvcfile import MultiStageFileLoadError
tmp_dir.gen("foo", "foo")
dvc.run(
cmd="cp foo foo2",
deps=["foo"],
name="copy-foo-foo2",
metrics=["foo2"],
always_changed=True,
)
with pytest.raises(MultiStageFileLoadError):
Dvcfile(dvc, "Dvcfile").load()
def test_is_multistage(tmp_dir, dvc):
tmp_dir.gen({"foo": "foo", "bar": "bar"})
stage1 = dvc.run(
cmd="cp foo foo2",
deps=["foo"],
name="copy-foo-foo2",
metrics=["foo2"],
always_changed=True,
)
assert Dvcfile(dvc, stage1.path).is_multi_stage()
stage2 = dvc.run(
cmd="cp bar bar2", deps=["bar"], metrics=["bar2"], always_changed=True,
)
assert not Dvcfile(dvc, stage2.path).is_multi_stage()
def test_stage_collection(tmp_dir, dvc):
tmp_dir.gen(
{
"dir": {"file1": "file1", "file2": "file2"},
"foo": "foo",
"bar": "bar",
}
)
(stage1,) = dvc.add("dir")
stage2 = dvc.run(
cmd="cp foo foo2",
deps=["foo"],
name="copy-foo-foo2",
metrics=["foo2"],
always_changed=True,
)
stage3 = dvc.run(
cmd="cp bar bar2", deps=["bar"], metrics=["bar2"], always_changed=True,
)
assert {s for s in dvc.pipeline_stages} == {stage3, stage2, stage1}
assert {s for s in dvc.stages} == {Stage(dvc, "foo2.dvc"), stage1, stage3}
tests/func/test_pipeline.py
+44 -1
View File
@@ -1,7 +1,7 @@
import logging
from dvc.main import main
from dvc.command.pipeline import CmdPipelineShow
from dvc.command.pipeline import CmdPipelineShow, CmdPipelineList
from tests.basic_env import TestDvc
from tests.func.test_repro import TestRepro
from tests.func.test_repro import TestReproChangedDeepData
@@ -266,3 +266,46 @@ def test_split_pipeline(tmp_dir, scm, dvc):
("data_train", "data"),
("data_valid", "data"),
}
def test_pipeline_list_show_multistage(tmp_dir, dvc, run_copy, caplog):
tmp_dir.gen("foo", "foo")
run_copy("foo", "bar", name="copy-foo-bar")
run_copy("bar", "foobar")
command = CmdPipelineShow([])
caplog.clear()
with caplog.at_level(logging.INFO, "dvc"):
command._show("foobar.dvc", False, False, False)
output = caplog.text.splitlines()
assert "Dvcfile:copy-foo-bar" in output[0]
assert "foobar.dvc" in output[1]
caplog.clear()
with caplog.at_level(logging.INFO, "dvc"):
command._show("Dvcfile:copy-foo-bar", False, False, False)
assert "Dvcfile:copy-foo-bar" in caplog.text
assert "foobar.dvc" not in caplog.text
command = CmdPipelineList([])
caplog.clear()
with caplog.at_level(logging.INFO, "dvc"):
command.run()
assert "Dvcfile:copy-foo-bar" in caplog.text
assert "foobar.dvc" in caplog.text
assert "1 pipelines in total"
def test_pipeline_ascii_multistage(tmp_dir, dvc, run_copy):
tmp_dir.gen("foo", "foo")
run_copy("foo", "bar", name="copy-foo-bar")
run_copy("bar", "foobar")
command = CmdPipelineShow([])
nodes, edges, is_tree = command._build_graph("foobar.dvc")
assert set(nodes) == {"Dvcfile:copy-foo-bar", "foobar.dvc"}
assert set(edges) == {
("foobar.dvc", "Dvcfile:copy-foo-bar"),
}
nodes, edges, is_tree = command._build_graph("Dvcfile:copy-foo-bar")
assert set(nodes) == {"Dvcfile:copy-foo-bar"}
tests/func/test_repro.py
+137 -90
View File
@@ -50,7 +50,17 @@ from tests.remotes import (
from tests.utils.httpd import StaticFileServer, ContentMD5Handler
class TestRepro(TestDvc):
class SingleStageRun:
def _run(self, **kwargs):
kwargs.pop("name", None)
return self.dvc.run(**kwargs)
@staticmethod
def _get_stage_target(stage):
return stage.path
class TestRepro(SingleStageRun, TestDvc):
def setUp(self):
super().setUp()
@@ -61,11 +71,12 @@ class TestRepro(TestDvc):
self.file1 = "file1"
self.file1_stage = self.file1 + ".dvc"
self.dvc.run(
self.stage = self._run(
fname=self.file1_stage,
outs=[self.file1],
deps=[self.FOO, self.CODE],
cmd="python {} {} {}".format(self.CODE, self.FOO, self.file1),
name="run1",
)
@@ -73,18 +84,24 @@ class TestReproFail(TestRepro):
def test(self):
os.unlink(self.CODE)
ret = main(["repro", self.file1_stage])
ret = main(["repro", self._get_stage_target(self.stage)])
self.assertNotEqual(ret, 0)
class TestReproCyclicGraph(TestDvc):
class TestReproCyclicGraph(SingleStageRun, TestDvc):
def test(self):
self.dvc.run(
deps=[self.FOO], outs=["bar.txt"], cmd="echo bar > bar.txt"
self._run(
deps=[self.FOO],
outs=["bar.txt"],
cmd="echo bar > bar.txt",
name="copybarbar.txt",
)
self.dvc.run(
deps=["bar.txt"], outs=["baz.txt"], cmd="echo baz > baz.txt"
self._run(
deps=["bar.txt"],
outs=["baz.txt"],
cmd="echo baz > baz.txt",
name="copybazbaz.txt",
)
stage_dump = {
@@ -174,7 +191,7 @@ class TestReproWorkingDirectoryAsOutput(TestDvc):
# NOTE: os.walk() walks in a sorted order and we need dir2 subdirs to
# be processed before dir1 to load error.dvc first.
self.dvc.stages = [
self.dvc.pipeline_stages = [
nested_stage,
Dvcfile(self.dvc, error_stage_path).load(),
]
@@ -208,7 +225,10 @@ class TestReproWorkingDirectoryAsOutput(TestDvc):
self.fail("should not raise StagePathAsOutputError")
class TestReproDepUnderDir(TestDvc):
# TODO: Test ^ for multistage
class TestReproDepUnderDir(SingleStageRun, TestDvc):
def test(self):
stages = self.dvc.add(self.DATA_DIR)
self.assertEqual(len(stages), 1)
@@ -216,12 +236,12 @@ class TestReproDepUnderDir(TestDvc):
self.assertTrue(self.dir_stage is not None)
self.file1 = "file1"
self.file1_stage = self.file1 + ".dvc"
self.dvc.run(
fname=self.file1_stage,
stage = self._run(
fname=self.file1 + ".dvc",
outs=[self.file1],
deps=[self.DATA, self.CODE],
cmd="python {} {} {}".format(self.CODE, self.DATA, self.file1),
name="copy-data-file1",
)
self.assertTrue(filecmp.cmp(self.file1, self.DATA, shallow=False))
@@ -229,12 +249,12 @@ class TestReproDepUnderDir(TestDvc):
os.unlink(self.DATA)
shutil.copyfile(self.FOO, self.DATA)
stages = self.dvc.reproduce(self.file1_stage)
stages = self.dvc.reproduce(self._get_stage_target(stage))
self.assertEqual(len(stages), 2)
self.assertTrue(filecmp.cmp(self.file1, self.FOO, shallow=False))
class TestReproDepDirWithOutputsUnderIt(TestDvc):
class TestReproDepDirWithOutputsUnderIt(SingleStageRun, TestDvc):
def test(self):
stages = self.dvc.add(self.DATA)
self.assertEqual(len(stages), 1)
@@ -244,23 +264,26 @@ class TestReproDepDirWithOutputsUnderIt(TestDvc):
self.assertEqual(len(stages), 1)
self.assertTrue(stages[0] is not None)
stage = self.dvc.run(fname="Dvcfile", deps=[self.DATA, self.DATA_SUB])
stage = self.dvc.run(
fname="dvcfile2.dvc", deps=[self.DATA, self.DATA_SUB]
)
self.assertTrue(stage is not None)
file1 = "file1"
file1_stage = file1 + ".dvc"
stage = self.dvc.run(
stage = self._run(
fname=file1_stage,
deps=[self.DATA_DIR],
outs=[file1],
cmd="python {} {} {}".format(self.CODE, self.DATA, file1),
name="copy-data-file1",
)
self.assertTrue(stage is not None)
os.unlink(self.DATA)
shutil.copyfile(self.FOO, self.DATA)
stages = self.dvc.reproduce(file1_stage)
stages = self.dvc.reproduce(self._get_stage_target(stage))
self.assertEqual(len(stages), 2)
@@ -275,17 +298,22 @@ class TestReproNoDeps(TestRepro):
)
with open(code_file, "w+") as fd:
fd.write(code)
self.dvc.run(
fname=stage_file, outs=[out], cmd="python {}".format(code_file)
stage = self._run(
fname=stage_file,
outs=[out],
cmd="python {}".format(code_file),
name="uuid",
)
stages = self.dvc.reproduce(stage_file)
stages = self.dvc.reproduce(self._get_stage_target(stage))
self.assertEqual(len(stages), 1)
class TestReproForce(TestRepro):
def test(self):
stages = self.dvc.reproduce(self.file1_stage, force=True)
stages = self.dvc.reproduce(
self._get_stage_target(self.stage), force=True
)
self.assertEqual(len(stages), 2)
@@ -293,7 +321,7 @@ class TestReproChangedCode(TestRepro):
def test(self):
self.swap_code()
stages = self.dvc.reproduce(self.file1_stage)
stages = self.dvc.reproduce(self._get_stage_target(self.stage))
self.assertTrue(filecmp.cmp(self.file1, self.BAR, shallow=False))
self.assertEqual(len(stages), 1)
@@ -311,7 +339,7 @@ class TestReproChangedData(TestRepro):
def test(self):
self.swap_foo_with_bar()
stages = self.dvc.reproduce(self.file1_stage)
stages = self.dvc.reproduce(self._get_stage_target(self.stage))
self.assertTrue(filecmp.cmp(self.file1, self.BAR, shallow=False))
self.assertEqual(len(stages), 2)
@@ -325,19 +353,21 @@ class TestReproDry(TestReproChangedData):
def test(self):
self.swap_foo_with_bar()
stages = self.dvc.reproduce(self.file1_stage, dry=True)
stages = self.dvc.reproduce(
self._get_stage_target(self.stage), dry=True
)
self.assertTrue(len(stages), 2)
self.assertFalse(filecmp.cmp(self.file1, self.BAR, shallow=False))
ret = main(["repro", "--dry", self.file1_stage])
ret = main(["repro", "--dry", self._get_stage_target(self.stage)])
self.assertEqual(ret, 0)
self.assertFalse(filecmp.cmp(self.file1, self.BAR, shallow=False))
class TestReproUpToDate(TestRepro):
def test(self):
ret = main(["repro", self.file1_stage])
ret = main(["repro", self._get_stage_target(self.stage)])
self.assertEqual(ret, 0)
@@ -383,18 +413,18 @@ class TestReproChangedDeepData(TestReproChangedData):
super().setUp()
self.file2 = "file2"
self.file2_stage = self.file2 + ".dvc"
self.dvc.run(
fname=self.file2_stage,
self.stage = self._run(
fname=self.file2 + ".dvc",
outs=[self.file2],
deps=[self.file1, self.CODE],
cmd="python {} {} {}".format(self.CODE, self.file1, self.file2),
name="copy-file-file2",
)
def test(self):
self.swap_foo_with_bar()
stages = self.dvc.reproduce(self.file2_stage)
stages = self.dvc.reproduce(self._get_stage_target(self.stage))
self.assertTrue(filecmp.cmp(self.file1, self.BAR, shallow=False))
self.assertTrue(filecmp.cmp(self.file2, self.BAR, shallow=False))
@@ -450,16 +480,18 @@ class TestReproIgnoreBuildCache(TestDvc):
class TestReproPipeline(TestReproChangedDeepData):
def test(self):
stages = self.dvc.reproduce(
self.file1_stage, force=True, pipeline=True
self._get_stage_target(self.stage), force=True, pipeline=True
)
self.assertEqual(len(stages), 3)
def test_cli(self):
ret = main(["repro", "--pipeline", "-f", self.file1_stage])
ret = main(
["repro", "--pipeline", "-f", self._get_stage_target(self.stage)]
)
self.assertEqual(ret, 0)
class TestReproPipelines(TestDvc):
class TestReproPipelines(SingleStageRun, TestDvc):
def setUp(self):
super().setUp()
@@ -474,31 +506,27 @@ class TestReproPipelines(TestDvc):
self.assertTrue(self.bar_stage is not None)
self.file1 = "file1"
self.file1_stage = self.file1 + ".dvc"
self.dvc.run(
fname=self.file1_stage,
self.file1_stage = self.dvc.run(
fname=self.file1 + ".dvc",
outs=[self.file1],
deps=[self.FOO, self.CODE],
cmd="python {} {} {}".format(self.CODE, self.FOO, self.file1),
)
self.file2 = "file2"
self.file2_stage = self.file2 + ".dvc"
self.dvc.run(
fname=self.file2_stage,
self.file2_stage = self._run(
fname=self.file2 + ".dvc",
outs=[self.file2],
deps=[self.BAR, self.CODE],
cmd="python {} {} {}".format(self.CODE, self.BAR, self.file2),
name="copy-BAR-file2",
)
def test(self):
stages = self.dvc.reproduce(all_pipelines=True, force=True)
self.assertEqual(len(stages), 4)
names = [stage.relpath for stage in stages]
self.assertTrue(self.foo_stage.relpath in names)
self.assertTrue(self.bar_stage.relpath in names)
self.assertTrue(self.file1_stage in names)
self.assertTrue(self.file2_stage in names)
self.assertTrue(self.file1_stage in stages)
self.assertTrue(self.file2_stage in stages)
def test_cli(self):
ret = main(["repro", "-f", "-P"])
@@ -508,24 +536,24 @@ class TestReproPipelines(TestDvc):
class TestReproLocked(TestReproChangedData):
def test(self):
file2 = "file2"
file2_stage = file2 + ".dvc"
self.dvc.run(
fname=file2_stage,
file2_stage = self._run(
fname=file2 + ".dvc",
outs=[file2],
deps=[self.file1, self.CODE],
cmd="python {} {} {}".format(self.CODE, self.file1, file2),
name="copy-file1-file2",
)
self.swap_foo_with_bar()
ret = main(["lock", file2_stage])
ret = main(["lock", self._get_stage_target(file2_stage)])
self.assertEqual(ret, 0)
stages = self.dvc.reproduce(file2_stage)
stages = self.dvc.reproduce(self._get_stage_target(file2_stage))
self.assertEqual(len(stages), 0)
ret = main(["unlock", file2_stage])
ret = main(["unlock", self._get_stage_target(file2_stage)])
self.assertEqual(ret, 0)
stages = self.dvc.reproduce(file2_stage)
stages = self.dvc.reproduce(self._get_stage_target(file2_stage))
self.assertTrue(filecmp.cmp(self.file1, self.BAR, shallow=False))
self.assertTrue(filecmp.cmp(file2, self.BAR, shallow=False))
self.assertEqual(len(stages), 3)
@@ -538,29 +566,29 @@ class TestReproLocked(TestReproChangedData):
self.assertNotEqual(ret, 0)
class TestReproLockedCallback(TestDvc):
class TestReproLockedCallback(SingleStageRun, TestDvc):
def test(self):
file1 = "file1"
file1_stage = file1 + ".dvc"
# NOTE: purposefully not specifying dependencies
# to create a callback stage.
stage = self.dvc.run(
stage = self._run(
fname=file1_stage,
outs=[file1],
cmd="python {} {} {}".format(self.CODE, self.FOO, file1),
name="copy-FOO-file1",
)
self.assertTrue(stage is not None)
self.assertEqual(stage.relpath, file1_stage)
stages = self.dvc.reproduce(file1_stage)
stages = self.dvc.reproduce(self._get_stage_target(stage))
self.assertEqual(len(stages), 1)
self.dvc.lock_stage(file1_stage)
stages = self.dvc.reproduce(file1_stage)
self.dvc.lock_stage(self._get_stage_target(stage))
stages = self.dvc.reproduce(self._get_stage_target(stage))
self.assertEqual(len(stages), 0)
self.dvc.lock_stage(file1_stage, unlock=True)
stages = self.dvc.reproduce(file1_stage)
self.dvc.lock_stage(self._get_stage_target(stage), unlock=True)
stages = self.dvc.reproduce(self._get_stage_target(stage))
self.assertEqual(len(stages), 1)
@@ -569,12 +597,13 @@ class TestReproLockedUnchanged(TestRepro):
"""
Check that locking/unlocking doesn't affect stage state
"""
self.dvc.lock_stage(self.file1_stage)
stages = self.dvc.reproduce(self.file1_stage)
target = self._get_stage_target(self.stage)
self.dvc.lock_stage(target)
stages = self.dvc.reproduce(target)
self.assertEqual(len(stages), 0)
self.dvc.lock_stage(self.file1_stage, unlock=True)
stages = self.dvc.reproduce(self.file1_stage)
self.dvc.lock_stage(target, unlock=True)
stages = self.dvc.reproduce(target)
self.assertEqual(len(stages), 0)
@@ -610,11 +639,13 @@ class TestReproMetricsAddUnchanged(TestDvc):
class TestReproPhony(TestReproChangedData):
def test(self):
stage = self.dvc.run(deps=[self.file1])
stage = self._run(
cmd="cat " + self.file1, deps=[self.file1], name="no_cmd?"
)
self.swap_foo_with_bar()
self.dvc.reproduce(stage.path)
self.dvc.reproduce(self._get_stage_target(stage))
self.assertTrue(filecmp.cmp(self.file1, self.BAR, shallow=False))
@@ -624,7 +655,7 @@ class TestNonExistingOutput(TestRepro):
os.unlink(self.FOO)
with self.assertRaises(ReproductionError):
self.dvc.reproduce(self.file1_stage)
self.dvc.reproduce(self._get_stage_target(self.stage))
class TestReproDataSource(TestReproChangedData):
@@ -637,12 +668,11 @@ class TestReproDataSource(TestReproChangedData):
self.assertEqual(stages[0].outs[0].checksum, file_md5(self.BAR)[0])
class TestReproChangedDir(TestDvc):
class TestReproChangedDir(SingleStageRun, TestDvc):
def test(self):
file_name = "file"
shutil.copyfile(self.FOO, file_name)
stage_name = "dir.dvc"
dir_name = "dir"
dir_code = "dir.py"
code = (
@@ -653,24 +683,25 @@ class TestReproChangedDir(TestDvc):
with open(dir_code, "w+") as fd:
fd.write(code.format(dir_name, file_name, dir_name, file_name))
self.dvc.run(
fname=stage_name,
stage = self._run(
outs=[dir_name],
deps=[file_name, dir_code],
cmd="python {}".format(dir_code),
name="copy-in-dir",
)
target = self._get_stage_target(stage)
stages = self.dvc.reproduce(stage_name)
stages = self.dvc.reproduce(target)
self.assertEqual(len(stages), 0)
os.unlink(file_name)
shutil.copyfile(self.BAR, file_name)
stages = self.dvc.reproduce(stage_name)
stages = self.dvc.reproduce(target)
self.assertEqual(len(stages), 1)
class TestReproChangedDirData(TestDvc):
class TestReproChangedDirData(SingleStageRun, TestDvc):
def test(self):
dir_name = "dir"
dir_code = "dir_code.py"
@@ -681,32 +712,35 @@ class TestReproChangedDirData(TestDvc):
"shutil.copytree(sys.argv[1], sys.argv[2])"
)
stage = self.dvc.run(
stage = self._run(
outs=[dir_name],
deps=[self.DATA_DIR, dir_code],
cmd="python {} {} {}".format(dir_code, self.DATA_DIR, dir_name),
name="copy-dir",
)
target = self._get_stage_target(stage)
self.assertTrue(stage is not None)
stages = self.dvc.reproduce(stage.path)
stages = self.dvc.reproduce(target)
self.assertEqual(len(stages), 0)
with open(self.DATA_SUB, "a") as fd:
fd.write("add")
stages = self.dvc.reproduce(stage.path)
stages = self.dvc.reproduce(target)
self.assertEqual(len(stages), 1)
self.assertTrue(stages[0] is not None)
# Check that dvc indeed registers changed output dir
shutil.move(self.BAR, dir_name)
stages = self.dvc.reproduce(stage.path)
stages = self.dvc.reproduce(target)
self.assertEqual(len(stages), 1)
self.assertTrue(stages[0] is not None)
# Check that dvc registers mtime change for the directory.
System.hardlink(self.DATA_SUB, self.DATA_SUB + ".lnk")
stages = self.dvc.reproduce(stage.path)
stages = self.dvc.reproduce(target)
self.assertEqual(len(stages), 1)
self.assertTrue(stages[0] is not None)
@@ -729,7 +763,7 @@ class TestCmdRepro(TestReproChangedData):
ret = main(["status"])
self.assertEqual(ret, 0)
ret = main(["repro", self.file1_stage])
ret = main(["repro", self._get_stage_target(self.stage)])
self.assertEqual(ret, 0)
ret = main(["repro", "non-existing-file"])
@@ -1206,33 +1240,41 @@ class TestReproShell(TestDvc):
self.assertEqual(os.getenv("SHELL"), fd.read().strip())
class TestReproAllPipelines(TestDvc):
class TestReproAllPipelines(SingleStageRun, TestDvc):
def test(self):
stages = [
self.dvc.run(
self._run(
fname="start.dvc",
outs=["start.txt"],
cmd="echo start > start.txt",
name="start",
),
self.dvc.run(
self._run(
fname="middle.dvc",
deps=["start.txt"],
outs=["middle.txt"],
cmd="echo middle > middle.txt",
name="middle",
),
self.dvc.run(
self._run(
fname="final.dvc",
deps=["middle.txt"],
outs=["final.txt"],
cmd="echo final > final.txt",
name="final",
),
self.dvc.run(
self._run(
fname="disconnected.dvc",
outs=["disconnected.txt"],
cmd="echo other > disconnected.txt",
name="disconnected",
),
]
from dvc.state import StateNoop
self.dvc.state = StateNoop()
with patch.object(
Stage, "reproduce", side_effect=stages
) as mock_reproduce:
@@ -1244,21 +1286,26 @@ class TestReproAllPipelines(TestDvc):
class TestReproNoCommit(TestRepro):
def test(self):
remove(self.dvc.cache.local.cache_dir)
ret = main(["repro", self.file1_stage, "--no-commit"])
ret = main(
["repro", self._get_stage_target(self.stage), "--no-commit"]
)
self.assertEqual(ret, 0)
self.assertFalse(os.path.exists(self.dvc.cache.local.cache_dir))
class TestReproAlreadyCached(TestRepro):
def test(self):
run_out = self.dvc.run(
stage = self._run(
fname="datetime.dvc",
deps=[],
outs=["datetime.txt"],
cmd='python -c "import time; print(time.time())" > datetime.txt',
).outs[0]
repro_out = self.dvc.reproduce(target="datetime.dvc")[0].outs[0]
name="datetime",
)
run_out = stage.outs[0]
repro_out = self.dvc.reproduce(self._get_stage_target(stage))[0].outs[
0
]
self.assertNotEqual(run_out.checksum, repro_out.checksum)
tests/func/test_repro_multistage.py
+412
View File
@@ -0,0 +1,412 @@
import os
from textwrap import dedent
import pytest
from dvc.stage import PipelineStage
from tests.func import test_repro
from dvc.main import main
COPY_SCRIPT_FORMAT = dedent(
"""\
import sys
import shutil
shutil.copyfile({}, {})
"""
)
COPY_SCRIPT = COPY_SCRIPT_FORMAT.format("sys.argv[1]", "sys.argv[2]")
class MultiStageRun:
def _run(self, **kwargs):
assert kwargs.get("name")
kwargs.pop("fname", None)
# ignore fname for now
return self.dvc.run(**kwargs)
@staticmethod
def _get_stage_target(stage):
return stage.path + ":" + stage.name
class TestReproFailMultiStage(MultiStageRun, test_repro.TestReproFail):
pass
class TestReproCyclicGraphMultiStage(
MultiStageRun, test_repro.TestReproCyclicGraph
):
# TODO: Also test with new-style forced dump
pass
class TestReproUnderDirMultiStage(
MultiStageRun, test_repro.TestReproDepUnderDir
):
pass
class TestReproDepDirWithOutputsUnderItMultiStage(
MultiStageRun, test_repro.TestReproDepDirWithOutputsUnderIt
):
pass
class TestReproNoDepsMultiStage(MultiStageRun, test_repro.TestReproNoDeps):
pass
class TestReproForceMultiStage(MultiStageRun, test_repro.TestReproForce):
pass
class TestReproChangedCodeMultiStage(
MultiStageRun, test_repro.TestReproChangedCode
):
pass
class TestReproChangedDataMultiStage(
MultiStageRun, test_repro.TestReproChangedData
):
pass
class TestReproDry(MultiStageRun, test_repro.TestReproDry):
pass
class TestReproUpToDateMultiStage(MultiStageRun, test_repro.TestReproUpToDate):
pass
class TestReproChangedDeepDataMultiStage(
MultiStageRun, test_repro.TestReproChangedDeepData
):
pass
class TestReproPipelineMultiStage(MultiStageRun, test_repro.TestReproPipeline):
pass
class TestReproPipelinesMultiStage(
MultiStageRun, test_repro.TestReproPipelines
):
pass
class TestReproLockedMultiStage(MultiStageRun, test_repro.TestReproLocked):
pass
class TestReproLockedCallbackMultiStage(
MultiStageRun, test_repro.TestReproLockedCallback
):
pass
class TestReproLockedUnchangedMultiStage(
MultiStageRun, test_repro.TestReproLockedUnchanged
):
pass
class TestReproPhonyMultiStage(MultiStageRun, test_repro.TestReproPhony):
pass
class TestCmdReproMultiStage(MultiStageRun, test_repro.TestCmdRepro):
pass
class TestReproAllPipelinesMultiStage(
MultiStageRun, test_repro.TestReproAllPipelines
):
pass
class TestReproNoCommit(MultiStageRun, test_repro.TestReproNoCommit):
pass
class TestNonExistingOutputMultiStage(
MultiStageRun, test_repro.TestNonExistingOutput
):
pass
class TestReproAlreadyCachedMultiStage(
MultiStageRun, test_repro.TestReproAlreadyCached
):
pass
class TestReproChangedDirMultiStage(
MultiStageRun, test_repro.TestReproChangedDir
):
pass
class TestReproChangedDirDataMultiStage(
MultiStageRun, test_repro.TestReproChangedDirData
):
pass
def test_non_existing_stage_name(tmp_dir, dvc, run_copy):
from dvc.exceptions import DvcException
tmp_dir.gen("file1", "file1")
run_copy("file1", "file2", name="copy-file1-file2")
with pytest.raises(DvcException):
dvc.lock_stage(":copy-file1-file3")
assert main(["lock", ":copy-file1-file3"]) != 0
# TODO: TestReproWorkingDirectoryAsOutput
def test_downstream(tmp_dir, dvc):
# The dependency graph should look like this:
#
# E
# / \
# D F
# / \ \
# B C G
# \ /
# A
#
assert main(["run", "-n", "A-gen", "-o", "A", "echo A>A"]) == 0
assert main(["run", "-n", "B-gen", "-d", "A", "-o", "B", "echo B>B"]) == 0
assert main(["run", "-d", "A", "-o", "C", "echo C>C"]) == 0
assert (
main(
["run", "-n", "D-gen", "-d", "B", "-d", "C", "-o", "D", "echo D>D"]
)
== 0
)
assert main(["run", "-o", "G", "echo G>G"]) == 0
assert main(["run", "-n", "F-gen", "-d", "G", "-o", "F", "echo F>F"]) == 0
assert main(["run", "-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("Dvcfile:B-gen", downstream=True, force=True)
assert len(evaluation) == 3
assert (
isinstance(evaluation[0], PipelineStage)
and evaluation[0].relpath == "Dvcfile"
and evaluation[0].name == "B-gen"
)
assert (
isinstance(evaluation[1], PipelineStage)
and evaluation[1].relpath == "Dvcfile"
and evaluation[1].name == "D-gen"
)
assert (
not isinstance(evaluation[2], PipelineStage)
and evaluation[2].relpath == "E.dvc"
)
def test_repro_when_cmd_changes(tmp_dir, dvc, run_copy):
from dvc.dvcfile import Dvcfile
tmp_dir.gen("foo", "foo")
stage = run_copy("foo", "bar", fname="copy-process.dvc", name="copy-file")
target = "copy-process.dvc:copy-file"
assert not dvc.reproduce(target)
stage.cmd = " ".join(stage.cmd.split()) # change cmd spacing by two
Dvcfile(dvc, "copy-process.dvc").dump_multistage_dvcfile(stage)
assert dvc.reproduce(target)[0] == stage
def test_repro_when_new_deps_is_added_in_dvcfile(tmp_dir, dvc, run_copy):
from dvc.dvcfile import Dvcfile
from dvc.utils.stage import dump_stage_file
tmp_dir.gen("copy.py", COPY_SCRIPT)
tmp_dir.gen({"foo": "foo", "bar": "bar"})
stage = dvc.run(
cmd="python copy.py {} {}".format("foo", "foobar"),
outs=["foobar"],
deps=["foo"],
fname="copy-process.dvc",
name="copy-file",
)
target = "copy-process.dvc:copy-file"
assert not dvc.reproduce(target)
dvcfile = Dvcfile(dvc, stage.path)
data, _ = dvcfile._load()
data["stages"]["copy-file"]["deps"] += ["copy.py"]
dump_stage_file(stage.path, data)
assert dvc.reproduce(target)[0] == stage
def test_repro_when_new_outs_is_added_in_dvcfile(tmp_dir, dvc):
from dvc.dvcfile import Dvcfile
from dvc.utils.stage import dump_stage_file
tmp_dir.gen("copy.py", COPY_SCRIPT)
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"],
fname="copy-process.dvc",
name="copy-file",
)
target = "copy-process.dvc:copy-file"
assert not dvc.reproduce(target)
dvcfile = Dvcfile(dvc, stage.path)
data, _ = dvcfile._load()
data["stages"]["copy-file"]["outs"] = ["foobar"]
dump_stage_file(stage.path, data)
assert dvc.reproduce(target)[0] == stage
def test_repro_when_new_deps_is_moved(tmp_dir, dvc):
from dvc.dvcfile import Dvcfile
from dvc.utils.stage import dump_stage_file
tmp_dir.gen("copy.py", COPY_SCRIPT)
tmp_dir.gen({"foo": "foo", "bar": "foo"})
stage = dvc.run(
cmd="python copy.py {} {}".format("foo", "foobar"),
outs=["foobar"],
deps=["foo"],
fname="copy-process.dvc",
name="copy-file",
)
target = "copy-process.dvc:copy-file"
assert not dvc.reproduce(target)
tmp_dir.gen("copy.py", COPY_SCRIPT_FORMAT.format("'bar'", "'foobar'"))
from shutil import move
move("foo", "bar")
dvcfile = Dvcfile(dvc, stage.path)
data, _ = dvcfile._load()
data["stages"]["copy-file"]["deps"] = ["bar"]
dump_stage_file(stage.path, data)
assert dvc.reproduce(target)[0] == stage
def test_repro_when_new_out_overlaps_others_stage_outs(tmp_dir, dvc):
from dvc.utils.stage import dump_stage_file
from dvc.exceptions import OverlappingOutputPathsError
tmp_dir.gen({"dir": {"file1": "file1"}, "foo": "foo"})
dvc.add("dir")
dump_stage_file(
"Dvcfile",
{
"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):
from dvc.utils.stage import dump_stage_file
from dvc.exceptions import ReproductionError
tmp_dir.gen("copy.py", COPY_SCRIPT)
tmp_dir.gen("foo", "foo")
dump_stage_file(
"Dvcfile",
{
"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):
from dvc.utils.stage import dump_stage_file
from dvc.exceptions import ReproductionError
tmp_dir.gen("copy.py", COPY_SCRIPT)
tmp_dir.gen("foo", "foo")
dump_stage_file(
"Dvcfile",
{
"stages": {
"run-copy": {
"cmd": "python copy {} {}".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):
from dvc.utils.stage import dump_stage_file
tmp_dir.gen("copy.py", COPY_SCRIPT)
tmp_dir.gen("foo", "foo")
dump_stage_file(
"Dvcfile",
{
"stages": {
"run-copy": {
"cmd": "python copy.py {} {}".format("foo", "foobar"),
"deps": ["foo"],
"outs": ["foobar"],
}
}
},
)
assert dvc.reproduce(":run-copy")
assert os.path.exists("Dvcfile.lock")
assert os.path.exists("foobar.dvc")
assert not dvc.reproduce(":run-copy")
os.unlink("Dvcfile.lock")
stages = dvc.reproduce(":run-copy")
assert (
stages
and stages[0].relpath == "Dvcfile"
and stages[0].name == "run-copy"
)
assert os.path.exists("foobar.dvc")
tests/func/test_run_multistage.py
+180
View File
@@ -0,0 +1,180 @@
import pytest
import os
def test_run_with_name(tmp_dir, dvc, run_copy):
from dvc.stage import PipelineStage
from dvc.dvcfile import DVC_FILE, DVC_FILE_SUFFIX
tmp_dir.dvc_gen("foo", "foo")
assert not os.path.exists(DVC_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(DVC_FILE)
assert os.path.exists(DVC_FILE + ".lock")
assert os.path.exists("foo" + DVC_FILE_SUFFIX)
def test_run_with_multistage_and_single_stage(tmp_dir, dvc, run_copy):
from dvc.stage import PipelineStage, Stage
tmp_dir.dvc_gen("foo", "foo")
stage1 = run_copy("foo", "foo1")
stage2 = run_copy("foo1", "foo2", name="copy-foo1-foo2")
stage3 = run_copy("foo2", "foo3")
assert isinstance(stage2, PipelineStage)
assert isinstance(stage1, Stage)
assert isinstance(stage3, Stage)
assert stage2.name == "copy-foo1-foo2"
def test_run_multi_stage_repeat(tmp_dir, dvc, run_copy):
from dvc.stage import PipelineStage
from dvc.dvcfile import Dvcfile, DVC_FILE, DVC_FILE_SUFFIX
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")
stages = Dvcfile(dvc, DVC_FILE).load_multi()
assert len(stages) == 2
assert all(isinstance(stage, PipelineStage) for stage in stages)
assert set(stage.name for stage in stages) == {
"copy-foo-foo1",
"copy-foo1-foo2",
}
assert all(
os.path.exists(file + DVC_FILE_SUFFIX)
for file in ["foo1", "foo2", "foo3"]
)
def test_multi_stage_try_writing_on_single_stage_file(tmp_dir, dvc, run_copy):
from dvc.exceptions import DvcException
from dvc.dvcfile import MultiStageFileLoadError
tmp_dir.dvc_gen("foo")
dvc.run(cmd="echo foo", deps=["foo"])
with pytest.raises(DvcException):
run_copy("foo", "foo2", name="copy-foo1-foo2")
run_copy("foo", "foo2", name="copy-foo1-foo2", fname="DIFFERENT-FILE.dvc")
with pytest.raises(MultiStageFileLoadError):
run_copy("foo2", "foo3", fname="DIFFERENT-FILE.dvc")
def test_multi_stage_run_cached(tmp_dir, dvc, run_copy):
tmp_dir.dvc_gen("foo", "foo")
run_copy("foo", "foo2", name="copy-foo1-foo2")
stage2 = run_copy("foo", "foo2", name="copy-foo1-foo2")
assert stage2 is None
def test_multistage_dump_on_non_cached_outputs(tmp_dir, dvc):
from dvc.dvcfile import DVC_FILE_SUFFIX
tmp_dir.dvc_gen("foo")
dvc.run(
cmd="cp foo foo1",
deps=["foo"],
name="copy-foo1-foo2",
outs_no_cache=["foo1"],
)
assert not os.path.exists("foo1" + DVC_FILE_SUFFIX)
def test_multistage_with_wdir(tmp_dir, dvc):
from dvc.dvcfile import Dvcfile
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, _ = Dvcfile(dvc, stage.path)._load()
assert "dir" == data["stages"]["copy-foo1-foo2"]["wdir"]
def test_multistage_always_changed(tmp_dir, dvc):
from dvc.dvcfile import Dvcfile
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, _ = Dvcfile(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):
from dvc.dvcfile import Dvcfile
tmp_dir.gen({"dir": {"foo": "foo", "bar": "bar"}})
dvc.run(
cmd="cp foo foo1",
deps=["foo"],
name="copy-foo-foo1",
outs=["foo1"],
wdir="dir",
)
data, _ = Dvcfile(dvc, "Dvcfile")._load()
assert data == {
"stages": {
"copy-foo-foo1": {
"cmd": "cp foo foo1",
"wdir": "dir",
"deps": ["foo"],
"outs": ["foo1"],
}
}
}
dvc.run(
cmd="cp foo foo2",
deps=["foo"],
name="copy-foo-foo2",
wdir="dir",
outs_persist=["foo2"],
always_changed=True,
)
assert Dvcfile(dvc, "Dvcfile")._load()[0] == {
"stages": {
"copy-foo-foo2": {
"cmd": "cp foo foo2",
"deps": ["foo"],
"outs_persist": ["foo2"],
"always_changed": True,
"wdir": "dir",
},
**data["stages"],
}
}
tests/func/test_run.py → tests/func/test_run_single_stage.py
View File
tests/unit/command/test_run.py
+5
View File
@@ -6,6 +6,8 @@ def test_run(mocker, dvc):
cli_args = parse_args(
[
"run",
"--name",
"nam",
"--deps",
"deps",
"--outs",
@@ -61,6 +63,7 @@ def test_run(mocker, dvc):
no_commit=True,
always_changed=True,
cmd="command",
name="nam",
)
@@ -86,6 +89,7 @@ def test_run_args_from_cli(mocker, dvc):
no_commit=False,
always_changed=False,
cmd="echo foo",
name=None,
)
@@ -111,4 +115,5 @@ def test_run_args_with_spaces(mocker, dvc):
no_commit=False,
always_changed=False,
cmd='echo "foo bar"',
name=None,
)
tests/unit/test_lockfile.py
+78
View File
@@ -0,0 +1,78 @@
from dvc.stage import PipelineStage
from dvc import lockfile
import json
import pytest
def test_stage_dump_no_outs_deps(tmp_dir, dvc):
stage = PipelineStage(name="s1", repo=dvc, path="path", cmd="command")
lockfile.dump(dvc, "path.lock", stage)
assert lockfile.load(dvc, "path.lock") == {
"s1": {"cmd": "command", "deps": {}, "outs": {}}
}
def test_stage_dump_when_already_exists(tmp_dir, dvc):
data = {"s1": {"cmd": "command", "deps": {}, "outs": {}}}
with open("path.lock", "w+") as f:
json.dump(data, f)
stage = PipelineStage(name="s2", repo=dvc, path="path", cmd="command2")
lockfile.dump(dvc, "path.lock", stage)
assert lockfile.load(dvc, "path.lock") == {
**data,
"s2": {"cmd": "command2", "deps": {}, "outs": {}},
}
def test_stage_dump_with_deps_and_outs(tmp_dir, dvc):
data = {
"s1": {
"cmd": "command",
"deps": {"1.txt": "checksum"},
"outs": {"2.txt": "checksum"},
}
}
with open("path.lock", "w+") as f:
json.dump(data, f)
stage = PipelineStage(name="s2", repo=dvc, path="path", cmd="command2")
lockfile.dump(dvc, "path.lock", stage)
assert lockfile.load(dvc, "path.lock") == {
**data,
"s2": {"cmd": "command2", "deps": {}, "outs": {}},
}
def test_stage_overwrites_if_already_exists(tmp_dir, dvc):
stage = PipelineStage(name="s2", repo=dvc, path="path", cmd="command2")
lockfile.dump(dvc, "path.lock", stage)
stage = PipelineStage(name="s2", repo=dvc, path="path", cmd="command3")
lockfile.dump(dvc, "path.lock", stage)
assert lockfile.load(dvc, "path.lock") == {
"s2": {"cmd": "command3", "deps": {}, "outs": {}},
}
def test_load_when_lockfile_does_not_exist(tmp_dir, dvc):
assert {} == lockfile.load(dvc, "dvcfile.lock")
@pytest.mark.parametrize(
"corrupt_data",
[
{"s1": {"cmd": "command", "outs": {}}},
{"s1": {"outs": {}}},
{"s1": {"cmd": "command", "deps": {}}},
{"s1": {}},
{"s1": {"cmd": "command", "outs": {"file": "checksum"}}},
{"s1": {"cmd": "command", "deps": {"file": "checksum"}}},
],
)
def test_load_when_lockfile_is_corrupted(tmp_dir, dvc, corrupt_data):
with open("Dvcfile.lock", "w+") as f:
json.dump(corrupt_data, f)
with pytest.raises(lockfile.LockfileCorruptedError) as exc_info:
lockfile.load(dvc, "Dvcfile.lock")
assert "Dvcfile.lock" in str(exc_info.value)