Repository: iterative/dvc

Issue:
plots: support yaml
DVC supports JSON and YAML for metrics, but not for plots.

Details: https://github.com/iterative/dvc.org/pull/1382#discussion_r434967192

---
UPDATE by Jorge:

- [ ] Let's remember to update docs when this is figured out 🙂 Either in iterative/dvc.org/pull/1382 or in it's own PR on top of those changes.

Discussion:

Patch:
diff --git a/dvc/command/plots.py b/dvc/command/plots.py
index e4d007f60..f495a55b5 100644
--- a/dvc/command/plots.py
+++ b/dvc/command/plots.py
@@ -156,7 +156,7 @@ def add_parser(subparsers, parent_parser):
     _add_output_arguments(plots_diff_parser)
     plots_diff_parser.set_defaults(func=CmdPlotsDiff)
 
-    PLOTS_MODIFY_HELP = "Modify plot props associated with a target file."
+    PLOTS_MODIFY_HELP = "Modify plot properties associated with a target file."
     plots_modify_parser = plots_subparsers.add_parser(
         "modify",
         parents=[parent_parser],
@@ -165,11 +165,14 @@ def add_parser(subparsers, parent_parser):
         formatter_class=argparse.RawDescriptionHelpFormatter,
     )
     plots_modify_parser.add_argument(
-        "target", help="Plot file to set props to.",
+        "target", help="Plot file to set properties to.",
     )
     _add_props_arguments(plots_modify_parser)
     plots_modify_parser.add_argument(
-        "--unset", nargs="*", help="Props to unset.",
+        "--unset",
+        nargs="*",
+        metavar="<property>",
+        help="Properties to unset.",
     )
     plots_modify_parser.set_defaults(func=CmdPlotsModify)
 
@@ -196,9 +199,15 @@ def _add_props_arguments(parser):
         default=None,  # Use default None to distinguish when it's not used
         help="Provided CSV ot TSV datafile does not have a header.",
     )
-    parser.add_argument("--title", default=None, help="Plot title.")
-    parser.add_argument("--x-label", default=None, help="X axis label.")
-    parser.add_argument("--y-label", default=None, help="Y axis lebel.")
+    parser.add_argument(
+        "--title", default=None, metavar="<text>", help="Plot title."
+    )
+    parser.add_argument(
+        "--x-label", default=None, metavar="<text>", help="X axis label."
+    )
+    parser.add_argument(
+        "--y-label", default=None, metavar="<text>", help="Y axis lebel."
+    )
 
 
 def _add_output_arguments(parser):
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/repo/plots/data.py b/dvc/repo/plots/data.py
index 68430e6a2..901548928 100644
--- a/dvc/repo/plots/data.py
+++ b/dvc/repo/plots/data.py
@@ -237,6 +237,10 @@ class YAMLPlotData(PlotData):
 
         return yaml.load(self.content, OrderedLoader)
 
+    def _processors(self):
+        parent_processors = super()._processors()
+        return [_find_data] + parent_processors
+
 
 def _load_from_revision(repo, datafile, revision):
     from dvc.repo.tree import RepoTree
diff --git a/dvc/stage/__init__.py b/dvc/stage/__init__.py
index 0728827e9..866203001 100644
--- a/dvc/stage/__init__.py
+++ b/dvc/stage/__init__.py
@@ -493,7 +493,11 @@ class Stage(params.StageParams):
             ret.append("changed checksum")
 
     def already_cached(self):
-        return self.deps_cached() and self.outs_cached()
+        return (
+            not self.changed_stage()
+            and self.deps_cached()
+            and self.outs_cached()
+        )
 
     def deps_cached(self):
         return all(not dep.changed() for dep in self.deps)
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/stage/imports.py b/dvc/stage/imports.py
index 49703c206..96a626e35 100644
--- a/dvc/stage/imports.py
+++ b/dvc/stage/imports.py
@@ -23,7 +23,7 @@ def sync_import(stage, dry=False, force=False):
     if dry:
         return
 
