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

test: metric/cache/persist keys (#3785)

Browse Source
This commit is contained in:
Saugat Pachhai
2020-05-15 13:58:50 +03:00
committed by GitHub
parent cf7f125c14
commit 9897f7c470
7 changed files with 158 additions and 87 deletions
Download Patch File Download Diff File Expand all files Collapse all files
dvc/output/__init__.py
+5 -12
View File
@@ -110,16 +110,9 @@ def loadd_from(stage, d_list):
def loads_from(stage, s_list, use_cache=True, metric=False, persist=False):
ret = []
for s in s_list:
ret.append(
_get(
stage,
s,
info={},
cache=use_cache,
metric=metric,
persist=persist,
)
return [
_get(
stage, s, info={}, cache=use_cache, metric=metric, persist=persist,
)
return ret
for s in s_list
]
dvc/schema.py
+17 -4
View File
@@ -1,8 +1,8 @@
from voluptuous import Any, Optional, Required, Schema
from dvc import dependency, output
from dvc.output import CHECKSUMS_SCHEMA
from dvc.stage.params import OutputParams, StageParams
from dvc.output import CHECKSUMS_SCHEMA, BaseOutput
from dvc.stage.params import StageParams
STAGES = "stages"
SINGLE_STAGE_SCHEMA = {
@@ -25,16 +25,29 @@ LOCK_FILE_STAGE_SCHEMA = {
}
LOCKFILE_SCHEMA = {str: LOCK_FILE_STAGE_SCHEMA}
OUT_PSTAGE_DETAILED_SCHEMA = {
str: {
BaseOutput.PARAM_CACHE: bool,
BaseOutput.PARAM_PERSIST: bool,
BaseOutput.PARAM_METRIC: bool,
}
}
PARAM_PSTAGE_NON_DEFAULT_SCHEMA = {str: [str]}
SINGLE_PIPELINE_STAGE_SCHEMA = {
str: {
StageParams.PARAM_CMD: str,
Optional(StageParams.PARAM_WDIR): str,
Optional(StageParams.PARAM_DEPS): [str],
Optional(StageParams.PARAM_PARAMS): [Any(str, {str: [str]})],
Optional(StageParams.PARAM_PARAMS): [
Any(str, PARAM_PSTAGE_NON_DEFAULT_SCHEMA)
],
Optional(StageParams.PARAM_LOCKED): bool,
Optional(StageParams.PARAM_META): object,
Optional(StageParams.PARAM_ALWAYS_CHANGED): bool,
**{Optional(p.value): [str] for p in OutputParams},
Optional(StageParams.PARAM_OUTS): [
Any(str, OUT_PSTAGE_DETAILED_SCHEMA)
],
}
}
MULTI_STAGE_SCHEMA = {STAGES: SINGLE_PIPELINE_STAGE_SCHEMA}
dvc/serialize.py
+31 -17
View File
@@ -6,6 +6,8 @@ from typing import TYPE_CHECKING, List
from funcy import lsplit, rpartial
from dvc.dependency import ParamsDependency
from dvc.output import BaseOutput
from dvc.stage.params import StageParams
from dvc.stage.utils import resolve_wdir
from dvc.utils.collections import apply_diff
from dvc.utils.stage import parse_stage_for_update
@@ -13,26 +15,35 @@ from dvc.utils.stage import parse_stage_for_update
if TYPE_CHECKING:
from dvc.stage import PipelineStage, Stage
PARAM_PATH = ParamsDependency.PARAM_PATH
PARAM_PARAMS = ParamsDependency.PARAM_PARAMS
PARAM_PATH = ParamsDependency.PARAM_PATH
PARAM_DEPS = StageParams.PARAM_DEPS
PARAM_OUTS = StageParams.PARAM_OUTS
PARAM_CACHE = BaseOutput.PARAM_CACHE
PARAM_METRIC = BaseOutput.PARAM_METRIC
PARAM_PERSIST = BaseOutput.PARAM_PERSIST
DEFAULT_PARAMS_FILE = ParamsDependency.DEFAULT_PARAMS_FILE
sort_by_path = partial(sorted, key=attrgetter("def_path"))
def _get_outs(stage: "PipelineStage"):
outs_bucket = {}
for o in sort_by_path(stage.outs):
bucket_key = ["metrics"] if o.metric else ["outs"]
def _get_out(out):
res = OrderedDict()
if not out.use_cache:
res[PARAM_CACHE] = False
if out.metric:
res[PARAM_METRIC] = True
elif out.persist:
res[PARAM_PERSIST] = True
return out.def_path if not res else {out.def_path: res}
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 [(key, outs_bucket[key]) for key in sorted(outs_bucket.keys())]
def _get_outs(outs):
return [_get_out(out) for out in sort_by_path(outs)]
def get_params_deps(stage: "PipelineStage"):
@@ -81,7 +92,7 @@ def to_pipeline_file(stage: "PipelineStage"):
(stage.PARAM_WDIR, resolve_wdir(stage.wdir, stage.path)),
(stage.PARAM_DEPS, sorted([d.def_path for d in deps])),
(stage.PARAM_PARAMS, serialized_params),
*_get_outs(stage),
(PARAM_OUTS, _get_outs(stage.outs)),
(stage.PARAM_LOCKED, stage.locked),
(stage.PARAM_ALWAYS_CHANGED, stage.always_changed),
]
@@ -98,18 +109,21 @@ def to_single_stage_lockfile(stage: "Stage") -> dict:
deps, outs = [
[
OrderedDict(
[("path", item.def_path), (item.checksum_type, item.checksum)]
[
(PARAM_PATH, item.def_path),
(item.checksum_type, item.checksum),
]
)
for item in sort_by_path(items)
]
for items in [deps, stage.outs]
]
if deps:
res["deps"] = deps
res[PARAM_DEPS] = deps
if params:
_, res["params"] = _serialize_params(params)
_, res[PARAM_PARAMS] = _serialize_params(params)
if outs:
res["outs"] = outs
res[PARAM_OUTS] = outs
return res
dvc/stage/loader.py
+32 -6
View File
@@ -11,7 +11,6 @@ from dvc import dependency, output
from ..dependency import ParamsDependency
from .exceptions import StageNameUnspecified, StageNotFound
from .utils import fill_stage_dependencies, fill_stage_outputs
logger = logging.getLogger(__name__)
@@ -112,11 +111,35 @@ class StageLoader(Mapping):
path = first(key)
res[path].extend(key[path])
stage.deps += dependency.loadd_from(
stage,
[{"path": key, "params": params} for key, params in res.items()],
stage.deps.extend(
dependency.loadd_from(
stage,
[
{"path": key, "params": params}
for key, params in res.items()
],
)
)
@classmethod
def _load_outs(cls, stage, data):
d = []
for key in data:
extra_kwargs = {}
if isinstance(key, str):
path = key
elif isinstance(key, dict):
path = first(key)
extra_kwargs = key[path]
else:
continue
d.append({"path": path, **extra_kwargs})
stage.outs.extend(output.loadd_from(stage, d))
@classmethod
def _load_deps(cls, stage, data):
stage.deps.extend(dependency.loads_from(stage, data))
@classmethod
def load_stage(cls, dvcfile, name, stage_data, lock_data):
from . import PipelineStage, Stage, loads_from
@@ -126,9 +149,12 @@ class StageLoader(Mapping):
)
stage = loads_from(PipelineStage, dvcfile.repo, path, wdir, stage_data)
stage.name = name
fill_stage_dependencies(stage, deps=stage_data.get("deps", []))
fill_stage_outputs(stage, **stage_data)
stage.deps, stage.outs = [], []
cls._load_outs(stage, stage_data.get("outs", []))
cls._load_deps(stage, stage_data.get("deps", []))
cls._load_params(stage, stage_data.get("params", []))
if lock_data:
stage.cmd_changed = lock_data.get(
Stage.PARAM_CMD
tests/dir_helpers.py
+35
View File
@@ -46,6 +46,7 @@ from global repo template to creating everything inplace, which:
import os
import pathlib
from contextlib import contextmanager
from textwrap import dedent
import pytest
from funcy import lmap, retry
@@ -59,6 +60,7 @@ __all__ = [
"scm",
"dvc",
"run_copy",
"run_head",
"erepo_dir",
"git_dir",
"setup_remote",
@@ -270,6 +272,39 @@ def run_copy(tmp_dir, dvc):
return run_copy
@pytest.fixture
def run_head(tmp_dir, dvc):
"""Output first line of each file to different file with '-1' appended.
Useful for tracking multiple outputs/dependencies which are not a copy
of each others.
"""
tmp_dir.gen(
{
"head.py": dedent(
"""
import sys
for file in sys.argv[1:]:
with open(file) as f, open(file +"-1","w+") as w:
w.write(f.readline())
"""
)
}
)
script = os.path.abspath(tmp_dir / "head.py")
def run(*args, **run_kwargs):
return dvc.run(
**{
"cmd": "python {} {}".format(script, " ".join(args)),
"outs": [dep + "-1" for dep in args],
"deps": list(args),
**run_kwargs,
}
)
return run
@pytest.fixture
def erepo_dir(make_tmp_dir):
return make_tmp_dir("erepo", scm=True, dvc=True)
tests/func/test_lockfile.py
-28
View File
@@ -1,8 +1,6 @@
from collections import OrderedDict
from operator import itemgetter
from textwrap import dedent
import pytest
import yaml
from dvc.dvcfile import PIPELINE_LOCK
@@ -20,32 +18,6 @@ FS_STRUCTURE = {
}
@pytest.fixture
def run_head(tmp_dir, dvc):
"""Output first line of each file to different file with '-1' appended."""
tmp_dir.gen(
"head.py",
dedent(
"""
import sys
for file in sys.argv[1:]:
with open(file) as f, open(file +"-1","w+") as w:
w.write(f.readline())
"""
),
)
def run(*args, **run_kwargs):
return dvc.run(
cmd="python head.py {}".format(" ".join(args)),
outs=[dep + "-1" for dep in args],
deps=args,
**run_kwargs
)
return run
def read_lock_file(file=PIPELINE_LOCK):
with open(file) as f:
data = parse_stage_for_update(f.read(), file)
tests/func/test_run_multistage.py
+38 -20
View File
@@ -115,28 +115,18 @@ def test_graph(tmp_dir, dvc):
dvc.run(deps=["baz"], outs=["foo"], cmd="echo baz > foo", name="3")
def test_run_dump_on_multistage(tmp_dir, dvc):
def test_run_dump_on_multistage(tmp_dir, dvc, run_head):
from dvc.dvcfile import Dvcfile, PIPELINE_FILE
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, PIPELINE_FILE)._load()
assert data == {
"stages": {
"copy-foo-foo1": {
"cmd": "cp foo foo1",
"wdir": "dir",
"deps": ["foo"],
"outs": ["foo1"],
tmp_dir.gen(
{
"dir": {
"foo": "foo\nfoo",
"bar": "bar\nbar",
"foobar": "foobar\foobar",
}
}
}
)
dvc.run(
cmd="cp foo foo2",
@@ -146,15 +136,43 @@ def test_run_dump_on_multistage(tmp_dir, dvc):
outs_persist=["foo2"],
always_changed=True,
)
assert Dvcfile(dvc, PIPELINE_FILE)._load()[0] == {
data = Dvcfile(dvc, PIPELINE_FILE)._load()[0]
assert data == {
"stages": {
"copy-foo-foo2": {
"cmd": "cp foo foo2",
"deps": ["foo"],
"outs_persist": ["foo2"],
"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 Dvcfile(dvc, PIPELINE_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}},
{"foobar-1": {"metric": True, "cache": False}},
],
},
**data["stages"],
}
}