From d201fad1e31716fea3c745f3f2e36f1e96145374 Mon Sep 17 00:00:00 2001 From: Saugat Pachhai Date: Tue, 9 Jun 2020 11:22:55 +0000 Subject: [PATCH] tests: serialize pipeline file (#3985) serialize: make params ordered --- dvc/dependency/__init__.py | 1 + dvc/dvcfile.py | 2 +- dvc/stage/cache.py | 7 +- dvc/{ => stage}/serialize.py | 81 ++++---- dvc/stage/utils.py | 7 + tests/func/test_lockfile.py | 4 +- tests/func/test_repro_multistage.py | 4 +- tests/unit/stage/test_loader_pipeline_file.py | 8 +- .../stage/test_serialize_pipeline_file.py | 196 ++++++++++++++++++ 9 files changed, 259 insertions(+), 51 deletions(-) rename dvc/{ => stage}/serialize.py (70%) create mode 100644 tests/unit/stage/test_serialize_pipeline_file.py diff --git a/dvc/dependency/__init__.py b/dvc/dependency/__init__.py index 10a2f787a..1a778dbfc 100644 --- a/dvc/dependency/__init__.py +++ b/dvc/dependency/__init__.py @@ -80,6 +80,7 @@ def loadd_from(stage, d_list): def loads_from(stage, s_list, erepo=None): + assert isinstance(s_list, list) ret = [] for s in s_list: info = {RepoDependency.PARAM_REPO: erepo} if erepo else {} diff --git a/dvc/dvcfile.py b/dvc/dvcfile.py index 298fe1bdf..df8a61b7c 100644 --- a/dvc/dvcfile.py +++ b/dvc/dvcfile.py @@ -5,8 +5,8 @@ import os from voluptuous import MultipleInvalid import dvc.prompt as prompt -from dvc import serialize from dvc.exceptions import DvcException +from dvc.stage import serialize from dvc.stage.exceptions import ( StageFileAlreadyExistsError, StageFileBadNameError, diff --git a/dvc/stage/cache.py b/dvc/stage/cache.py index 8f26173bb..f20be749f 100644 --- a/dvc/stage/cache.py +++ b/dvc/stage/cache.py @@ -7,12 +7,13 @@ from funcy import first from voluptuous import Invalid from dvc.schema import COMPILED_LOCK_FILE_STAGE_SCHEMA -from dvc.serialize import to_single_stage_lockfile -from dvc.stage.loader import StageLoader from dvc.utils import dict_sha256, relpath from dvc.utils.fs import makedirs from dvc.utils.yaml import dump_yaml +from .loader import StageLoader +from .serialize import to_single_stage_lockfile + logger = logging.getLogger(__name__) @@ -78,7 +79,7 @@ class StageCache: return None def _create_stage(self, cache, wdir=None): - from dvc.stage import create_stage, PipelineStage + from . import create_stage, PipelineStage stage = create_stage( PipelineStage, diff --git a/dvc/serialize.py b/dvc/stage/serialize.py similarity index 70% rename from dvc/serialize.py rename to dvc/stage/serialize.py index 169bb7b50..65914b0bf 100644 --- a/dvc/serialize.py +++ b/dvc/stage/serialize.py @@ -3,15 +3,16 @@ from functools import partial from operator import attrgetter from typing import TYPE_CHECKING, List -from funcy import lsplit, rpartial +from funcy import post_processing 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.yaml import parse_yaml_for_update +from .params import StageParams +from .utils import resolve_wdir, split_params_deps + if TYPE_CHECKING: from dvc.stage import PipelineStage, Stage @@ -32,23 +33,34 @@ DEFAULT_PARAMS_FILE = ParamsDependency.DEFAULT_PARAMS_FILE sort_by_path = partial(sorted, key=attrgetter("def_path")) -def _get_out(out): - res = OrderedDict() +@post_processing(OrderedDict) +def _get_flags(out): if not out.use_cache: - res[PARAM_CACHE] = False + yield PARAM_CACHE, False if out.persist: - res[PARAM_PERSIST] = True + yield PARAM_PERSIST, True if out.plot and isinstance(out.plot, dict): - res.update(out.plot) - return out.def_path if not res else {out.def_path: res} + # notice `out.plot` is not sorted + # `out.plot` is in the same order as is in the file when read + # and, should be dumped as-is without any sorting + yield from out.plot.items() -def _get_outs(outs): - return [_get_out(out) for out in sort_by_path(outs)] +def _serialize_out(out): + flags = _get_flags(out) + return out.def_path if not flags else {out.def_path: flags} -def get_params_deps(stage: "PipelineStage"): - return lsplit(rpartial(isinstance, ParamsDependency), stage.deps) +def _serialize_outs(outputs: List[BaseOutput]): + outs, metrics, plots = [], [], [] + for out in sort_by_path(outputs): + bucket = outs + if out.plot: + bucket = plots + elif out.metric: + bucket = metrics + bucket.append(_serialize_out(out)) + return outs, metrics, plots def _serialize_params(params: List[ParamsDependency]): @@ -70,16 +82,15 @@ def _serialize_params(params: List[ParamsDependency]): dump = param_dep.dumpd() path, params = dump[PARAM_PATH], dump[PARAM_PARAMS] if isinstance(params, dict): - k = list(params.keys()) + k = sorted(params.keys()) if not k: continue - key_vals[path] = OrderedDict( - [(key, params[key]) for key in sorted(k)] - ) + key_vals[path] = OrderedDict([(key, params[key]) for key in k]) else: assert isinstance(params, list) - k = params - key_vals = OrderedDict() + # no params values available here, entry will be skipped for lock + k = sorted(params) + # params from default file is always kept at the start of the `params:` if path == DEFAULT_PARAMS_FILE: keys = k + keys @@ -93,28 +104,20 @@ def _serialize_params(params: List[ParamsDependency]): def to_pipeline_file(stage: "PipelineStage"): - params, deps = get_params_deps(stage) - serialized_params, _ = _serialize_params(params) + wdir = resolve_wdir(stage.wdir, stage.path) + params, deps = split_params_deps(stage) + deps = sorted([d.def_path for d in deps]) + params, _ = _serialize_params(params) + outs, metrics, plots = _serialize_outs(stage.outs) res = [ (stage.PARAM_CMD, stage.cmd), - (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), - ( - PARAM_OUTS, - _get_outs( - [out for out in stage.outs if not (out.metric or out.plot)] - ), - ), - ( - stage.PARAM_METRICS, - _get_outs([out for out in stage.outs if out.metric]), - ), - ( - stage.PARAM_PLOTS, - _get_outs([out for out in stage.outs if out.plot]), - ), + (stage.PARAM_WDIR, wdir), + (stage.PARAM_DEPS, deps), + (stage.PARAM_PARAMS, params), + (stage.PARAM_OUTS, outs), + (stage.PARAM_METRICS, metrics), + (stage.PARAM_PLOTS, plots), (stage.PARAM_FROZEN, stage.frozen), (stage.PARAM_ALWAYS_CHANGED, stage.always_changed), ] @@ -127,7 +130,7 @@ def to_single_stage_lockfile(stage: "Stage") -> dict: assert stage.cmd res = OrderedDict([("cmd", stage.cmd)]) - params, deps = get_params_deps(stage) + params, deps = split_params_deps(stage) deps, outs = [ [ OrderedDict( diff --git a/dvc/stage/utils.py b/dvc/stage/utils.py index 34cb879af..939803e8b 100644 --- a/dvc/stage/utils.py +++ b/dvc/stage/utils.py @@ -2,9 +2,12 @@ import os import pathlib from itertools import product +from funcy import lsplit, rpartial + from dvc import dependency, output from dvc.utils.fs import path_isin +from ..dependency import ParamsDependency from ..remote import LocalRemote, S3Remote from ..utils import dict_md5, format_link, relpath from .exceptions import ( @@ -196,3 +199,7 @@ def get_dump(stage): }.items() if value } + + +def split_params_deps(stage): + return lsplit(rpartial(isinstance, ParamsDependency), stage.deps) diff --git a/tests/func/test_lockfile.py b/tests/func/test_lockfile.py index 87b3ebb00..2d2231368 100644 --- a/tests/func/test_lockfile.py +++ b/tests/func/test_lockfile.py @@ -4,7 +4,7 @@ from operator import itemgetter import yaml from dvc.dvcfile import PIPELINE_LOCK -from dvc.serialize import get_params_deps +from dvc.stage.utils import split_params_deps from dvc.utils.fs import remove from dvc.utils.yaml import parse_yaml_for_update from tests.func.test_run_multistage import supported_params @@ -137,7 +137,7 @@ def test_params_dump(tmp_dir, dvc, run_head): assert not dvc.reproduce(stage.addressing) # let's change the order of params and dump them in pipeline file - params, _ = get_params_deps(stage) + params, _ = split_params_deps(stage) for param in params: param.params.reverse() diff --git a/tests/func/test_repro_multistage.py b/tests/func/test_repro_multistage.py index c26f3ad62..0507aacbc 100644 --- a/tests/func/test_repro_multistage.py +++ b/tests/func/test_repro_multistage.py @@ -509,7 +509,7 @@ def test_cyclic_graph_error(tmp_dir, dvc, run_copy): def test_repro_multiple_params(tmp_dir, dvc): from tests.func.test_run_multistage import supported_params - from dvc.serialize import get_params_deps + from dvc.stage.utils import split_params_deps with (tmp_dir / "params2.yaml").open("w+") as f: yaml.dump(supported_params, f) @@ -529,7 +529,7 @@ def test_repro_multiple_params(tmp_dir, dvc): cmd="cat params2.yaml params.yaml > bar", ) - params, deps = get_params_deps(stage) + params, deps = split_params_deps(stage) assert len(params) == 2 assert len(deps) == 1 assert len(stage.outs) == 1 diff --git a/tests/unit/stage/test_loader_pipeline_file.py b/tests/unit/stage/test_loader_pipeline_file.py index 5ee89c661..264576c86 100644 --- a/tests/unit/stage/test_loader_pipeline_file.py +++ b/tests/unit/stage/test_loader_pipeline_file.py @@ -5,9 +5,9 @@ from itertools import chain import pytest from dvc.dvcfile import PIPELINE_FILE, Dvcfile -from dvc.serialize import get_params_deps from dvc.stage import PipelineStage, create_stage from dvc.stage.loader import StageLoader +from dvc.stage.serialize import split_params_deps @pytest.fixture @@ -61,7 +61,7 @@ def test_fill_from_lock_params(dvc, lock_data): "ipsum": "ipsum" }, } - params_deps = get_params_deps(stage)[0] + params_deps = split_params_deps(stage)[0] assert set(params_deps[0].params) == {"lorem", "lorem.ipsum"} assert set(params_deps[1].params) == {"ipsum", "foobar"} assert not params_deps[0].info @@ -81,7 +81,7 @@ def test_fill_from_lock_missing_params_section(dvc, lock_data): outs=["bar"], params=["lorem", "lorem.ipsum", {"myparams.yaml": ["ipsum"]}], ) - params_deps = get_params_deps(stage)[0] + params_deps = split_params_deps(stage)[0] StageLoader.fill_from_lock(stage, lock_data) assert not params_deps[0].info and not params_deps[1].info @@ -180,7 +180,7 @@ def test_load_stage_with_params(dvc, stage_data, lock_data): dvcfile = Dvcfile(dvc, PIPELINE_FILE) stage = StageLoader.load_stage(dvcfile, "stage-1", stage_data, lock_data) - params, deps = get_params_deps(stage) + params, deps = split_params_deps(stage) assert deps[0].def_path == "foo" and stage.outs[0].def_path == "bar" assert params[0].def_path == "params.yaml" assert params[0].info == {"lorem": "ipsum"} diff --git a/tests/unit/stage/test_serialize_pipeline_file.py b/tests/unit/stage/test_serialize_pipeline_file.py new file mode 100644 index 000000000..113591bd0 --- /dev/null +++ b/tests/unit/stage/test_serialize_pipeline_file.py @@ -0,0 +1,196 @@ +import os + +import pytest +from voluptuous import Schema as _Schema + +from dvc import output +from dvc.dvcfile import PIPELINE_FILE +from dvc.schema import SINGLE_PIPELINE_STAGE_SCHEMA +from dvc.stage import PipelineStage, create_stage +from dvc.stage.serialize import to_pipeline_file as _to_pipeline_file + +kwargs = {"name": "something", "cmd": "command", "path": PIPELINE_FILE} +Schema = _Schema(SINGLE_PIPELINE_STAGE_SCHEMA) + + +def to_pipeline_file(stage): + """Validate schema on each serialization.""" + e = _to_pipeline_file(stage) + assert len(Schema(e)) == 1 + return e + + +def test_cmd(dvc): + stage = create_stage(PipelineStage, dvc, **kwargs) + entry = to_pipeline_file(stage) + assert entry == {"something": {"cmd": "command"}} + + +def test_wdir(dvc): + stage = create_stage(PipelineStage, dvc, **kwargs) + assert stage.PARAM_WDIR not in to_pipeline_file(stage)["something"] + + stage.wdir = os.curdir + assert stage.PARAM_WDIR not in to_pipeline_file(stage)["something"] + + stage.wdir = "some-dir" + assert to_pipeline_file(stage)["something"][stage.PARAM_WDIR] == "some-dir" + + +def test_deps_sorted(dvc): + stage = create_stage( + PipelineStage, dvc, deps=["a", "quick", "lazy", "fox"], **kwargs + ) + assert to_pipeline_file(stage)["something"][stage.PARAM_DEPS] == [ + "a", + "fox", + "lazy", + "quick", + ] + + +def test_outs_sorted(dvc): + stage = create_stage( + PipelineStage, + dvc, + outs=["too", "many", "outs"], + deps=["foo"], + **kwargs, + ) + assert to_pipeline_file(stage)["something"][stage.PARAM_OUTS] == [ + "many", + "outs", + "too", + ] + + +def test_params_sorted(dvc): + params = [ + "lorem", + "ipsum", + {"custom.yaml": ["wxyz", "pqrs", "baz"]}, + {"params.yaml": ["barr"]}, + ] + stage = create_stage( + PipelineStage, dvc, outs=["bar"], deps=["foo"], params=params, **kwargs + ) + assert to_pipeline_file(stage)["something"][stage.PARAM_PARAMS] == [ + "barr", + "ipsum", + "lorem", + {"custom.yaml": ["baz", "pqrs", "wxyz"]}, + ] + + +def test_params_file_sorted(dvc): + params = [ + "lorem", + "ipsum", + {"custom.yaml": ["wxyz", "pqrs", "baz"]}, + {"a-file-of-params.yaml": ["barr"]}, + ] + stage = create_stage( + PipelineStage, dvc, outs=["bar"], deps=["foo"], params=params, **kwargs + ) + assert to_pipeline_file(stage)["something"][stage.PARAM_PARAMS] == [ + "ipsum", + "lorem", + {"a-file-of-params.yaml": ["barr"]}, + {"custom.yaml": ["baz", "pqrs", "wxyz"]}, + ] + + +@pytest.mark.parametrize( + "typ, extra", + [("plots", {"plot": True}), ("metrics", {"metric": True}), ("outs", {})], +) +def test_outs_and_outs_flags_are_sorted(dvc, typ, extra): + stage = create_stage(PipelineStage, dvc, deps=["input"], **kwargs) + stage.outs += output.loads_from(stage, ["barr"], use_cache=False, **extra) + stage.outs += output.loads_from( + stage, ["foobar"], use_cache=False, persist=True, **extra + ) + stage.outs += output.loads_from(stage, ["foo"], persist=True, **extra) + stage.outs += output.loads_from(stage, ["bar"], **extra) + + serialized_outs = to_pipeline_file(stage)["something"][typ] + assert serialized_outs == [ + "bar", + {"barr": {"cache": False}}, + {"foo": {"persist": True}}, + {"foobar": {"cache": False, "persist": True}}, + ] + assert list(serialized_outs[3]["foobar"].keys()) == ["cache", "persist"] + + +def test_plot_props(dvc): + props = {"x": "1"} + stage = create_stage(PipelineStage, dvc, plots=["plot_file"], **kwargs) + stage.outs[0].plot = props + + assert to_pipeline_file(stage)["something"][stage.PARAM_PLOTS] == [ + {"plot_file": props} + ] + + +def test_frozen(dvc): + stage = create_stage( + PipelineStage, dvc, outs=["output"], deps=["input"], **kwargs + ) + assert stage.PARAM_FROZEN not in to_pipeline_file(stage)["something"] + + stage = create_stage(PipelineStage, dvc, **kwargs, frozen=True) + assert to_pipeline_file(stage)["something"][stage.PARAM_FROZEN] is True + + +def test_always_changed(dvc): + stage = create_stage( + PipelineStage, dvc, outs=["output"], deps=["input"], **kwargs + ) + assert ( + stage.PARAM_ALWAYS_CHANGED not in to_pipeline_file(stage)["something"] + ) + + stage = create_stage(PipelineStage, dvc, **kwargs, always_changed=True) + assert ( + to_pipeline_file(stage)["something"][stage.PARAM_ALWAYS_CHANGED] + is True + ) + + +def test_order(dvc): + stage = create_stage( + PipelineStage, + dvc, + outs=["output"], + deps=["input"], + **kwargs, + always_changed=True, + frozen=True, + ) + # `create_stage` checks for existence of `wdir` + stage.wdir = "some-dir" + assert list(to_pipeline_file(stage)["something"].keys()) == [ + "cmd", + "wdir", + "deps", + "outs", + "frozen", + "always_changed", + ] + + +@pytest.mark.parametrize( + "typ", ["outs", "metrics", "plots", "params", "deps", None] +) +def test_order_deps_outs(dvc, typ): + all_types = ["deps", "params", "outs", "metrics", "plots"] + all_types = [item for item in all_types if item != typ] + extra = {key: [f"foo-{i}"] for i, key in enumerate(all_types)} + + stage = create_stage(PipelineStage, dvc, **kwargs, **extra) + assert typ not in to_pipeline_file(stage)["something"] + assert ( + list(to_pipeline_file(stage)["something"].keys()) + == ["cmd"] + all_types + )