better exception messages for schema errors (#6679)

* Generate prettier schema error messages

* add test for #6285

dvc/dvcfile.py

@@ -1,7 +1,7 @@
 import contextlib
 import logging
 import os
-from typing import TYPE_CHECKING, Any, Callable, Tuple, Type, TypeVar, Union
+from typing import TYPE_CHECKING, Any, Callable, Tuple, TypeVar, Union
 
 from dvc.exceptions import DvcException
 from dvc.parsing.versions import LOCKFILE_VERSION, SCHEMA_KWD
@@ -9,14 +9,12 @@ from dvc.stage import serialize
 from dvc.stage.exceptions import (
     StageFileBadNameError,
     StageFileDoesNotExistError,
-    StageFileFormatError,
     StageFileIsNotDvcFileError,
 )
 from dvc.types import AnyPath
 from dvc.utils import relpath
 from dvc.utils.collections import apply_diff
 from dvc.utils.serialize import dump_yaml, modify_yaml
-from dvc.utils.strictyaml import YAMLValidationError
 
 if TYPE_CHECKING:
     from dvc.repo import Repo
@@ -39,11 +37,6 @@ class FileIsGitIgnored(DvcException):
         )
 
 
-class LockfileCorruptedError(DvcException):
-    def __init__(self, path):
-        super().__init__(f"Lockfile '{path}' is corrupted.")
-
-
 class ParametrizedDumpError(DvcException):
     pass
 
@@ -92,7 +85,6 @@ def check_dvcfile_path(repo, path):
 
 class FileMixin:
     SCHEMA: Callable[[_T], _T]
-    ValidationError: Type[DvcException] = StageFileFormatError
 
     def __init__(self, repo, path, verify=True, **kwargs):
         self.repo = repo
@@ -134,7 +126,11 @@ class FileMixin:
         if self._is_git_ignored():
             raise FileIsGitIgnored(self.path)
 
-    def _load(self) -> Tuple[Any, str]:
+    def load(self, **kwargs: Any) -> Any:
+        d, _ = self._load(**kwargs)
+        return d
+
+    def _load(self, **kwargs: Any) -> Tuple[Any, str]:
         # it raises the proper exceptions by priority:
         # 1. when the file doesn't exists
         # 2. filename is not a DVC file
@@ -151,29 +147,23 @@ class FileMixin:
             raise StageFileIsNotDvcFileError(self.path)
 
         self._check_gitignored()
-        return self._load_yaml()
+        return self._load_yaml(**kwargs)
 
     @classmethod
     def validate(cls, d: _T, fname: str = None) -> _T:
         from dvc.utils.strictyaml import validate
 
-        try:
-            return validate(d, cls.SCHEMA)  # type: ignore[arg-type]
-        except YAMLValidationError as exc:
-            raise cls.ValidationError(fname) from exc
+        return validate(d, cls.SCHEMA, path=fname)  # type: ignore[arg-type]
 
     def _load_yaml(self, **kwargs: Any) -> Tuple[Any, str]:
         from dvc.utils import strictyaml
 
-        try:
-            return strictyaml.load(
-                self.path,
-                self.SCHEMA,  # type: ignore[arg-type]
-                self.repo.fs,
-                **kwargs,
-            )
-        except YAMLValidationError as exc:
-            raise self.ValidationError(self.relpath) from exc
+        return strictyaml.load(
+            self.path,
+            self.SCHEMA,  # type: ignore[arg-type]
+            self.repo.fs,
+            **kwargs,
+        )
 
     def remove(self, force=False):  # pylint: disable=unused-argument
         with contextlib.suppress(FileNotFoundError):
@@ -354,24 +344,19 @@ def lockfile_schema(data: _T) -> _T:
 
 class Lockfile(FileMixin):
     SCHEMA = staticmethod(lockfile_schema)  # type: ignore[assignment]
-    ValidationError = LockfileCorruptedError
 
     def _verify_filename(self):
         pass  # lockfile path is hardcoded, so no need to verify here
 
-    def _load(self):
+    def _load(self, **kwargs: Any):
         try:
-            return super()._load()
+            return super()._load(**kwargs)
         except StageFileDoesNotExistError:
             # we still need to account for git-ignored dvc.lock file
             # even though it may not exist or have been .dvcignored
             self._check_gitignored()
             return {}, ""
 
