diff --git a/dvc/dvcfile.py b/dvc/dvcfile.py index 37144805b..44207352c 100644 --- a/dvc/dvcfile.py +++ b/dvc/dvcfile.py @@ -6,6 +6,7 @@ import os from voluptuous import MultipleInvalid from dvc.exceptions import DvcException +from dvc.parsing.versions import LOCKFILE_VERSION, META_KWD from dvc.stage import serialize from dvc.stage.exceptions import ( StageFileBadNameError, @@ -113,14 +114,13 @@ class FileMixin: with self.repo.tree.open(self.path) as fd: stage_text = fd.read() d = parse_yaml(stage_text, self.path) - self.validate(d, self.relpath) - return d, stage_text + return self.validate(d, self.relpath), stage_text @classmethod def validate(cls, d, fname=None): assert isinstance(cls.SCHEMA, collections.abc.Callable) try: - cls.SCHEMA(d) # pylint: disable=not-callable + return cls.SCHEMA(d) # pylint: disable=not-callable except MultipleInvalid as exc: raise StageFileFormatError(f"'{fname}' format error: {exc}") @@ -273,8 +273,40 @@ class PipelineFile(FileMixin): raise NotImplementedError +def get_lockfile_schema(d): + from dvc.schema import ( + COMPILED_LOCKFILE_V1_SCHEMA, + COMPILED_LOCKFILE_V2_SCHEMA, + ) + + schema = { + LOCKFILE_VERSION.V1: COMPILED_LOCKFILE_V1_SCHEMA, + LOCKFILE_VERSION.V2: COMPILED_LOCKFILE_V2_SCHEMA, + } + + version = LOCKFILE_VERSION.from_dict(d) + return schema[version] + + +def migrate_lock_v1_to_v2(d, meta): + stages = {k: v for k, v in d.items()} + + for key in stages: + d.pop(key) + + # forcing order, meta should always be at the top + d.update(meta) + d["stages"] = stages + + class Lockfile(FileMixin): - from dvc.schema import COMPILED_LOCKFILE_SCHEMA as SCHEMA + @classmethod + def validate(cls, d, fname=None): + schema = get_lockfile_schema(d) + try: + return schema(d) + except MultipleInvalid as exc: + raise StageFileFormatError(f"'{fname}' format error: {exc}") def load(self): if not self.exists(): @@ -282,26 +314,42 @@ class Lockfile(FileMixin): data = load_yaml(self.path, tree=self.repo.tree) try: - self.validate(data, fname=self.relpath) - except StageFileFormatError: + data = self.validate(data, fname=self.relpath) + except StageFileFormatError as exc: raise LockfileCorruptedError( f"Lockfile '{self.relpath}' is corrupted." - ) + ) from exc return data + @property + def meta(self): + version = LOCKFILE_VERSION.V2.value # pylint:disable=no-member + return {META_KWD: {"version": version}} + def dump(self, stage, **kwargs): stage_data = serialize.to_lockfile(stage) with modify_yaml(self.path, tree=self.repo.tree) as data: - if not data: - logger.info("Generating lock file '%s'", self.relpath) + version = LOCKFILE_VERSION.from_dict(data) + if version == LOCKFILE_VERSION.V1: + logger.info( + "Migrating lock file '%s' from v1 to v2", self.relpath + ) + migrate_lock_v1_to_v2(data, self.meta) + else: + if not data: + data.update(self.meta) + # order is important, meta should always be at the top + logger.info("Generating lock file '%s'", self.relpath) - modified = data.get(stage.name, {}) != stage_data.get( + data["stages"] = data.get("stages", {}) + modified = data["stages"].get(stage.name, {}) != stage_data.get( stage.name, {} ) if modified: logger.info("Updating lock file '%s'", self.relpath) - data.update(stage_data) + + data["stages"].update(stage_data) if modified: self.repo.scm.track_file(self.relpath) @@ -314,13 +362,15 @@ class Lockfile(FileMixin): d = parse_yaml_for_update(f.read(), self.path) self.validate(d, self.path) - if stage.name not in d: + version = LOCKFILE_VERSION.from_dict(d) + data = d if version == LOCKFILE_VERSION.V1 else d.get("stages", {}) + if stage.name not in data: return logger.debug("Removing '%s' from '%s'", stage.name, self.path) - del d[stage.name] + del data[stage.name] - if d: + if data: dump_yaml(self.path, d) else: self.remove() diff --git a/dvc/parsing/versions.py b/dvc/parsing/versions.py new file mode 100644 index 000000000..b0a82c1dc --- /dev/null +++ b/dvc/parsing/versions.py @@ -0,0 +1,38 @@ +import enum +from collections.abc import Mapping + +from voluptuous import validators + +VERSION_KWD = "version" +META_KWD = "meta" + + +def lockfile_version_schema(value): + expected = [LOCKFILE_VERSION.V2.value] # pylint: disable=no-member + msg = "invalid version {}, expected one of {}".format(value, expected) + return validators.Any(*expected, msg=msg)(value) + + +class VersionEnum(str, enum.Enum): + @classmethod + def all_versions(cls): + return [v.value for v in cls] + + +class LOCKFILE_VERSION(VersionEnum): + V1 = "1.0" + V2 = "2.0" + + @classmethod + def from_dict(cls, data): + # 1) if it's empty or or is not a dict, use the latest one (V2). + # 2) use the `meta.version` + # 3) if it's not in any of the supported version, use the latest schema + # 4) if there's no version identifier, it's a V1 + if not data or not isinstance(data, Mapping): + return cls(cls.V2) + + version = data.get(META_KWD, {}).get(VERSION_KWD) + if version: + return cls(version if version in cls.all_versions() else cls.V2) + return cls(cls.V1) diff --git a/dvc/schema.py b/dvc/schema.py index a7199ef4f..143f46180 100644 --- a/dvc/schema.py +++ b/dvc/schema.py @@ -4,6 +4,7 @@ from dvc import dependency, output from dvc.hash_info import HashInfo from dvc.output import CHECKSUMS_SCHEMA, BaseOutput from dvc.parsing import DO_KWD, FOREACH_KWD, VARS_KWD +from dvc.parsing.versions import META_KWD, VERSION_KWD, lockfile_version_schema from dvc.stage.params import StageParams STAGES = "stages" @@ -13,7 +14,7 @@ SINGLE_STAGE_SCHEMA = { StageParams.PARAM_WDIR: Any(str, None), StageParams.PARAM_DEPS: Any([dependency.SCHEMA], None), StageParams.PARAM_OUTS: Any([output.SCHEMA], None), - StageParams.PARAM_LOCKED: bool, # backard compatibility + StageParams.PARAM_LOCKED: bool, # backward compatibility StageParams.PARAM_FROZEN: bool, StageParams.PARAM_META: object, StageParams.PARAM_ALWAYS_CHANGED: bool, @@ -33,7 +34,14 @@ LOCK_FILE_STAGE_SCHEMA = { StageParams.PARAM_PARAMS: {str: {str: object}}, StageParams.PARAM_OUTS: [DATA_SCHEMA], } -LOCKFILE_SCHEMA = {str: LOCK_FILE_STAGE_SCHEMA} + +LOCKFILE_STAGES_SCHEMA = {str: LOCK_FILE_STAGE_SCHEMA} + +LOCKFILE_V1_SCHEMA = LOCKFILE_STAGES_SCHEMA +LOCKFILE_V2_SCHEMA = { + STAGES: LOCKFILE_STAGES_SCHEMA, + META_KWD: {Required(VERSION_KWD): lockfile_version_schema}, +} OUT_PSTAGE_DETAILED_SCHEMA = { str: { @@ -100,4 +108,5 @@ MULTI_STAGE_SCHEMA = { COMPILED_SINGLE_STAGE_SCHEMA = Schema(SINGLE_STAGE_SCHEMA) COMPILED_MULTI_STAGE_SCHEMA = Schema(MULTI_STAGE_SCHEMA) COMPILED_LOCK_FILE_STAGE_SCHEMA = Schema(LOCK_FILE_STAGE_SCHEMA) -COMPILED_LOCKFILE_SCHEMA = Schema(LOCKFILE_SCHEMA) +COMPILED_LOCKFILE_V1_SCHEMA = Schema(LOCKFILE_V1_SCHEMA) +COMPILED_LOCKFILE_V2_SCHEMA = Schema(LOCKFILE_V2_SCHEMA) diff --git a/dvc/stage/loader.py b/dvc/stage/loader.py index 4dd0a9274..761f8ca54 100644 --- a/dvc/stage/loader.py +++ b/dvc/stage/loader.py @@ -8,6 +8,7 @@ from funcy import cached_property, get_in, lcat, once, project from dvc import dependency, output from dvc.hash_info import HashInfo from dvc.parsing import FOREACH_KWD, JOIN, DataResolver, EntryNotFound +from dvc.parsing.versions import LOCKFILE_VERSION from dvc.path_info import PathInfo from . import PipelineStage, Stage, loads_from @@ -24,7 +25,13 @@ class StageLoader(Mapping): self.data = data or {} self.stages_data = self.data.get("stages", {}) self.repo = self.dvcfile.repo - self._lockfile_data = lockfile_data or {} + + lockfile_data = lockfile_data or {} + version = LOCKFILE_VERSION.from_dict(lockfile_data) + if version == LOCKFILE_VERSION.V1: + self._lockfile_data = lockfile_data + else: + self._lockfile_data = lockfile_data.get("stages", {}) @cached_property def resolver(self): diff --git a/tests/func/test_dvcfile.py b/tests/func/test_dvcfile.py index ce19c1f09..4d66592b9 100644 --- a/tests/func/test_dvcfile.py +++ b/tests/func/test_dvcfile.py @@ -201,10 +201,12 @@ def test_remove_stage_lockfile(tmp_dir, dvc, run_copy): lock_file = dvc_file._lockfile assert dvc_file.exists() assert lock_file.exists() - assert {"copy-bar-foobar", "copy-foo-bar"} == set(lock_file.load().keys()) + assert {"copy-bar-foobar", "copy-foo-bar"} == set( + lock_file.load()["stages"].keys() + ) lock_file.remove_stage(stage) - assert ["copy-bar-foobar"] == list(lock_file.load().keys()) + assert ["copy-bar-foobar"] == list(lock_file.load()["stages"].keys()) # sanity check stage2.reload() @@ -374,7 +376,9 @@ def test_dvcfile_dump_preserves_comments(tmp_dir, dvc): ], ) def test_dvcfile_try_dumping_parametrized_stage(tmp_dir, dvc, data, name): - dump_yaml("dvc.yaml", {"stages": data, "vars": [{"foo": "foobar"}]}) + dump_yaml( + "dvc.yaml", {"stages": data, "vars": [{"foo": "foobar"}]}, + ) stage = dvc.stage.load_one(name=name) dvcfile = stage.dvcfile diff --git a/tests/func/test_lockfile.py b/tests/func/test_lockfile.py index c75fde2af..27c84455b 100644 --- a/tests/func/test_lockfile.py +++ b/tests/func/test_lockfile.py @@ -1,10 +1,20 @@ +import logging +import os from collections import OrderedDict from operator import itemgetter -from dvc.dvcfile import PIPELINE_LOCK +import pytest + +from dvc.dvcfile import PIPELINE_LOCK, Lockfile, LockfileCorruptedError +from dvc.hash_info import HashInfo from dvc.stage.utils import split_params_deps from dvc.utils.fs import remove -from dvc.utils.serialize import dumps_yaml, parse_yaml_for_update +from dvc.utils.serialize import ( + dump_yaml, + dumps_yaml, + load_yaml, + parse_yaml_for_update, +) from tests.func.test_run_multistage import supported_params FS_STRUCTURE = { @@ -37,7 +47,7 @@ def test_deps_outs_are_sorted_by_path(tmp_dir, dvc, run_head): run_head(*deps, name="copy-first-line") initial_content = read_lock_file() - lock = initial_content["copy-first-line"] + lock = initial_content["stages"]["copy-first-line"] # lock stage key order: assert list(lock.keys()) == ["cmd", "deps", "outs"] @@ -54,9 +64,9 @@ def test_deps_outs_are_sorted_by_path(tmp_dir, dvc, run_head): assert list(map(itemgetter("path"), lock["deps"])) == sorted(deps) # outs are too - assert list( - map(itemgetter("path"), initial_content["copy-first-line"]["outs"]) - ) == [d + "-1" for d in sorted(deps)] + assert list(map(itemgetter("path"), lock["outs"])) == [ + d + "-1" for d in sorted(deps) + ] def test_order_is_preserved_when_pipeline_order_changes( @@ -98,7 +108,7 @@ def test_cmd_changes_other_orders_are_preserved(tmp_dir, dvc, run_head): stage.cmd = " ".join(stage.cmd.split()) stage.dvcfile._dump_pipeline_file(stage) - initial_content["copy-first-line"]["cmd"] = stage.cmd + initial_content["stages"]["copy-first-line"]["cmd"] = stage.cmd assert dvc.reproduce(stage.addressing) == [stage] @@ -121,7 +131,7 @@ def test_params_dump(tmp_dir, dvc, run_head): ) initial_content = read_lock_file() - lock = initial_content["copy-first-line"] + lock = initial_content["stages"]["copy-first-line"] # lock stage key order: assert list(lock.keys()) == ["cmd", "deps", "params", "outs"] @@ -155,3 +165,64 @@ def test_params_dump(tmp_dir, dvc, run_head): remove(item) assert dvc.reproduce(stage.addressing) == [stage] assert_eq_lockfile(initial_content, read_lock_file()) + + +@pytest.fixture +def v1_repo_lock(tmp_dir, dvc): + """Generates a repo having v1 format lockfile""" + size = 5 if os.name == "nt" else 4 + hi = HashInfo( + name="md5", size=size, value="c157a79031e1c40f85931829bc5fc552" + ) + v1_lockdata = { + "foo": {"cmd": "echo foo"}, + "bar": { + "cmd": "echo bar>bar.txt", + "outs": [{"path": "bar.txt", **hi.to_dict()}], + }, + } + dvc.run(cmd="echo foo", name="foo", no_exec=True) + dvc.run(cmd="echo bar>bar.txt", outs=["bar.txt"], name="bar", no_exec=True) + dump_yaml(tmp_dir / "dvc.lock", v1_lockdata) + yield v1_lockdata + + +def test_can_read_v1_lockfile(tmp_dir, dvc, v1_repo_lock): + assert dvc.status() == { + "bar": [ + {"changed outs": {"bar.txt": "not in cache"}}, + "always changed", + ], + "foo": ["always changed"], + } + + +def test_migrates_v1_lockfile_to_v2_during_dump( + tmp_dir, dvc, v1_repo_lock, caplog +): + caplog.clear() + with caplog.at_level(logging.INFO, logger="dvc.dvcfile"): + assert dvc.reproduce() + + assert "Migrating lock file 'dvc.lock' from v1 to v2" in caplog.messages + d = load_yaml(tmp_dir / "dvc.lock") + assert d == {"stages": v1_repo_lock, "meta": {"version": "2.0"}} + + +@pytest.mark.parametrize( + "version_info", + [{"version": "1.1"}, {"version": "2.1"}, {"version": "3.0"}], +) +def test_lockfile_invalid_versions(tmp_dir, dvc, version_info): + lockdata = {"meta": version_info, "stages": {"foo": {"cmd": "echo foo"}}} + dump_yaml("dvc.lock", lockdata) + with pytest.raises(LockfileCorruptedError) as exc_info: + Lockfile(dvc, tmp_dir / "dvc.lock").load() + + assert str(exc_info.value) == "Lockfile 'dvc.lock' is corrupted." + assert ( + str(exc_info.value.__cause__) == "'dvc.lock' format error: " + f"invalid version {version_info['version']}, " + "expected one of ['2.0'] for dictionary value @ " + "data['meta']['version']" + ) diff --git a/tests/func/test_repro_multistage.py b/tests/func/test_repro_multistage.py index 98c73ac13..8c74a364b 100644 --- a/tests/func/test_repro_multistage.py +++ b/tests/func/test_repro_multistage.py @@ -502,7 +502,7 @@ def test_repro_multiple_params(tmp_dir, dvc): assert len(stage.outs) == 1 lockfile = stage.dvcfile._lockfile - assert lockfile.load()["read_params"]["params"] == { + assert lockfile.load()["stages"]["read_params"]["params"] == { "params2.yaml": { "lists": [42, 42.0, "42"], "floats": 42.0, diff --git a/tests/func/test_run_multistage.py b/tests/func/test_run_multistage.py index 98094b7e1..dc74238bf 100644 --- a/tests/func/test_run_multistage.py +++ b/tests/func/test_run_multistage.py @@ -34,14 +34,10 @@ def test_run_no_exec(tmp_dir, dvc, run_copy): assert not os.path.exists(PIPELINE_LOCK) data, _ = stage.dvcfile._load() - assert data == { - "stages": { - "copy-foo-to-bar": { - "cmd": "python copy.py foo bar", - "deps": ["copy.py", "foo"], - "outs": ["bar"], - } - } + assert data["stages"]["copy-foo-to-bar"] == { + "cmd": "python copy.py foo bar", + "deps": ["copy.py", "foo"], + "outs": ["bar"], } @@ -172,7 +168,7 @@ def test_run_dump_on_multistage(tmp_dir, dvc, run_head): "always_changed": True, "wdir": "dir", }, - } + }, } run_head( @@ -197,7 +193,7 @@ def test_run_dump_on_multistage(tmp_dir, dvc, run_head): "metrics": [{"foobar-1": {"cache": False}}], }, **data["stages"], - } + }, } @@ -244,7 +240,7 @@ def test_run_params_default(tmp_dir, dvc): assert stage.deps[0].params == ["nested.nested1.nested2"] lockfile = stage.dvcfile._lockfile - assert lockfile.load()["read_params"]["params"] == { + assert lockfile.load()["stages"]["read_params"]["params"] == { "params.yaml": {"nested.nested1.nested2": "42"} } @@ -267,7 +263,7 @@ def test_run_params_custom_file(tmp_dir, dvc): isinstance(stage.deps[0], ParamsDependency) assert stage.deps[0].params == ["lists"] lockfile = stage.dvcfile._lockfile - assert lockfile.load()["read_params"]["params"] == { + assert lockfile.load()["stages"]["read_params"]["params"] == { "params2.yaml": {"lists": [42, 42.0, "42"]} } @@ -424,16 +420,19 @@ def test_run_external_outputs( assert (tmp_dir / "dvc.yaml").read_text() == dvc_yaml assert (tmp_dir / "dvc.lock").read_text() == ( - "mystage:\n" - " cmd: mycmd\n" - " deps:\n" - " - path: remote://workspace/foo\n" - f" {hash_name}: {foo_hash}\n" - " size: 3\n" - " outs:\n" - " - path: remote://workspace/bar\n" - f" {hash_name}: {bar_hash}\n" - " size: 3\n" + "meta:\n" + " version: '2.0'\n" + "stages:\n" + " mystage:\n" + " cmd: mycmd\n" + " deps:\n" + " - path: remote://workspace/foo\n" + f" {hash_name}: {foo_hash}\n" + " size: 3\n" + " outs:\n" + " - path: remote://workspace/bar\n" + f" {hash_name}: {bar_hash}\n" + " size: 3\n" ) assert (workspace / "foo").read_text() == "foo" diff --git a/tests/unit/stage/test_serialize_pipeline_lock.py b/tests/unit/stage/test_serialize_pipeline_lock.py index 85f8c67bd..18d14a9e7 100644 --- a/tests/unit/stage/test_serialize_pipeline_lock.py +++ b/tests/unit/stage/test_serialize_pipeline_lock.py @@ -5,7 +5,7 @@ from voluptuous import Schema as _Schema from dvc.dvcfile import PIPELINE_FILE from dvc.hash_info import HashInfo -from dvc.schema import LOCK_FILE_STAGE_SCHEMA, LOCKFILE_SCHEMA +from dvc.schema import LOCK_FILE_STAGE_SCHEMA, LOCKFILE_STAGES_SCHEMA from dvc.stage import PipelineStage, create_stage from dvc.stage.serialize import DEFAULT_PARAMS_FILE, to_lockfile from dvc.stage.serialize import ( @@ -227,7 +227,7 @@ def test_to_lockfile(dvc): stage.deps[0].hash_info = HashInfo("md5", "md-five") entry = to_lockfile(stage) assert len(entry) == 1 - _Schema(LOCKFILE_SCHEMA)(entry) + _Schema(LOCKFILE_STAGES_SCHEMA)(entry) assert entry == { "something": OrderedDict( [ diff --git a/tests/unit/test_lockfile.py b/tests/unit/test_lockfile.py index a5a8c867f..662fb4c3a 100644 --- a/tests/unit/test_lockfile.py +++ b/tests/unit/test_lockfile.py @@ -9,7 +9,10 @@ def test_stage_dump_no_outs_deps(tmp_dir, dvc): stage = PipelineStage(name="s1", repo=dvc, path="path", cmd="command") lockfile = Lockfile(dvc, "path.lock") lockfile.dump(stage) - assert lockfile.load() == {"s1": {"cmd": "command"}} + assert lockfile.load() == { + "stages": {"s1": {"cmd": "command"}}, + "meta": {"version": "2.0"}, + } def test_stage_dump_when_already_exists(tmp_dir, dvc): @@ -19,8 +22,8 @@ def test_stage_dump_when_already_exists(tmp_dir, dvc): lockfile = Lockfile(dvc, "path.lock") lockfile.dump(stage) assert lockfile.load() == { - **data, - "s2": {"cmd": "command2"}, + "stages": {**data, "s2": {"cmd": "command2"}}, + "meta": {"version": "2.0"}, } @@ -37,8 +40,8 @@ def test_stage_dump_with_deps_and_outs(tmp_dir, dvc): stage = PipelineStage(name="s2", repo=dvc, path="path", cmd="command2") lockfile.dump(stage) assert lockfile.load() == { - **data, - "s2": {"cmd": "command2"}, + "stages": {**data, "s2": {"cmd": "command2"}}, + "meta": {"version": "2.0"}, } @@ -49,7 +52,8 @@ def test_stage_overwrites_if_already_exists(tmp_dir, dvc): stage = PipelineStage(name="s2", repo=dvc, path="path", cmd="command3") lockfile.dump(stage) assert lockfile.load() == { - "s2": {"cmd": "command3"}, + "stages": {"s2": {"cmd": "command3"}}, + "meta": {"version": "2.0"}, }