-    if not force and not stage.changed_stage() and stage.already_cached():
+    if not force and stage.already_cached():
         stage.outs[0].checkout()
     else:
         stage.deps[0].download(stage.outs[0])
diff --git a/dvc/stage/run.py b/dvc/stage/run.py
index 590863656..53bab38d2 100644
--- a/dvc/stage/run.py
+++ b/dvc/stage/run.py
@@ -82,7 +82,6 @@ def _is_cached(stage):
     cached = (
         not stage.is_callback
         and not stage.always_changed
-        and not stage.changed_stage()
         and stage.already_cached()
     )
     if cached:
diff --git a/dvc/serialize.py b/dvc/stage/serialize.py
similarity index 54%
rename from dvc/serialize.py
rename to dvc/stage/serialize.py
index 169bb7b50..cf80e399e 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,89 +33,96 @@ 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]):
-    """Return two types of values from stage:
+def _serialize_params_keys(params):
+    """
+    Returns the following format of data:
+     ['lr', 'train', {'params2.yaml': ['lr']}]
 
-    `keys` - which is list of params without values, used in a pipeline file
+    The output is sorted, with keys of params from default params file being
+    at the first, and then followed by entry of other files in lexicographic
+    order. The keys of those custom files are also sorted in the same order.
+    """
+    keys = []
+    for param_dep in sort_by_path(params):
+        dump = param_dep.dumpd()
+        path, params = dump[PARAM_PATH], dump[PARAM_PARAMS]
+        assert isinstance(params, (dict, list))
+        # when on no_exec, params are not filled and are saved as list
+        k = sorted(params.keys() if isinstance(params, dict) else params)
+        if not k:
+            continue
 
-    which is in the shape of:
-        ['lr', 'train', {'params2.yaml': ['lr']}]
+        if path == DEFAULT_PARAMS_FILE:
+            keys = k + keys
+        else:
+            keys.append({path: k})
+    return keys
 
-    `key_vals` - which is list of params with values, used in a lockfile
-    which is in the shape of:
+
+def _serialize_params_values(params: List[ParamsDependency]):
+    """Returns output of following format, used for lockfile:
         {'params.yaml': {'lr': '1', 'train': 2}, {'params2.yaml': {'lr': '1'}}
+
+    Default params file are always kept at the start, followed by others in
+    alphabetical order. The param values are sorted too(not recursively though)
     """
-    keys = []
     key_vals = OrderedDict()
-
     for param_dep in sort_by_path(params):
         dump = param_dep.dumpd()
         path, params = dump[PARAM_PATH], dump[PARAM_PARAMS]
         if isinstance(params, dict):
-            k = list(params.keys())
-            if not k:
-                continue
-            key_vals[path] = OrderedDict(
-                [(key, params[key]) for key in sorted(k)]
-            )
-        else:
-            assert isinstance(params, list)
-            k = params
-            key_vals = OrderedDict()
-        # params from default file is always kept at the start of the `params:`
-        if path == DEFAULT_PARAMS_FILE:
-            keys = k + keys
-            if key_vals:
+            kv = [(key, params[key]) for key in sorted(params.keys())]
+            key_vals[path] = OrderedDict(kv)
+            if path == DEFAULT_PARAMS_FILE:
                 key_vals.move_to_end(path, last=False)
-        else:
-            # if it's not a default file, change the shape
-            # to: {path: k}
-            keys.append({path: k})
-    return keys, key_vals
+    return key_vals
 
 
 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_keys(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 +135,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(
@@ -140,10 +148,11 @@ def to_single_stage_lockfile(stage: "Stage") -> dict:
         ]
         for items in [deps, stage.outs]
     ]
+    params = _serialize_params_values(params)
     if deps:
         res[PARAM_DEPS] = deps
     if params:
-        _, res[PARAM_PARAMS] = _serialize_params(params)
+        res[PARAM_PARAMS] = params
     if outs:
         res[PARAM_OUTS] = outs
 
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/setup.cfg b/setup.cfg
index f756962f0..fbcef890d 100644
--- a/setup.cfg
+++ b/setup.cfg
@@ -14,4 +14,6 @@ include_trailing_comma=true
 known_first_party=dvc,tests
 known_third_party=PyInstaller,RangeHTTPServer,boto3,colorama,configobj,distro,dpath,flaky,flufl,funcy,git,google,grandalf,mock,mockssh,moto,nanotime,networkx,packaging,paramiko,pathspec,pytest,requests,ruamel,setuptools,shortuuid,tqdm,voluptuous,yaml,zc
 line_length=79
+force_grid_wrap=0
+use_parentheses=True
 multi_line_output=3


Test patch:
diff --git a/tests/func/plots/test_plots.py b/tests/func/plots/test_plots.py
index 75835e636..cbd0afc07 100644
--- a/tests/func/plots/test_plots.py
+++ b/tests/func/plots/test_plots.py
@@ -10,9 +10,11 @@ import yaml
 from funcy import first
 
 from dvc.repo.plots.data import (
+    JSONPlotData,
     NoMetricInHistoryError,
     PlotData,
     PlotMetricTypeError,
+    YAMLPlotData,
 )
 from dvc.repo.plots.template import (
     NoDataForTemplateError,
@@ -581,3 +583,29 @@ def test_raise_on_wrong_field(tmp_dir, scm, dvc, run_copy_metrics):
 
     with pytest.raises(NoFieldInDataError):
         dvc.plots.show("metric.json", props={"y": "no_val"})
+
+
+def test_load_metric_from_dict_json(tmp_dir):
+    metric = [{"acccuracy": 1, "loss": 2}, {"accuracy": 3, "loss": 4}]
+    dmetric = {"train": metric}
+
+    plot_data = JSONPlotData("-", "revision", json.dumps(dmetric))
+
+    expected = metric
+    for d in expected:
+        d["rev"] = "revision"
+
+    assert list(map(dict, plot_data.to_datapoints())) == expected
+
+
+def test_load_metric_from_dict_yaml(tmp_dir):
+    metric = [{"acccuracy": 1, "loss": 2}, {"accuracy": 3, "loss": 4}]
+    dmetric = {"train": metric}
+
+    plot_data = YAMLPlotData("-", "revision", yaml.dump(dmetric))
+
+    expected = metric
+    for d in expected:
+        d["rev"] = "revision"
+
+    assert list(map(dict, plot_data.to_datapoints())) == expected
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
+    )
diff --git a/tests/unit/stage/test_serialize_pipeline_lock.py b/tests/unit/stage/test_serialize_pipeline_lock.py
new file mode 100644
index 000000000..85721e07c
--- /dev/null
+++ b/tests/unit/stage/test_serialize_pipeline_lock.py
@@ -0,0 +1,208 @@
+from collections import OrderedDict
+
+import pytest
+from voluptuous import Schema as _Schema
+
+from dvc.dvcfile import PIPELINE_FILE
+from dvc.schema import LOCK_FILE_STAGE_SCHEMA, LOCKFILE_SCHEMA
+from dvc.stage import PipelineStage, create_stage
+from dvc.stage.serialize import DEFAULT_PARAMS_FILE, to_lockfile
+from dvc.stage.serialize import (
+    to_single_stage_lockfile as _to_single_stage_lockfile,
+)
+from dvc.stage.utils import split_params_deps
+
+kwargs = {"name": "something", "cmd": "command", "path": PIPELINE_FILE}
+Schema = _Schema(LOCK_FILE_STAGE_SCHEMA)
+
+
+def to_single_stage_lockfile(stage):
+    """Validate schema on each serialization."""
+    e = _to_single_stage_lockfile(stage)
+    assert Schema(e)
+    return e
+
+
+def test_lock(dvc):
+    stage = create_stage(PipelineStage, dvc, **kwargs)
+    assert to_single_stage_lockfile(stage) == {"cmd": "command"}
+
+
+def test_lock_deps(dvc):
+    stage = create_stage(PipelineStage, dvc, deps=["input"], **kwargs)
+    stage.deps[0].info = {"md5": "md-five"}
+    assert to_single_stage_lockfile(stage) == OrderedDict(
+        [
+            ("cmd", "command"),
+            ("deps", [OrderedDict([("path", "input"), ("md5", "md-five")])]),
+        ]
+    )
+
+
+def test_lock_deps_order(dvc):
+    stage = create_stage(
+        PipelineStage, dvc, deps=["input1", "input0"], **kwargs
+    )
+    stage.deps[0].info = {"md5": "md-one1"}
+    stage.deps[1].info = {"md5": "md-zer0"}
+    assert to_single_stage_lockfile(stage) == OrderedDict(
+        [
+            ("cmd", "command"),
+            (
+                "deps",
+                [
+                    OrderedDict([("path", "input0"), ("md5", "md-zer0")]),
+                    OrderedDict([("path", "input1"), ("md5", "md-one1")]),
+                ],
+            ),
+        ]
+    )
+
+
+def test_lock_params(dvc):
+    stage = create_stage(
+        PipelineStage, dvc, params=["lorem.ipsum", "abc"], **kwargs
+    )
+    stage.deps[0].info = {"lorem.ipsum": {"lorem1": 1, "lorem2": 2}, "abc": 3}
+    assert to_single_stage_lockfile(stage)["params"][
+        DEFAULT_PARAMS_FILE
+    ] == OrderedDict([("abc", 3), ("lorem.ipsum", {"lorem1": 1, "lorem2": 2})])
+
+
+def test_lock_params_file_sorted(dvc):
+    stage = create_stage(
+        PipelineStage,
+        dvc,
+        params=[
+            "lorem.ipsum",
+            "abc",
+            {"myparams.yaml": ["foo", "foobar"]},
+            {"a-params-file.yaml": ["bar", "barr"]},
+        ],
+        **kwargs
+    )
+    stage.deps[0].info = {"lorem.ipsum": {"lorem1": 1, "lorem2": 2}, "abc": 3}
+    stage.deps[1].info = {"foo": ["f", "o", "o"], "foobar": "foobar"}
+    stage.deps[2].info = {"bar": ["b", "a", "r"], "barr": "barr"}
+    assert to_single_stage_lockfile(stage)["params"] == OrderedDict(
+        [
+            (
+                DEFAULT_PARAMS_FILE,
+                OrderedDict(
+                    [("abc", 3), ("lorem.ipsum", {"lorem1": 1, "lorem2": 2})]
+                ),
+            ),
+            (
+                "a-params-file.yaml",
+                OrderedDict([("bar", ["b", "a", "r"]), ("barr", "barr")]),
+            ),
+            (
+                "myparams.yaml",
+                OrderedDict([("foo", ["f", "o", "o"]), ("foobar", "foobar")]),
+            ),
+        ]
+    )
+
+
+def test_lock_params_no_values_filled(dvc):
+    stage = create_stage(
+        PipelineStage, dvc, params=["lorem.ipsum", "abc"], **kwargs
+    )
+    assert to_single_stage_lockfile(stage) == {"cmd": "command"}
+
+
+@pytest.mark.parametrize("typ", ["plots", "metrics", "outs"])
+def test_lock_outs(dvc, typ):
+    stage = create_stage(PipelineStage, dvc, **{typ: ["input"]}, **kwargs)
+    stage.outs[0].info = {"md5": "md-five"}
+    assert to_single_stage_lockfile(stage) == OrderedDict(
+        [
+            ("cmd", "command"),
+            ("outs", [OrderedDict([("path", "input"), ("md5", "md-five")])]),
+        ]
+    )
+
+
+@pytest.mark.parametrize("typ", ["plots", "metrics", "outs"])
+def test_lock_outs_order(dvc, typ):
+    stage = create_stage(
+        PipelineStage, dvc, **{typ: ["input1", "input0"]}, **kwargs
+    )
+    stage.outs[0].info = {"md5": "md-one1"}
+    stage.outs[1].info = {"md5": "md-zer0"}
+    assert to_single_stage_lockfile(stage) == OrderedDict(
+        [
+            ("cmd", "command"),
+            (
+                "outs",
+                [
+                    OrderedDict([("path", "input0"), ("md5", "md-zer0")]),
+                    OrderedDict([("path", "input1"), ("md5", "md-one1")]),
+                ],
+            ),
+        ]
+    )
+
+
+def test_dump_appropriate_checksums(dvc):
+    stage = create_stage(
+        PipelineStage, dvc, deps=["s3://dvc-temp/file"], **kwargs
+    )
+    stage.deps[0].info = {"etag": "is-it-etag", "md5": "or-md5?"}
+    assert to_single_stage_lockfile(stage) == OrderedDict(
+        [
+            ("cmd", "command"),
+            (
+                "deps",
+                [
+                    OrderedDict(
+                        [
+                            ("path", "s3://dvc-temp/file"),
+                            ("etag", "is-it-etag"),
+                        ]
+                    )
+                ],
+            ),
+        ]
+    )
+
+
+def test_order(dvc):
+    stage = create_stage(
+        PipelineStage,
+        dvc,
+        deps=["input"],
+        outs=["output"],
+        params=["foo-param"],
+        **kwargs
+    )
+    params, deps = split_params_deps(stage)
+
+    deps[0].info = {"md5": "md-five"}
+    params[0].info = {"foo-param": "value"}
+    stage.outs[0].info = {"md5": "md5-output"}
+
+    assert to_single_stage_lockfile(stage) == OrderedDict(
+        [
+            ("cmd", "command"),
+            ("deps", [{"path": "input", "md5": "md-five"}]),
+            ("params", {"params.yaml": {"foo-param": "value"}}),
+            ("outs", [{"path": "output", "md5": "md5-output"}]),
+        ]
+    )
+
+
+def test_to_lockfile(dvc):
+    stage = create_stage(PipelineStage, dvc, deps=["input"], **kwargs)
+    stage.deps[0].info = {"md5": "md-five"}
+    entry = to_lockfile(stage)
+    assert len(entry) == 1
+    _Schema(LOCKFILE_SCHEMA)(entry)
+    assert entry == {
+        "something": OrderedDict(
+            [
+                ("cmd", "command"),
+                ("deps", [{"path": "input", "md5": "md-five"}]),
+            ]
+        )
+    }
diff --git a/tests/unit/stage/test_stage.py b/tests/unit/stage/test_stage.py
index 3da30f23a..7941322e9 100644
--- a/tests/unit/stage/test_stage.py
+++ b/tests/unit/stage/test_stage.py
@@ -2,7 +2,6 @@ import os
 import signal
 import subprocess
 import threading
-from unittest import TestCase
 
 import mock
 import pytest
@@ -50,12 +49,11 @@ def test_meta_ignored():
         assert stage.compute_md5() == "e9521a22111493406ea64a88cda63e0b"
 
 
-class TestPathConversion(TestCase):
-    def test(self):
-        stage = Stage(None, "path")
+def test_path_conversion(dvc):
+    stage = Stage(dvc, "path")
 
-        stage.wdir = os.path.join("..", "..")
-        self.assertEqual(stage.dumpd()["wdir"], "../..")
+    stage.wdir = os.path.join("..", "..")
+    assert stage.dumpd()["wdir"] == "../.."
 
 
 def test_stage_update(mocker):