tests: serialize pipeline file (#3985)

serialize: make params ordered

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 {}

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,

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,

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(

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)

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()
 

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

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"}

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
+    )