-    def load(self):
-        d, _ = self._load()
-        return d
-
     @property
     def latest_version_info(self):
         version = LOCKFILE_VERSION.V2.value  # pylint:disable=no-member

dvc/schema.py

@@ -1,3 +1,5 @@
+from collections.abc import Mapping
+
 from voluptuous import Any, Optional, Required, Schema
 
 from dvc import dependency, output
@@ -94,11 +96,27 @@ STAGE_DEFINITION = {
     Optional(StageParams.PARAM_LIVE): Any(str, LIVE_PSTAGE_SCHEMA),
 }
 
+
+def either_or(primary, fallback, fallback_includes=None):
+    def validator(data):
+        schema = primary
+        if (
+            isinstance(data, Mapping)
+            and set(fallback_includes or []) & data.keys()
+        ):
+            schema = fallback
+        return Schema(schema)(data)
+
+    return validator
+
+
 FOREACH_IN = {
     Required(FOREACH_KWD): Any(dict, list, str),
     Required(DO_KWD): STAGE_DEFINITION,
 }
-SINGLE_PIPELINE_STAGE_SCHEMA = {str: Any(FOREACH_IN, STAGE_DEFINITION)}
+SINGLE_PIPELINE_STAGE_SCHEMA = {
+    str: either_or(STAGE_DEFINITION, FOREACH_IN, [FOREACH_KWD, DO_KWD])
+}
 MULTI_STAGE_SCHEMA = {
     STAGES: SINGLE_PIPELINE_STAGE_SCHEMA,
     VARS_KWD: VARS_SCHEMA,

dvc/stage/exceptions.py

@@ -9,11 +9,6 @@ class StageCmdFailedError(DvcException):
         super().__init__(msg)
 
 
-class StageFileFormatError(DvcException):
-    def __init__(self, path):
-        super().__init__(f"'{path}' format error")
-
-
 class StageFileDoesNotExistError(DvcException):
     DVC_IGNORED = "is dvc-ignored"
     DOES_NOT_EXIST = "does not exist"

dvc/utils/strictyaml.py

@@ -6,42 +6,67 @@ Used for parsing dvc.yaml, dvc.lock and .dvc files.
 
 Not to be confused with strictyaml, a python library with similar motivations.
 """
-from typing import TYPE_CHECKING, Any, Callable, List, TypeVar
+import re
+import typing
+from contextlib import suppress
+from typing import TYPE_CHECKING, Any, Callable, List, Optional, Tuple, TypeVar
 
-from dvc.exceptions import DvcException, PrettyDvcException
+from dvc.exceptions import PrettyDvcException
 from dvc.ui import ui
 from dvc.utils.serialize import (
     EncodingError,
     YAMLFileCorruptedError,
     parse_yaml,
+    parse_yaml_for_update,
 )
 
 if TYPE_CHECKING:
     from rich.syntax import Syntax
     from ruamel.yaml import StreamMark
+    from voluptuous import MultipleInvalid
 
     from dvc.fs.base import BaseFileSystem
     from dvc.ui import RichText
 
 
 _T = TypeVar("_T")
+merge_conflict_marker = re.compile("^([<=>]{7}) .*$", re.MULTILINE)
+
+
+def make_relpath(path: str) -> str:
+    import os
+
+    from dvc.utils import relpath
+
+    rel = relpath(path)
+    prefix = ""
+    if not rel.startswith(".."):
+        prefix = "./" if os.name == "posix" else ".\\"
+    return prefix + rel
+
+
+def _prepare_message(message: str) -> "RichText":
+    return ui.rich_text(message, style="red")
 
 
 def _prepare_cause(cause: str) -> "RichText":
     return ui.rich_text(cause, style="bold")
 
 
-def _prepare_code_snippets(code: str, start_line: int = 1) -> "Syntax":
+def _prepare_code_snippets(
+    code: str, start_line: int = 1, **kwargs: Any
+) -> "Syntax":
     from rich.syntax import Syntax
 
+    kwargs.setdefault("start_line", start_line)
     return Syntax(
         code,
         "yaml",
-        start_line=start_line,
         theme="ansi_dark",
         word_wrap=True,
         line_numbers=True,
         indent_guides=True,
+        **kwargs,
     )
 
 
@@ -50,13 +75,14 @@ class YAMLSyntaxError(PrettyDvcException, YAMLFileCorruptedError):
         self.path: str = path
         self.yaml_text: str = yaml_text
         self.exc: Exception = exc
+
+        merge_conflicts = merge_conflict_marker.search(self.yaml_text)
+        self.hint = " (possible merge conflicts)" if merge_conflicts else ""
         super().__init__(self.path)
 
     def __pretty_exc__(self, **kwargs: Any) -> None:
         from ruamel.yaml.error import MarkedYAMLError
 
-        from dvc.utils import relpath
-
         exc = self.exc.__cause__
 
         if not isinstance(exc, MarkedYAMLError):
@@ -101,25 +127,140 @@ class YAMLSyntaxError(PrettyDvcException, YAMLFileCorruptedError):
             if exc.problem_mark is not None:
                 lines.append(prepare_code(exc.problem_mark))
 
-        if lines and lines[-1]:
+        if lines:
+            # we should not add a newline after the main message
+            # if there are no other outputs
             lines.insert(0, "")
-        lines.insert(0, f"[red]'{relpath(self.path)}' structure is corrupted.")
-        for message in lines:
-            ui.error_write(message, styled=True)
+
+        rel = make_relpath(self.path)
+        lines.insert(0, _prepare_message(f"'{rel}' is invalid{self.hint}."))
+        for line in lines:
+            ui.error_write(line, styled=True)
 
 
-class YAMLValidationError(DvcException):
-    def __init__(self, exc):
-        super().__init__(str(exc))
+def determine_linecol(
+    data, paths, max_steps=5
+) -> Tuple[Optional[int], Optional[int], int]:
+    """Determine linecol from the CommentedMap for the `paths` location.
+
+    CommentedMap from `ruamel.yaml` has `.lc` property from which we can read
+    `.line` and `.col`. This is available in the collections type,
+    i.e. list and dictionaries.
+
+    But this may fail on non-collection types. For example, if the `paths` is
+    ['stages', 'metrics'], metrics being a boolean type does not have `lc`
+    prop.
+    ```
+    stages:
+      metrics: true
+    ```
+
+    To provide some context to the user, we step up to the
+    path ['stages'], which being a collection type, will have `lc` prop
+    with which we can find line and col.
+
+    This may end up being not accurate, so we try to show the same amount of
+    lines of code for `n` number of steps taken upwards. In a worst case,
+    it may be just 1 step (as non-collection item cannot have child items),
+    but `schema validator` may provide us arbitrary path. So, this caps the
+    number of steps upward to just 5. If it does not find any linecols, it'll
+    abort.
+    """
+    from dpath.util import get
+
+    step = 1
+    line, col = None, None
+    while paths and step < max_steps:
+        value = get(data, paths, default=None)
+        if value is not None:
+            with suppress(AttributeError, TypeError):
+                line = value.lc.line + 1
+                col = value.lc.col + 1
+                break
+        step += 1
+        *paths, _ = paths
+
+    return line, col, step
 
 
-def validate(data: _T, schema: Callable[[_T], _T], _text: str = None) -> _T:
+class YAMLValidationError(PrettyDvcException):
+    def __init__(
+        self,
+        exc: "MultipleInvalid",
+        path: str = None,
+        text: str = None,
+    ) -> None:
+        self.text = text or ""
+        self.exc = exc
+
+        rel = make_relpath(path) if path else ""
+        self.path = path or ""
+
+        message = f"'{rel}' validation failed"
+        if len(self.exc.errors) > 1:
+            message += f": {len(self.exc.errors)} errors"
+        super().__init__(f"{message}")
+
+    def _prepare_context(self, data: typing.Mapping) -> List[object]:
+        lines: List[object] = []
+        for index, error in enumerate(self.exc.errors):
+            if index and lines[-1]:
+                lines.append("")
+            line, col, step = determine_linecol(data, error.path)
+            parts = [error.error_message]
+            if error.path:
+                parts.append("in " + " -> ".join(str(p) for p in error.path))
+            if line:
+                parts.append(f"line {line}")
+            if col:
+                parts.append(f"column {col}")
+            lines.append(_prepare_cause(", ".join(parts)))
+
+            if line:
+                # we show one line above the error
+                # we try to show few more lines if we could not
+                # reliably figure out where the error was
+                lr = (line - 1, line + step - 1)
+                code = _prepare_code_snippets(self.text, line_range=lr)
+                lines.append(code)
+        return lines
+
+    def __pretty_exc__(self, **kwargs: Any) -> None:
+        """Prettify exception message."""
+        from collections.abc import Mapping
+
+        lines: List[object] = []
+        data = parse_yaml_for_update(self.text, self.path)
+        if isinstance(data, Mapping):
+            lines.extend(self._prepare_context(data))
+
+        cause = ""
+        if lines:
+            # we should not add a newline after the main message
+            # if there are no other outputs
+            lines.insert(0, "")
+        else:
+            # if we don't have any context to show, we'll fallback to what we
+            # got from voluptuous and print them in the same line.
+            cause = f": {self.exc}"
+
+        lines.insert(0, _prepare_message(f"{self}{cause}."))
+        for line in lines:
+            ui.error_write(line, styled=True)
+
+
+def validate(
+    data: _T,
+    schema: Callable[[_T], _T],
+    text: str = None,
+    path: str = None,
+) -> _T:
     from voluptuous import MultipleInvalid
 
     try:
         return schema(data)
     except MultipleInvalid as exc:
-        raise YAMLValidationError(str(exc))
+        raise YAMLValidationError(exc, path, text) from exc
 
 
 def load(
@@ -143,5 +284,5 @@ def load(
     if schema:
         # not returning validated data, as it may remove
         # details from CommentedMap that we get from roundtrip parser
-        validate(data, schema, text)
+        validate(data, schema, text=text, path=path)
     return data, text

tests/func/test_dvcfile.py

@@ -7,15 +7,12 @@ from dvc.dvcfile import (
     PIPELINE_FILE,
     PIPELINE_LOCK,
     Dvcfile,
-    LockfileCorruptedError,
     ParametrizedDumpError,
     SingleStageFile,
 )
-from dvc.stage.exceptions import (
-    StageFileDoesNotExistError,
-    StageFileFormatError,
-)
+from dvc.stage.exceptions import StageFileDoesNotExistError
 from dvc.stage.loader import StageNotFound
+from dvc.utils.strictyaml import YAMLValidationError
 
 
 def test_run_load_one_for_multistage(tmp_dir, dvc):
@@ -246,14 +243,14 @@ def test_remove_stage_on_lockfile_format_error(tmp_dir, dvc, run_copy):
     lock_data["gibberish"] = True
     data["gibberish"] = True
     (tmp_dir / lock_file.relpath).dump(lock_data)
-    with pytest.raises(LockfileCorruptedError):
+    with pytest.raises(YAMLValidationError):
         dvc_file.remove_stage(stage)
 
     lock_file.remove()
     dvc_file.dump(stage, update_pipeline=False)
 
     (tmp_dir / dvc_file.relpath).dump(data)
-    with pytest.raises(StageFileFormatError):
+    with pytest.raises(YAMLValidationError):
         dvc_file.remove_stage(stage)
 
 

tests/func/test_lockfile.py

@@ -5,11 +5,12 @@ from operator import itemgetter
 
 import pytest
 
-from dvc.dvcfile import PIPELINE_LOCK, Lockfile, LockfileCorruptedError
+from dvc.dvcfile import PIPELINE_LOCK, Lockfile
 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.strictyaml import YAMLValidationError, make_relpath
 from tests.func.test_run_multistage import supported_params
 
 FS_STRUCTURE = {
@@ -207,10 +208,11 @@ def test_migrates_v1_lockfile_to_v2_during_dump(
 def test_lockfile_invalid_versions(tmp_dir, dvc, version_info):
     lockdata = {**version_info, "stages": {"foo": {"cmd": "echo foo"}}}
     (tmp_dir / "dvc.lock").dump(lockdata)
-    with pytest.raises(LockfileCorruptedError) as exc_info:
+    with pytest.raises(YAMLValidationError) as exc_info:
         Lockfile(dvc, tmp_dir / "dvc.lock").load()
 
-    assert str(exc_info.value) == "Lockfile 'dvc.lock' is corrupted."
+    rel = make_relpath("dvc.lock")
+    assert f"'{rel}' validation failed" in str(exc_info.value)
     assert (
         str(exc_info.value.__cause__)
         == f"invalid schema version {version_info['schema']}, "

tests/func/test_stage.py

@@ -9,14 +9,14 @@ from dvc.main import main
 from dvc.output import Output
 from dvc.repo import Repo, lock_repo
 from dvc.stage import PipelineStage, Stage
-from dvc.stage.exceptions import StageFileFormatError
 from dvc.stage.run import run_stage
 from dvc.utils.serialize import dump_yaml, load_yaml
+from dvc.utils.strictyaml import YAMLValidationError
 from tests.basic_env import TestDvc
 
 
 def test_cmd_obj():
-    with pytest.raises(StageFileFormatError):
+    with pytest.raises(YAMLValidationError):
         SingleStageFile.validate({Stage.PARAM_CMD: {}})
 
 
@@ -33,10 +33,10 @@ def test_cmd_str():
 
 
 def test_object():
-    with pytest.raises(StageFileFormatError):
+    with pytest.raises(YAMLValidationError):
         SingleStageFile.validate({Stage.PARAM_DEPS: {}})
 
-    with pytest.raises(StageFileFormatError):
+    with pytest.raises(YAMLValidationError):
         SingleStageFile.validate({Stage.PARAM_OUTS: {}})
 
 

tests/func/test_stage_load.py

@@ -10,12 +10,12 @@ from dvc.path_info import PathInfo
 from dvc.repo import Repo
 from dvc.stage.exceptions import (
     StageFileDoesNotExistError,
-    StageFileFormatError,
     StageNameUnspecified,
     StageNotFound,
 )
 from dvc.utils import relpath
 from dvc.utils.fs import remove
+from dvc.utils.strictyaml import YAMLValidationError
 
 
 def test_collect(tmp_dir, scm, dvc, run_copy):
@@ -450,7 +450,7 @@ def test_collect_repo_callback(tmp_dir, dvc, mocker):
 
     file_path, exc = mock.call_args[0]
     assert file_path == PIPELINE_FILE
-    assert isinstance(exc, StageFileFormatError)
+    assert isinstance(exc, YAMLValidationError)
 
 
 def test_gitignored_collect_repo(tmp_dir, dvc, scm):

tests/func/utils/test_strict_yaml.py

@@ -0,0 +1,396 @@
+import os
+
+import pytest
+from ruamel.yaml import __with_libyaml__ as ruamel_clib
+
+from dvc.main import main
+from dvc.ui import ui
+
+DUPLICATE_KEYS = """\
+stages:
+  stage1:
+    cmd: python train.py
+    cmd: python train.py
+"""
+
+DUPLICATE_KEYS_OUTPUT = """\
+'./dvc.yaml' is invalid.
+
+While constructing a mapping, in line 3, column 5
+  3 │   cmd: python train.py
+
+Found duplicate key "cmd" with value "python train.py" (original value:\
+ "python \ntrain.py"), in line 4, column 5
+  4 │   cmd: python train.py"""
+
+
+MAPPING_VALUES_NOT_ALLOWED = """\
+stages:
+  stage1
+    cmd: python script.py
+"""
+
+MAPPING_VALUES_NOT_ALLOWED_OUTPUT = """\
+'./dvc.yaml' is invalid.
+
+Mapping values are not allowed {}, in line 3, column 8
+  3 │   cmd: python script.py""".format(
+    "in this context" if ruamel_clib else "here"
+)
+
+
+NO_HYPHEN_INDICATOR_IN_BLOCK = """\
+stages:
+  stage1:
+    cmd: python script.py
+    outs:
+      - logs:
+          cache: false
+      metrics:
+"""
+
+NO_HYPHEN_INDICATOR_IN_BLOCK_OUTPUT = """\
+'./dvc.yaml' is invalid.
+
+While parsing a block collection, in line 5, column 7
+  5 │     - logs:
+
+{}, in line 7, column 7
+  7 │     metrics:""".format(
+    "Did not find expected '-' indicator"
+    if ruamel_clib
+    else "Expected <block end>, but found '?'"
+)
+
+
+UNCLOSED_SCALAR = """\
+stages:
+  stage1:
+    cmd: python script.py
+    desc: "this is my stage one
+"""
+
+UNCLOSED_SCALAR_OUTPUT = """\
+'./dvc.yaml' is invalid.
+
+While scanning a quoted scalar, in line 4, column 11
+  4 │   desc: "this is my stage one
+
+Found unexpected end of stream, in line 5, column 1
+  5"""
+
+
+NOT_A_DICT = "3"
+NOT_A_DICT_OUTPUT = "'./dvc.yaml' validation failed: expected a dictionary.\n"
+
+
+EMPTY_STAGE = """\
+stages:
+  stage1:
+"""
+
+EMPTY_STAGE_OUTPUT = """\
+'./dvc.yaml' validation failed.
+
+expected a dictionary, in stages -> stage1, line 2, column 3
+  1 stages:
+  2   stage1:
+  3"""
+
+
+MISSING_CMD = """\
+stages:
+  stage1:
+    cmd: {}
+"""
+
+MISSING_CMD_OUTPUT = """\
+'./dvc.yaml' validation failed.
+
+expected str, in stages -> stage1 -> cmd, line 3, column 10
+  2   stage1:
+  3 │   cmd: {}"""
+
+
+DEPS_AS_DICT = """\
+stages:
+  stage1:
+    cmd: python script.py
+    deps:
+      - src:
+"""
+
+DEPS_AS_DICT_OUTPUT = """\
+'./dvc.yaml' validation failed.
+
+expected str, in stages -> stage1 -> deps -> 0, line 5, column 9
+  4 │   deps:
+  5 │     - src:
+"""
+
+OUTS_AS_STR = """\
+stages:
+  train:
+    cmd:
+      - python train.py
+    deps:
+      - config.cfg
+    outs:
+      models/"""
+
+OUTS_AS_STR_OUTPUT = """\
+'./dvc.yaml' validation failed.
+
+expected a list, in stages -> train -> outs, line 3, column 5
+  2   train:
+  3 │   cmd:
+  4 │     - python train.py"""
+
+
+NULL_VALUE_ON_OUTS = """\
+stages:
+  stage1:
+    cmd: python script.py
+    outs:
+    - logs:
+        cache: false
+        persist: true
+        remote:
+"""
+
+NULL_VALUE_ON_OUTS_OUTPUT = """\
+'./dvc.yaml' validation failed.
+
+expected str, in stages -> stage1 -> outs -> 0 -> logs -> remote, line 6,\
+ column\n9
+  5 │   - logs:
+  6 │   │   cache: false
+  7 │   │   persist: true"""
+
+ADDITIONAL_KEY_ON_OUTS = """\
+stages:
+  stage1:
+    cmd: python script.py
+    outs:
+    - logs:
+        cache: false
+        not_existing_key: false
+"""
+
+ADDITIONAL_KEY_ON_OUTS_OUTPUT = """\
+'./dvc.yaml' validation failed.
+
+extra keys not allowed, in stages -> stage1 -> outs -> 0 -> logs -> \n\
+not_existing_key, line 6, column 9
+  5 │   - logs:
+  6 │   │   cache: false
+  7 │   │   not_existing_key: false"""
+
+
+FOREACH_SCALAR_VALUE = """\
+stages:
+  group:
+    foreach: 3
+    do:
+      cmd: python script${i}.py
+"""
+
+FOREACH_SCALAR_VALUE_OUTPUT = """\
+'./dvc.yaml' validation failed.
+
+expected dict, in stages -> group -> foreach, line 3, column 5
+  2   group:
+  3 │   foreach: 3
+  4 │   do:"""
+
+FOREACH_DO_NULL = """\
+stages:
+  stage1:
+    foreach: [1,2,3]
+    do:
+"""
+
+
+FOREACH_DO_NULL_OUTPUT = """\
+'./dvc.yaml' validation failed.
+
+expected a dictionary, in stages -> stage1 -> do, line 3, column 5
+  2   stage1:
+  3 │   foreach: [1,2,3]
+  4 │   do:"""
+
+
+FOREACH_DO_MISSING_CMD = """\
+stages:
+  stage1:
+    foreach: [1,2,3]
+    do:
+      outs:
+      - ${item}
+"""
+
+
+FOREACH_WITH_CMD_DO_MISSING = """\
+stages:
+  stage1:
+    foreach: [1,2,3]
+    cmd: python script${item}.py
+"""
+
+
+FOREACH_WITH_CMD_DO_MISSING_OUTPUT = """\
+'./dvc.yaml' validation failed: 2 errors.
+
+extra keys not allowed, in stages -> stage1 -> cmd, line 3, column 5
+  2   stage1:
+  3 │   foreach: [1,2,3]
+  4 │   cmd: python script${item}.py
+
+required key not provided, in stages -> stage1 -> do, line 3, column 5
+  2   stage1:
+  3 │   foreach: [1,2,3]
+  4 │   cmd: python script${item}.py"""
+
+
+FOREACH_DO_MISSING_CMD_OUTPUT = """\
+'./dvc.yaml' validation failed.
+
+required key not provided, in stages -> stage1 -> do -> cmd, line 5, column 7
+  4 │   do:
+  5 │     outs:
+  6 │     - ${item}"""
+
+
+MERGE_CONFLICTS = """\
+stages:
+  load_data:
+<<<<<<< HEAD
+    cmd: python src/load_data.py
+    deps:
+    - src/load_data.py
+=======
+    cmd: python load_data.py
+    deps:
+    - load_data.py
+>>>>>>> branch
+    outs:
+    - data
+"""
+
+MERGE_CONFLICTS_OUTPUT = """\
+'./dvc.yaml' is invalid (possible merge conflicts).
+
+While scanning a simple key, in line 3, column 1
+  3 <<<<<<< HEAD
+
+Could not find expected ':', in line 4, column 8
+  4 │   cmd: python src/load_data.py"""
+
+
+examples = {
+    # on parse errors
+    "duplicate_keys": (DUPLICATE_KEYS, DUPLICATE_KEYS_OUTPUT),
+    "mapping_values_not_allowed": (
+        MAPPING_VALUES_NOT_ALLOWED,
+        MAPPING_VALUES_NOT_ALLOWED_OUTPUT,
+    ),
+    "no_hyphen_block": (
+        NO_HYPHEN_INDICATOR_IN_BLOCK,
+        NO_HYPHEN_INDICATOR_IN_BLOCK_OUTPUT,
+    ),
+    "unclosed_scalar": (UNCLOSED_SCALAR, UNCLOSED_SCALAR_OUTPUT),
+    # schema validation errors
+    "not_a_dict": (NOT_A_DICT, NOT_A_DICT_OUTPUT),
+    "empty_stage": (EMPTY_STAGE, EMPTY_STAGE_OUTPUT),
+    "missing_cmd": (MISSING_CMD, MISSING_CMD_OUTPUT),
+    "deps_as_dict": (DEPS_AS_DICT, DEPS_AS_DICT_OUTPUT),
+    "outs_as_str": (OUTS_AS_STR, OUTS_AS_STR_OUTPUT),
+    "null_value_on_outs": (NULL_VALUE_ON_OUTS, NULL_VALUE_ON_OUTS_OUTPUT),
+    "additional_key_on_outs": (
+        ADDITIONAL_KEY_ON_OUTS,
+        ADDITIONAL_KEY_ON_OUTS_OUTPUT,
+    ),
+    "foreach_scalar": (FOREACH_SCALAR_VALUE, FOREACH_SCALAR_VALUE_OUTPUT),
+    "foreach_do_do_null": (FOREACH_DO_NULL, FOREACH_DO_NULL_OUTPUT),
+    "foreach_do_missing_cmd": (
+        FOREACH_DO_MISSING_CMD,
+        FOREACH_DO_MISSING_CMD_OUTPUT,
+    ),
+    "foreach_unknown_cmd_missing_do": (
+        FOREACH_WITH_CMD_DO_MISSING,
+        FOREACH_WITH_CMD_DO_MISSING_OUTPUT,
+    ),
+    # merge conflicts
+    "merge_conflicts": (MERGE_CONFLICTS, MERGE_CONFLICTS_OUTPUT),
+}
+
+
+@pytest.mark.parametrize(
+    "text, expected", examples.values(), ids=examples.keys()
+)
+def test_exceptions(tmp_dir, dvc, capsys, text, expected, mocker):
+    # make it always return posix path, easier for validating error messages
+    mocker.patch(
+        "dvc.utils.strictyaml.make_relpath", return_value="./dvc.yaml"
+    )
+
+    console = ui.error_console
+    original_printer = console.print
+
+    def print_with_fixed_width(*args, **kwargs):
+        console.options.min_width = console.options.max_width = 80
+        console.width = kwargs["width"] = 80
+        return original_printer(*args, **kwargs)
+
+    mocker.patch.object(console, "print", print_with_fixed_width)
+
+    tmp_dir.gen("dvc.yaml", text)
+
+    capsys.readouterr()  # clear outputs
+    assert main(["stage", "list"]) != 0
+    out, err = capsys.readouterr()
+
+    assert not out
+    assert expected in err
+
+
+def test_make_relpath(tmp_dir, monkeypatch):
+    from dvc.utils.strictyaml import make_relpath
+
+    path = tmp_dir / "dvc.yaml"
+    expected_path = "./dvc.yaml" if os.name == "posix" else ".\\dvc.yaml"
+    assert make_relpath(path) == expected_path
+
+    (tmp_dir / "dir").mkdir(exist_ok=True)
+    monkeypatch.chdir("dir")
+
+    expected_path = "../dvc.yaml" if os.name == "posix" else "..\\dvc.yaml"
+    assert make_relpath(path) == expected_path
+
+
+def test_fallback_exception_message(tmp_dir, dvc, mocker, caplog):
+    # When trying to pretty print exception messages, we fallback to old way
+    # of printing things.
+    mocker.patch(
+        "dvc.utils.strictyaml.YAMLSyntaxError.__pretty_exc__",
+        side_effect=ValueError,
+    )
+    mocker.patch(
+        "dvc.utils.strictyaml.YAMLValidationError.__pretty_exc__",
+        side_effect=ValueError,
+    )
+
+    # syntax errors
+    dvc_file = tmp_dir / "dvc.yaml"
+    dvc_file.write_text(MAPPING_VALUES_NOT_ALLOWED)
+    assert main(["stage", "list"]) != 0
+    assert (
+        "unable to read: 'dvc.yaml', "
+        "YAML file structure is corrupted" in caplog.text
+    )
+
+    caplog.clear()
+    # validation error
+    dvc_file.dump({"stages": {"stage1": None}})
+    assert main(["stage", "list"]) != 0
+    assert "dvc.yaml' validation failed" in caplog.text

tests/unit/test_dvcfile.py

@@ -11,11 +11,11 @@ from dvc.dvcfile import (
 from dvc.stage import PipelineStage
 from dvc.stage.exceptions import (
     StageFileDoesNotExistError,
-    StageFileFormatError,
     StageFileIsNotDvcFileError,
 )
 from dvc.utils.fs import remove
 from dvc.utils.serialize import EncodingError
+from dvc.utils.strictyaml import YAMLValidationError
 
 
 @pytest.mark.parametrize(
@@ -68,9 +68,9 @@ def test_stage_load_on_invalid_data(tmp_dir, dvc, file):
     data = {"is_this_a_valid_dvcfile": False}
     (tmp_dir / file).dump(data)
     dvcfile = Dvcfile(dvc, file)
-    with pytest.raises(StageFileFormatError):
+    with pytest.raises(YAMLValidationError):
         assert dvcfile.stages
-    with pytest.raises(StageFileFormatError):
+    with pytest.raises(YAMLValidationError):
         assert dvcfile.validate(data, file)
 
 

tests/unit/test_lockfile.py

@@ -1,7 +1,8 @@
 import pytest
 
-from dvc.dvcfile import FileIsGitIgnored, Lockfile, LockfileCorruptedError
+from dvc.dvcfile import FileIsGitIgnored, Lockfile
 from dvc.stage import PipelineStage
+from dvc.utils.strictyaml import YAMLValidationError
 
 
 def test_stage_dump_no_outs_deps(tmp_dir, dvc):
@@ -79,7 +80,7 @@ def test_load_when_lockfile_does_not_exist(tmp_dir, dvc):
 def test_load_when_lockfile_is_corrupted(tmp_dir, dvc, corrupt_data):
     (tmp_dir / "Dvcfile.lock").dump(corrupt_data)
     lockfile = Lockfile(dvc, "Dvcfile.lock")
-    with pytest.raises(LockfileCorruptedError) as exc_info:
+    with pytest.raises(YAMLValidationError) as exc_info:
         lockfile.load()
     assert "Dvcfile.lock" in str(exc_info.value)