tests: serialize pipeline file (#3985)

serialize: make params ordered

Author: skshetry

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 @@
 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 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 @@ def _load(self, stage):
         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/serialize.py renamed to dvc/stage/serialize.py

@@ -3,15 +3,16 @@
 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 @@
 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 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 @@
 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 @@
 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"}