diff --git a/dvc/utils/serialize/__init__.py b/dvc/utils/serialize/__init__.py
index b2b658454f..67eae15aa3 100644
--- a/dvc/utils/serialize/__init__.py
+++ b/dvc/utils/serialize/__init__.py
@@ -14,6 +14,13 @@
     {".toml": load_toml, ".json": load_json, ".py": load_py}  # noqa: F405
 )
 
+DUMPERS: DefaultDict[str, DumperFn] = defaultdict(  # noqa: F405
+    lambda: dump_yaml  # noqa: F405
+)
+DUMPERS.update(
+    {".toml": dump_toml, ".json": dump_json, ".py": dump_py}  # noqa: F405
+)
+
 MODIFIERS: DefaultDict[str, ModifierFn] = defaultdict(  # noqa: F405
     lambda: modify_yaml  # noqa: F405
 )
diff --git a/dvc/utils/serialize/_common.py b/dvc/utils/serialize/_common.py
index 71436523c5..0ef6ec8642 100644
--- a/dvc/utils/serialize/_common.py
+++ b/dvc/utils/serialize/_common.py
@@ -63,10 +63,16 @@ def _load_data(path: "AnyPath", parser: ParserFn, fs: "BaseFileSystem" = None):
             return parser(fd.read(), path)
 
 
-def _dump_data(path, data: Any, dumper: DumperFn, fs: "BaseFileSystem" = None):
+def _dump_data(
+    path,
+    data: Any,
+    dumper: DumperFn,
+    fs: "BaseFileSystem" = None,
+    **dumper_args,
+):
     open_fn = fs.open if fs else open
     with open_fn(path, "w+", encoding="utf-8") as fd:  # type: ignore
-        dumper(data, fd)
+        dumper(data, fd, **dumper_args)
 
 
 @contextmanager
diff --git a/dvc/utils/serialize/_json.py b/dvc/utils/serialize/_json.py
index 16c6950f7f..ee2a9ce493 100644
--- a/dvc/utils/serialize/_json.py
+++ b/dvc/utils/serialize/_json.py
@@ -19,8 +19,8 @@ def parse_json(text, path, **kwargs):
         return json.loads(text, **kwargs) or {}
 
 
-def dump_json(path, data, fs=None):
-    return _dump_data(path, data, dumper=json.dump, fs=fs)
+def dump_json(path, data, fs=None, **kwargs):
+    return _dump_data(path, data, dumper=json.dump, fs=fs, **kwargs)
 
 
 @contextmanager
diff --git a/dvc/utils/serialize/_toml.py b/dvc/utils/serialize/_toml.py
index 559b4a11ff..11fcaf8320 100644
--- a/dvc/utils/serialize/_toml.py
+++ b/dvc/utils/serialize/_toml.py
@@ -40,8 +40,8 @@ def _dump(data, stream):
     return toml.dump(data, stream, encoder=toml.TomlPreserveCommentEncoder())
 
 
-def dump_toml(path, data, fs=None):
-    return _dump_data(path, data, dumper=_dump, fs=fs)
+def dump_toml(path, data, fs=None, **kwargs):
+    return _dump_data(path, data, dumper=_dump, fs=fs, **kwargs)
 
 
 @contextmanager
diff --git a/dvc/utils/serialize/_yaml.py b/dvc/utils/serialize/_yaml.py
index 69cc4e8e18..12e0b70a4e 100644
--- a/dvc/utils/serialize/_yaml.py
+++ b/dvc/utils/serialize/_yaml.py
@@ -58,8 +58,8 @@ def _dump(data, stream):
     return yaml.dump(data, stream)
 
 
-def dump_yaml(path, data, fs=None):
-    return _dump_data(path, data, dumper=_dump, fs=fs)
+def dump_yaml(path, data, fs=None, **kwargs):
+    return _dump_data(path, data, dumper=_dump, fs=fs, **kwargs)
 
 
 def loads_yaml(s, typ="safe"):
diff --git a/tests/dir_helpers.py b/tests/dir_helpers.py
index f50ac02fc8..e10cd737e4 100644
--- a/tests/dir_helpers.py
+++ b/tests/dir_helpers.py
@@ -49,12 +49,14 @@
 import pathlib
 import sys
 from contextlib import contextmanager
+from functools import partialmethod
 from textwrap import dedent
 
 import pytest
 from funcy import lmap, retry
 
 from dvc.logger import disable_other_loggers
+from dvc.utils import serialize
 from dvc.utils.fs import makedirs
 
 __all__ = [
@@ -250,6 +252,22 @@ def read_text(self, *args, **kwargs):  # pylint: disable=signature-differs
     def hash_to_path_info(self, hash_):
         return self / hash_[0:2] / hash_[2:]
 
+    def dump(self, *args, **kwargs):
+        return serialize.DUMPERS[self.suffix](self, *args, **kwargs)
+
+    def parse(self, *args, **kwargs):
+        return serialize.LOADERS[self.suffix](self, *args, **kwargs)
+
+    def modify(self, *args, **kwargs):
+        return serialize.MODIFIERS[self.suffix](self, *args, **kwargs)
+
+    load_yaml = partialmethod(serialize.load_yaml)
+    dump_yaml = partialmethod(serialize.dump_yaml)
+    load_json = partialmethod(serialize.load_json)
+    dump_json = partialmethod(serialize.dump_json)
+    load_toml = partialmethod(serialize.load_toml)
+    dump_toml = partialmethod(serialize.dump_toml)
+
 
 def _coerce_filenames(filenames):
     if isinstance(filenames, (str, bytes, pathlib.PurePath)):
diff --git a/tests/func/experiments/test_experiments.py b/tests/func/experiments/test_experiments.py
index 587ff53f18..a06cedf2e9 100644
--- a/tests/func/experiments/test_experiments.py
+++ b/tests/func/experiments/test_experiments.py
@@ -8,7 +8,7 @@
 
 from dvc.dvcfile import PIPELINE_FILE
 from dvc.repo.experiments.utils import exp_refs_by_rev
-from dvc.utils.serialize import PythonFileCorruptedError, load_yaml
+from dvc.utils.serialize import PythonFileCorruptedError
 from tests.func.test_repro_multistage import COPY_SCRIPT
 
 
@@ -660,9 +660,9 @@ def test_modified_data_dep(tmp_dir, scm, dvc, workspace, params, target):
 
 
 def test_exp_run_recursive(tmp_dir, scm, dvc, run_copy_metrics):
-    tmp_dir.dvc_gen("metric_t.json", "foo: 1")
+    tmp_dir.dvc_gen("metric_t.json", '{"foo": 1}')
     run_copy_metrics(
         "metric_t.json", "metric.json", metrics=["metric.json"], no_exec=True
     )
     assert dvc.experiments.run(".", recursive=True)
-    assert load_yaml(tmp_dir / "metric.json") == {"foo": 1}
+    assert (tmp_dir / "metric.json").parse() == {"foo": 1}
diff --git a/tests/func/experiments/test_show.py b/tests/func/experiments/test_show.py
index a8e5127005..2e87018cf1 100644
--- a/tests/func/experiments/test_show.py
+++ b/tests/func/experiments/test_show.py
@@ -11,7 +11,7 @@
 from dvc.repo.experiments.executor.base import BaseExecutor, ExecutorInfo
 from dvc.repo.experiments.utils import exp_refs_by_rev
 from dvc.utils.fs import makedirs
-from dvc.utils.serialize import YAMLFileCorruptedError, dump_yaml
+from dvc.utils.serialize import YAMLFileCorruptedError
 from tests.func.test_repro_multistage import COPY_SCRIPT
 
 
@@ -286,7 +286,7 @@ def test_show_filter(
         "train/bar": 1,
         "nested": {"foo": 1, "bar": 1},
     }
-    dump_yaml(params_file, params_data)
+    (tmp_dir / params_file).dump(params_data)
 
     dvc.run(
         cmd="python copy.py params.yaml metrics.yaml",
@@ -386,7 +386,7 @@ def test_show_running_workspace(tmp_dir, scm, dvc, exp_stage, capsys):
     makedirs(pid_dir, True)
     info = ExecutorInfo(None, None, None, BaseExecutor.DEFAULT_LOCATION)
     pidfile = os.path.join(pid_dir, f"workspace{BaseExecutor.PIDFILE_EXT}")
-    dump_yaml(pidfile, info.to_dict())
+    (tmp_dir / pidfile).dump(info.to_dict())
 
     assert dvc.experiments.show()["workspace"] == {
         "baseline": {
@@ -417,7 +417,7 @@ def test_show_running_executor(tmp_dir, scm, dvc, exp_stage):
     makedirs(pid_dir, True)
     info = ExecutorInfo(None, None, None, BaseExecutor.DEFAULT_LOCATION)
     pidfile = os.path.join(pid_dir, f"{exp_rev}{BaseExecutor.PIDFILE_EXT}")
-    dump_yaml(pidfile, info.to_dict())
+    (tmp_dir / pidfile).dump(info.to_dict())
 
     results = dvc.experiments.show()
     exp_data = get_in(results, [baseline_rev, exp_rev, "data"])
@@ -455,7 +455,7 @@ def test_show_running_checkpoint(
     info = ExecutorInfo(123, "foo.git", baseline_rev, executor)
     rev = "workspace" if workspace else stash_rev
     pidfile = os.path.join(pid_dir, f"{rev}{BaseExecutor.PIDFILE_EXT}")
-    dump_yaml(pidfile, info.to_dict())
+    (tmp_dir / pidfile).dump(info.to_dict())
 
     mocker.patch.object(
         BaseExecutor, "fetch_exps", return_value=[str(exp_ref)]
diff --git a/tests/func/metrics/test_diff.py b/tests/func/metrics/test_diff.py
index 5c6e847c3c..4eac8ffd1f 100644
--- a/tests/func/metrics/test_diff.py
+++ b/tests/func/metrics/test_diff.py
@@ -1,7 +1,6 @@
 import json
 
 from dvc.main import main
-from dvc.utils.serialize import dump_yaml
 
 
 def test_metrics_diff_simple(tmp_dir, scm, dvc, run_copy_metrics):
@@ -22,7 +21,7 @@ def _gen(val):
 def test_metrics_diff_yaml(tmp_dir, scm, dvc, run_copy_metrics):
     def _gen(val):
         metrics = {"a": {"b": {"c": val, "d": 1, "e": str(val)}}}
-        dump_yaml("m_temp.yaml", metrics)
+        (tmp_dir / "m_temp.yaml").dump(metrics)
         run_copy_metrics(
             "m_temp.yaml", "m.yaml", metrics=["m.yaml"], commit=str(val)
         )
@@ -39,7 +38,7 @@ def _gen(val):
 def test_metrics_diff_json(tmp_dir, scm, dvc, run_copy_metrics):
     def _gen(val):
         metrics = {"a": {"b": {"c": val, "d": 1, "e": str(val)}}}
-        tmp_dir.gen({"m_temp.json": json.dumps(metrics)})
+        (tmp_dir / "m_temp.json").dump(metrics)
         run_copy_metrics(
             "m_temp.json", "m.json", metrics=["m.json"], commit=str(val)
         )
@@ -55,7 +54,7 @@ def _gen(val):
 def test_metrics_diff_json_unchanged(tmp_dir, scm, dvc, run_copy_metrics):
     def _gen(val):
         metrics = {"a": {"b": {"c": val, "d": 1, "e": str(val)}}}
-        tmp_dir.gen({"m_temp.json": json.dumps(metrics)})
+        (tmp_dir / "m_temp.json").dump(metrics)
         run_copy_metrics(
             "m_temp.json", "m.json", metrics=["m.json"], commit=str(val)
         )
@@ -69,7 +68,7 @@ def _gen(val):
 
 def test_metrics_diff_broken_json(tmp_dir, scm, dvc, run_copy_metrics):
     metrics = {"a": {"b": {"c": 1, "d": 1, "e": "3"}}}
-    tmp_dir.gen({"m_temp.json": json.dumps(metrics)})
+    (tmp_dir / "m_temp.json").dump(metrics)
     run_copy_metrics(
         "m_temp.json",
         "m.json",
@@ -94,7 +93,7 @@ def test_metrics_diff_no_metrics(tmp_dir, scm, dvc):
 
 def test_metrics_diff_new_metric(tmp_dir, scm, dvc, run_copy_metrics):
     metrics = {"a": {"b": {"c": 1, "d": 1, "e": "3"}}}
-    tmp_dir.gen({"m_temp.json": json.dumps(metrics)})
+    (tmp_dir / "m_temp.json").dump(metrics)
     run_copy_metrics("m_temp.json", "m.json", metrics_no_cache=["m.json"])
 
     assert dvc.metrics.diff() == {
@@ -107,7 +106,7 @@ def test_metrics_diff_new_metric(tmp_dir, scm, dvc, run_copy_metrics):
 
 def test_metrics_diff_deleted_metric(tmp_dir, scm, dvc, run_copy_metrics):
     metrics = {"a": {"b": {"c": 1, "d": 1, "e": "3"}}}
-    tmp_dir.gen({"m_temp.json": json.dumps(metrics)})
+    (tmp_dir / "m_temp.json").dump(metrics)
     run_copy_metrics(
         "m_temp.json",
         "m.json",
diff --git a/tests/func/metrics/test_show.py b/tests/func/metrics/test_show.py
index 2299093ae9..7cdbb3cc72 100644
--- a/tests/func/metrics/test_show.py
+++ b/tests/func/metrics/test_show.py
@@ -6,10 +6,9 @@
 
 from dvc.dvcfile import PIPELINE_FILE
 from dvc.exceptions import OverlappingOutputPathsError
-from dvc.path_info import PathInfo
 from dvc.repo import Repo
 from dvc.utils.fs import remove
-from dvc.utils.serialize import YAMLFileCorruptedError, dump_yaml, modify_yaml
+from dvc.utils.serialize import YAMLFileCorruptedError
 
 
 def test_show_simple(tmp_dir, dvc, run_copy_metrics):
@@ -227,10 +226,10 @@ def test_show_no_metrics_files(tmp_dir, dvc, caplog):
 def test_metrics_show_overlap(
     tmp_dir, dvc, run_copy_metrics, clear_before_run
 ):
-    data_dir = PathInfo("data")
-    (tmp_dir / data_dir).mkdir()
+    data_dir = tmp_dir / "data"
+    data_dir.mkdir()
 
-    dump_yaml(data_dir / "m1_temp.yaml", {"a": {"b": {"c": 2, "d": 1}}})
+    (data_dir / "m1_temp.yaml").dump({"a": {"b": {"c": 2, "d": 1}}})
     run_copy_metrics(
         str(data_dir / "m1_temp.yaml"),
         str(data_dir / "m1.yaml"),
@@ -239,7 +238,7 @@ def test_metrics_show_overlap(
         name="cp-m1",
         metrics=[str(data_dir / "m1.yaml")],
     )
-    with modify_yaml("dvc.yaml") as d:
+    with (tmp_dir / "dvc.yaml").modify() as d:
         # trying to make an output overlaps error
         d["stages"]["corrupted-stage"] = {
             "cmd": "mkdir data",
diff --git a/tests/func/params/test_diff.py b/tests/func/params/test_diff.py
index 613ac27770..2c1cc3a94a 100644
--- a/tests/func/params/test_diff.py
+++ b/tests/func/params/test_diff.py
@@ -1,5 +1,4 @@
 from dvc.utils import relpath
-from dvc.utils.serialize import dump_yaml
 
 
 def test_diff_no_params(tmp_dir, scm, dvc):
@@ -140,7 +139,7 @@ def test_no_commits(tmp_dir):
 def test_vars_shows_on_params_diff(tmp_dir, scm, dvc):
     params_file = tmp_dir / "test_params.yaml"
     param_data = {"vars": {"model1": {"epoch": 15}, "model2": {"epoch": 35}}}
-    dump_yaml(params_file, param_data)
+    (tmp_dir / params_file).dump(param_data)
     d = {
         "vars": ["test_params.yaml"],
         "stages": {
@@ -150,7 +149,7 @@ def test_vars_shows_on_params_diff(tmp_dir, scm, dvc):
             }
         },
     }
-    dump_yaml("dvc.yaml", d)
+    (tmp_dir / "dvc.yaml").dump(d)
     assert dvc.params.diff() == {
         "test_params.yaml": {
             "vars.model1.epoch": {"new": 15, "old": None},
@@ -161,7 +160,7 @@ def test_vars_shows_on_params_diff(tmp_dir, scm, dvc):
     scm.commit("added stages")
 
     param_data["vars"]["model1"]["epoch"] = 20
-    dump_yaml(params_file, param_data)
+    (tmp_dir / params_file).dump(param_data)
     assert dvc.params.diff() == {
         "test_params.yaml": {
             "vars.model1.epoch": {"new": 20, "old": 15, "diff": 5}
diff --git a/tests/func/parsing/test_errors.py b/tests/func/parsing/test_errors.py
index 35a0f8fcc3..d49fb08698 100644
--- a/tests/func/parsing/test_errors.py
+++ b/tests/func/parsing/test_errors.py
@@ -10,7 +10,6 @@
 from dvc.parsing.context import Context
 from dvc.parsing.interpolate import embrace
 from dvc.utils.humanize import join
-from dvc.utils.serialize import dump_yaml
 
 from . import make_entry_definition, make_foreach_def
 
@@ -133,7 +132,7 @@ def test_interpolate_non_string(tmp_dir, dvc):
 
 
 def test_partial_vars_doesnot_exist(tmp_dir, dvc):
-    dump_yaml("test_params.yaml", {"sub1": "sub1", "sub2": "sub2"})
+    (tmp_dir / "test_params.yaml").dump({"sub1": "sub1", "sub2": "sub2"})
 
     definition = make_entry_definition(
         tmp_dir,
@@ -281,7 +280,7 @@ def test_item_key_in_generated_stage_vars(tmp_dir, dvc, redefine, from_file):
     context = Context(foo="bar")
     vars_ = [redefine]
     if from_file:
-        dump_yaml("test_params.yaml", redefine)
+        (tmp_dir / "test_params.yaml").dump(redefine)
         vars_ = ["test_params.yaml"]
 
     definition = make_foreach_def(
diff --git a/tests/func/parsing/test_foreach.py b/tests/func/parsing/test_foreach.py
index a864beebdb..bec9a532ab 100644
--- a/tests/func/parsing/test_foreach.py
+++ b/tests/func/parsing/test_foreach.py
@@ -5,7 +5,6 @@
 
 from dvc.parsing import DEFAULT_PARAMS_FILE, DataResolver, ForeachDefinition
 from dvc.parsing.context import Context
-from dvc.utils.serialize import dump_yaml
 
 
 def test_with_simple_list_data(tmp_dir, dvc):
@@ -111,7 +110,7 @@ def test_foreach_interpolate_with_composite_data(
 def test_params_file_with_dict_tracked(tmp_dir, dvc):
     foreach_data = {"model1": {"thresh": "foo"}, "model2": {"thresh": "bar"}}
     params = {"models": foreach_data}
-    dump_yaml("params.yaml", params)
+    (tmp_dir / "params.yaml").dump(params)
 
     resolver = DataResolver(dvc, tmp_dir, {})
     data = {"foreach": "${models}", "do": {"cmd": "echo ${item.thresh}"}}
@@ -132,7 +131,7 @@ def test_params_file_with_dict_tracked(tmp_dir, dvc):
 def test_params_file_tracked_for_composite_list(tmp_dir, dvc):
     foreach_data = [{"thresh": "foo"}, {"thresh": "bar"}]
     params = {"models": foreach_data}
-    dump_yaml("params.yaml", params)
+    (tmp_dir / "params.yaml").dump(params)
 
     resolver = DataResolver(dvc, tmp_dir, {})
     data = {"foreach": "${models}", "do": {"cmd": "echo ${item.thresh}"}}
@@ -178,8 +177,8 @@ def test_foreach_partial_interpolations(tmp_dir, dvc):
 
 
 def test_mixed_vars_for_foreach_data(tmp_dir, dvc):
-    dump_yaml("params.yaml", {"models": {"model1": "foo"}})
-    dump_yaml("test_params.yaml", {"models": {"model2": "bar"}})
+    (tmp_dir / "params.yaml").dump({"models": {"model1": "foo"}})
+    (tmp_dir / "test_params.yaml").dump({"models": {"model2": "bar"}})
 
     resolver = DataResolver(dvc, tmp_dir, {"vars": ["test_params.yaml"]})
     data = {"foreach": "${models}", "do": {"cmd": "echo ${item}"}}
@@ -197,12 +196,10 @@ def test_mixed_vars_for_foreach_data(tmp_dir, dvc):
 
 
 def test_mixed_vars_for_foreach_data_2(tmp_dir, dvc):
-    dump_yaml(
-        "params.yaml",
+    (tmp_dir / "params.yaml").dump(
         {"models": {"model1": {"thresh": 10}, "model2": {"thresh": 15}}},
     )
-    dump_yaml(
-        "test_params.yaml",
+    (tmp_dir / "test_params.yaml").dump(
         {"models": {"model1": {"epochs": 5}, "model2": {"epochs": 10}}},
     )
 
@@ -288,9 +285,9 @@ def test_foreach_with_local_vars(tmp_dir, dvc):
     ],
 )
 def test_foreach_with_imported_vars(tmp_dir, dvc, local_import):
-    dump_yaml("params.yaml", {"models": {"model1": {"thresh": "foo"}}})
-    dump_yaml(
-        "test_params.yaml", {"train": {"epochs": 10}, "prepare": {"nums": 25}}
+    (tmp_dir / "params.yaml").dump({"models": {"model1": {"thresh": "foo"}}})
+    (tmp_dir / "test_params.yaml").dump(
+        {"train": {"epochs": 10}, "prepare": {"nums": 25}}
     )
     resolver = DataResolver(dvc, tmp_dir, {})
     foreach_data = ["foo", "bar"]
@@ -321,13 +318,12 @@ def test_foreach_with_imported_vars(tmp_dir, dvc, local_import):
 def test_foreach_with_interpolated_wdir_and_local_vars(
     tmp_dir, dvc, local_import
 ):
-    dump_yaml("params.yaml", {"models": {"model1": {"thresh": "foo"}}})
+    (tmp_dir / "params.yaml").dump({"models": {"model1": {"thresh": "foo"}}})
 
     for i in range(5):
         build_dir = tmp_dir / ("model-" + str(i))
         build_dir.mkdir()
-        dump_yaml(
-            build_dir / "params.yaml",
+        (build_dir / "params.yaml").dump(
             {"train": {"epochs": 1 + i}, "prepare": {"nums": 10 * i}},
         )
 
diff --git a/tests/func/parsing/test_interpolated_entry.py b/tests/func/parsing/test_interpolated_entry.py
index 5c5072c17b..312f2be6c4 100644
--- a/tests/func/parsing/test_interpolated_entry.py
+++ b/tests/func/parsing/test_interpolated_entry.py
@@ -6,7 +6,6 @@
 from dvc.parsing import DEFAULT_PARAMS_FILE, DataResolver
 from dvc.parsing.context import recurse_not_a_node
 from dvc.path_info import PathInfo
-from dvc.utils.serialize import dump_json, dump_yaml
 
 from . import (
     CONTEXT_DATA,
@@ -30,7 +29,7 @@ def assert_stage_equal(d1, d2):
 
 
 def test_simple(tmp_dir, dvc):
-    dump_yaml(tmp_dir / DEFAULT_PARAMS_FILE, CONTEXT_DATA)
+    (tmp_dir / DEFAULT_PARAMS_FILE).dump(CONTEXT_DATA)
     resolver = DataResolver(
         dvc, PathInfo(str(tmp_dir)), deepcopy(TEMPLATED_DVC_YAML_DATA)
     )
@@ -46,7 +45,7 @@ def test_vars_import(tmp_dir, dvc):
     Test that different file can be loaded using `vars`
     instead of default params.yaml.
     """
-    dump_yaml(tmp_dir / "params2.yaml", CONTEXT_DATA)
+    (tmp_dir / "params2.yaml").dump(CONTEXT_DATA)
     d = deepcopy(TEMPLATED_DVC_YAML_DATA)
     d["vars"] = ["params2.yaml"]
     resolver = DataResolver(dvc, PathInfo(str(tmp_dir)), d)
@@ -68,7 +67,7 @@ def test_vars_and_params_import(tmp_dir, dvc):
         "vars": [DEFAULT_PARAMS_FILE, {"dict": {"foo": "foobar"}}],
         "stages": {"stage1": {"cmd": "echo ${dict.foo} ${dict.bar}"}},
     }
-    dump_yaml(tmp_dir / DEFAULT_PARAMS_FILE, {"dict": {"bar": "bar"}})
+    (tmp_dir / DEFAULT_PARAMS_FILE).dump({"dict": {"bar": "bar"}})
     resolver = DataResolver(dvc, PathInfo(str(tmp_dir)), d)
 
     assert_stage_equal(
@@ -96,8 +95,8 @@ def test_stage_with_wdir(tmp_dir, dvc):
 
     data_dir = tmp_dir / "data"
     data_dir.mkdir()
-    dump_yaml(tmp_dir / DEFAULT_PARAMS_FILE, {"dict": {"bar": "bar"}})
-    dump_json(data_dir / DEFAULT_PARAMS_FILE, {"dict": {"foo": "foo"}})
+    (tmp_dir / DEFAULT_PARAMS_FILE).dump({"dict": {"bar": "bar"}})
+    (data_dir / DEFAULT_PARAMS_FILE).dump({"dict": {"foo": "foo"}})
     resolver = DataResolver(dvc, PathInfo(str(tmp_dir)), d)
 
     assert_stage_equal(
@@ -135,12 +134,12 @@ def test_with_templated_wdir(tmp_dir, dvc):
             }
         }
     }
-    dump_yaml(
-        tmp_dir / DEFAULT_PARAMS_FILE, {"dict": {"bar": "bar", "ws": "data"}}
+    (tmp_dir / DEFAULT_PARAMS_FILE).dump(
+        {"dict": {"bar": "bar", "ws": "data"}}
     )
     data_dir = tmp_dir / "data"
     data_dir.mkdir()
-    dump_json(data_dir / DEFAULT_PARAMS_FILE, {"dict": {"foo": "foo"}})
+    (data_dir / DEFAULT_PARAMS_FILE).dump({"dict": {"foo": "foo"}})
     resolver = DataResolver(dvc, PathInfo(str(tmp_dir)), d)
 
     assert_stage_equal(
@@ -167,7 +166,7 @@ def test_with_templated_wdir(tmp_dir, dvc):
 
 def test_resolve_local_tries_to_load_globally_used_files(tmp_dir, dvc):
     iterable = {"bar": "bar", "foo": "foo"}
-    dump_json(tmp_dir / "params.json", iterable)
+    (tmp_dir / "params.json").dump(iterable)
 
     d = {
         "vars": ["params.json"],
@@ -196,7 +195,7 @@ def test_resolve_local_tries_to_load_globally_used_files(tmp_dir, dvc):
 
 def test_resolve_local_tries_to_load_globally_used_params_yaml(tmp_dir, dvc):
     iterable = {"bar": "bar", "foo": "foo"}
-    dump_yaml(tmp_dir / "params.yaml", iterable)
+    (tmp_dir / "params.yaml").dump(iterable)
 
     d = {
         "stages": {
@@ -224,7 +223,7 @@ def test_resolve_local_tries_to_load_globally_used_params_yaml(tmp_dir, dvc):
 
 def test_vars_relpath_overwrite(tmp_dir, dvc):
     iterable = {"bar": "bar", "foo": "foo"}
-    dump_yaml(tmp_dir / "params.yaml", iterable)
+    (tmp_dir / "params.yaml").dump(iterable)
     d = {
         "vars": ["params.yaml"],
         "stages": {
@@ -252,7 +251,7 @@ def test_vars_relpath_overwrite(tmp_dir, dvc):
 )
 def test_vars_load_partial(tmp_dir, dvc, local, vars_):
     iterable = {"bar": "bar", "foo": "foo"}
-    dump_yaml(tmp_dir / "test_params.yaml", iterable)
+    (tmp_dir / "test_params.yaml").dump(iterable)
     d = {"stages": {"build": {"cmd": "echo ${bar}"}}}
     if local:
         d["stages"]["build"]["vars"] = vars_
diff --git a/tests/func/parsing/test_resolver.py b/tests/func/parsing/test_resolver.py
index 30f4cb8d2a..6fd56e4590 100644
--- a/tests/func/parsing/test_resolver.py
+++ b/tests/func/parsing/test_resolver.py
@@ -5,7 +5,7 @@
 
 from dvc.parsing import DEFAULT_PARAMS_FILE, DataResolver, ResolveError
 from dvc.parsing.context import Context
-from dvc.utils.serialize import dump_json, dump_yaml, dumps_yaml
+from dvc.utils.serialize import dumps_yaml
 
 from . import CONTEXT_DATA, RESOLVED_DVC_YAML_DATA, TEMPLATED_DVC_YAML_DATA
 
@@ -50,16 +50,16 @@ def test_vars_interpolation_errors(tmp_dir, dvc, vars_):
     "vars_", [{}, {"vars": []}, {"vars": [DEFAULT_PARAMS_FILE]}]
 )
 def test_default_params_file(tmp_dir, dvc, vars_):
-    dump_yaml(DEFAULT_PARAMS_FILE, DATA)
+    (tmp_dir / DEFAULT_PARAMS_FILE).dump(DATA)
     resolver = DataResolver(dvc, tmp_dir, vars_)
     assert resolver.context == DATA
 
 
 def test_load_vars_from_file(tmp_dir, dvc):
-    dump_yaml(DEFAULT_PARAMS_FILE, DATA)
+    (tmp_dir / DEFAULT_PARAMS_FILE).dump(DATA)
 
     datasets = {"datasets": ["foo", "bar"]}
-    dump_json("params.json", datasets)
+    (tmp_dir / "params.json").dump(datasets)
     d = {"vars": [DEFAULT_PARAMS_FILE, "params.json"]}
     resolver = DataResolver(dvc, tmp_dir, d)
 
@@ -82,7 +82,7 @@ def test_load_vars_with_relpath(tmp_dir, scm, dvc):
 
 
 def test_partial_vars_doesnot_exist(tmp_dir, dvc):
-    dump_yaml("test_params.yaml", {"sub1": "sub1"})
+    (tmp_dir / "test_params.yaml").dump({"sub1": "sub1"})
 
     with pytest.raises(ResolveError) as exc_info:
         DataResolver(dvc, tmp_dir, {"vars": ["test_params.yaml:sub2"]})
@@ -94,8 +94,8 @@ def test_partial_vars_doesnot_exist(tmp_dir, dvc):
 
 
 def test_global_overwrite_error_on_imports(tmp_dir, dvc):
-    dump_yaml(DEFAULT_PARAMS_FILE, DATA)
-    dump_json("params.json", DATA)
+    (tmp_dir / DEFAULT_PARAMS_FILE).dump(DATA)
+    (tmp_dir / "params.json").dump(DATA)
 
     d = {"vars": [DEFAULT_PARAMS_FILE, "params.json"]}
     with pytest.raises(ResolveError) as exc_info:
@@ -109,7 +109,7 @@ def test_global_overwrite_error_on_imports(tmp_dir, dvc):
 
 
 def test_global_overwrite_vars(tmp_dir, dvc):
-    dump_yaml(DEFAULT_PARAMS_FILE, DATA)
+    (tmp_dir / DEFAULT_PARAMS_FILE).dump(DATA)
     d = {"vars": [DATA]}
 
     with pytest.raises(ResolveError) as exc_info:
@@ -123,7 +123,7 @@ def test_global_overwrite_vars(tmp_dir, dvc):
 
 
 def test_local_declared_vars_overwrite(tmp_dir, dvc):
-    dump_yaml(DEFAULT_PARAMS_FILE, DATA)
+    (tmp_dir / DEFAULT_PARAMS_FILE).dump(DATA)
 
     d = {"vars": [DATA["models"], DATA["models"]]}
     with pytest.raises(ResolveError) as exc_info:
@@ -158,7 +158,7 @@ def test_specified_params_file_not_exist(tmp_dir, dvc):
 )
 def test_vars_already_loaded_message(tmp_dir, dvc, local, vars_):
     d = {"stages": {"build": {"cmd": "echo ${sub1} ${sub2}"}}}
-    dump_yaml("test_params.yaml", {"sub1": "sub1", "sub2": "sub2"})
+    (tmp_dir / "test_params.yaml").dump({"sub1": "sub1", "sub2": "sub2"})
     if not local:
         d["vars"] = vars_
     else:
@@ -175,8 +175,8 @@ def test_vars_already_loaded_message(tmp_dir, dvc, local, vars_):
     "vars_, loc", [(DATA, "build.vars[0]"), ("params.json", "params.json")]
 )
 def test_local_overwrite_error(tmp_dir, dvc, vars_, loc):
-    dump_yaml(DEFAULT_PARAMS_FILE, DATA)
-    dump_json("params.json", DATA)
+    (tmp_dir / DEFAULT_PARAMS_FILE).dump(DATA)
+    (tmp_dir / "params.json").dump(DATA)
 
     d = {"stages": {"build": {"cmd": "echo ${models.foo}", "vars": [vars_]}}}
 
diff --git a/tests/func/plots/test_diff.py b/tests/func/plots/test_diff.py
index 3d20410e64..739938449c 100644
--- a/tests/func/plots/test_diff.py
+++ b/tests/func/plots/test_diff.py
@@ -1,8 +1,5 @@
-from tests.func.plots.utils import _write_json
-
-
 def test_diff_dirty(tmp_dir, scm, dvc, run_copy_metrics):
-    _write_json(tmp_dir, [{"y": 2}, {"y": 3}], "metric_t.json")
+    (tmp_dir / "metric_t.json").dump([{"y": 2}, {"y": 3}], sort_keys=True)
     run_copy_metrics(
         "metric_t.json",
         "metric.json",
@@ -11,7 +8,7 @@ def test_diff_dirty(tmp_dir, scm, dvc, run_copy_metrics):
     )
 
     metric_head = [{"y": 3}, {"y": 5}]
-    _write_json(tmp_dir, metric_head, "metric_t.json")
+    (tmp_dir / "metric_t.json").dump_json(metric_head, sort_keys=True)
     run_copy_metrics(
         "metric_t.json",
         "metric.json",
@@ -20,7 +17,7 @@ def test_diff_dirty(tmp_dir, scm, dvc, run_copy_metrics):
     )
 
     metric_1 = [{"y": 5}, {"y": 6}]
-    _write_json(tmp_dir, metric_1, "metric_t.json")
+    (tmp_dir / "metric_t.json").dump_json(metric_1, sort_keys=True)
     run_copy_metrics(
         "metric_t.json", "metric.json", plots_no_cache=["metric.json"]
     )
@@ -35,7 +32,7 @@ def test_diff_dirty(tmp_dir, scm, dvc, run_copy_metrics):
         },
     }
     metric_2 = [{"y": 7}, {"y": 8}]
-    _write_json(tmp_dir, metric_2, "metric.json")
+    (tmp_dir / "metric.json").dump_json(metric_2, sort_keys=True)
 
     diff_result = dvc.plots.diff(props=props)
     assert diff_result == {
diff --git a/tests/func/plots/test_modify.py b/tests/func/plots/test_modify.py
index 7c1aa07604..ef4e2918e2 100644
--- a/tests/func/plots/test_modify.py
+++ b/tests/func/plots/test_modify.py
@@ -6,14 +6,13 @@
 from dvc.repo.plots import PropsNotFoundError
 from dvc.repo.plots.template import TemplateNotFoundError
 from dvc.utils import relpath
-from tests.func.plots.utils import _write_json
 
 
 def test_plots_modify_existing_template(
     tmp_dir, dvc, run_copy_metrics, custom_template
 ):
     metric = [{"a": 1, "b": 2}, {"a": 2, "b": 3}]
-    _write_json(tmp_dir, metric, "metric_t.json")
+    (tmp_dir / "metric_t.json").dump_json(metric, sort_keys=True)
     stage = run_copy_metrics(
         "metric_t.json",
         "metric.json",
@@ -31,7 +30,7 @@ def test_plots_modify_existing_template(
 def test_plots_modify_should_not_change_lockfile(
     tmp_dir, dvc, run_copy_metrics, custom_template
 ):
-    _write_json(tmp_dir, [{"a": 1, "b": 2}], "metric_t.json")
+    (tmp_dir / "metric_t.json").dump_json([{"a": 1, "b": 2}], sort_keys=True)
     run_copy_metrics(
         "metric_t.json",
         "metric.json",
@@ -56,7 +55,7 @@ def test_plots_modify_not_existing_template(dvc):
 
 def test_unset_nonexistent(tmp_dir, dvc, run_copy_metrics, custom_template):
     metric = [{"a": 1, "b": 2}, {"a": 2, "b": 3}]
-    _write_json(tmp_dir, metric, "metric_t.json")
+    (tmp_dir / "metric_t.json").dump_json(metric, sort_keys=True)
     run_copy_metrics(
         "metric_t.json",
         "metric.json",
@@ -76,7 +75,7 @@ def test_dir_plots(tmp_dir, dvc, run_copy_metrics):
     metric = [{"first_val": 100, "val": 2}, {"first_val": 200, "val": 3}]
 
     fname = "file.json"
-    _write_json(tmp_dir, metric, fname)
+    (tmp_dir / fname).dump_json(metric, sort_keys=True)
 
     p1 = os.path.join("subdir", "p1.json")
     p2 = os.path.join("subdir", "p2.json")
diff --git a/tests/func/plots/test_show.py b/tests/func/plots/test_show.py
index 94e57049fe..e1b1c8a237 100644
--- a/tests/func/plots/test_show.py
+++ b/tests/func/plots/test_show.py
@@ -6,23 +6,16 @@
 from dvc.dvcfile import PIPELINE_FILE
 from dvc.exceptions import OverlappingOutputPathsError
 from dvc.main import main
-from dvc.path_info import PathInfo
 from dvc.render.vega import PlotMetricTypeError
 from dvc.repo import Repo
 from dvc.utils import onerror_collect
 from dvc.utils.fs import remove
-from dvc.utils.serialize import (
-    EncodingError,
-    YAMLFileCorruptedError,
-    dump_yaml,
-    modify_yaml,
-)
-from tests.func.plots.utils import _write_json
+from dvc.utils.serialize import EncodingError, YAMLFileCorruptedError
 
 
 def test_plot_cache_missing(tmp_dir, scm, dvc, caplog, run_copy_metrics):
     metric1 = [{"y": 2}, {"y": 3}]
-    _write_json(tmp_dir, metric1, "metric_t.json")
+    (tmp_dir / "metric_t.json").dump_json(metric1, sort_keys=True)
     run_copy_metrics(
         "metric_t.json",
         "metric.json",
@@ -33,7 +26,7 @@ def test_plot_cache_missing(tmp_dir, scm, dvc, caplog, run_copy_metrics):
 
     # Make a different plot and then remove its datafile
     metric2 = [{"y": 3}, {"y": 4}]
-    _write_json(tmp_dir, metric2, "metric_t.json")
+    (tmp_dir / "metric_t.json").dump_json(metric2, sort_keys=True)
     stage = run_copy_metrics(
         "metric_t.json",
         "metric.json",
@@ -71,7 +64,7 @@ def test_plot_wrong_metric_type(tmp_dir, scm, dvc, run_copy_metrics):
 @pytest.mark.parametrize("use_dvc", [True, False])
 def test_show_non_plot(tmp_dir, scm, use_dvc):
     metric = [{"first_val": 100, "val": 2}, {"first_val": 200, "val": 3}]
-    _write_json(tmp_dir, metric, "metric.json")
+    (tmp_dir / "metric.json").dump_json(metric, sort_keys=True)
 
     if use_dvc:
         dvc = Repo.init()
@@ -87,7 +80,7 @@ def test_show_non_plot_and_plot_with_params(
     tmp_dir, scm, dvc, run_copy_metrics
 ):
     metric = [{"first_val": 100, "val": 2}, {"first_val": 200, "val": 3}]
-    _write_json(tmp_dir, metric, "metric.json")
+    (tmp_dir / "metric.json").dump_json(metric, sort_keys=True)
     run_copy_metrics(
         "metric.json", "metric2.json", plots_no_cache=["metric2.json"]
     )
@@ -105,7 +98,7 @@ def test_show_from_subdir(tmp_dir, dvc, capsys):
 
     subdir.mkdir()
     metric = [{"first_val": 100, "val": 2}, {"first_val": 200, "val": 3}]
-    _write_json(subdir, metric, "metric.json")
+    (subdir / "metric.json").dump_json(metric, sort_keys=True)
 
     with subdir.chdir():
         assert main(["plots", "show", "metric.json"]) == 0
@@ -127,10 +120,10 @@ def test_plots_show_non_existing(tmp_dir, dvc, caplog):
 
 @pytest.mark.parametrize("clear_before_run", [True, False])
 def test_plots_show_overlap(tmp_dir, dvc, run_copy_metrics, clear_before_run):
-    data_dir = PathInfo("data")
-    (tmp_dir / data_dir).mkdir()
+    data_dir = tmp_dir / "data"
+    data_dir.mkdir()
 
-    dump_yaml(data_dir / "m1_temp.yaml", {"a": {"b": {"c": 2, "d": 1}}})
+    (data_dir / "m1_temp.yaml").dump({"a": {"b": {"c": 2, "d": 1}}})
     run_copy_metrics(
         str(data_dir / "m1_temp.yaml"),
         str(data_dir / "m1.yaml"),
@@ -139,7 +132,7 @@ def test_plots_show_overlap(tmp_dir, dvc, run_copy_metrics, clear_before_run):
         name="cp-m1",
         plots=[str(data_dir / "m1.yaml")],
     )
-    with modify_yaml("dvc.yaml") as d:
+    with (tmp_dir / "dvc.yaml").modify() as d:
         # trying to make an output overlaps error
         d["stages"]["corrupted-stage"] = {
             "cmd": "mkdir data",
@@ -167,7 +160,7 @@ def test_dir_plots(tmp_dir, dvc, run_copy_metrics):
     metric = [{"first_val": 100, "val": 2}, {"first_val": 200, "val": 3}]
 
     fname = "file.json"
-    _write_json(tmp_dir, metric, fname)
+    (tmp_dir / fname).dump_json(metric, sort_keys=True)
 
     p1 = os.path.join("subdir", "p1.json")
     p2 = os.path.join("subdir", "p2.json")
@@ -213,7 +206,7 @@ def test_log_errors(
     tmp_dir, scm, dvc, run_copy_metrics, file, error_path, capsys
 ):
     metric = [{"val": 2}, {"val": 3}]
-    dump_yaml("metric_t.yaml", metric)
+    (tmp_dir / "metric_t.yaml").dump(metric)
     run_copy_metrics(
         "metric_t.yaml",
         "plot.yaml",
diff --git a/tests/func/plots/utils.py b/tests/func/plots/utils.py
deleted file mode 100644
index 3ebb039110..0000000000
--- a/tests/func/plots/utils.py
+++ /dev/null
@@ -1,12 +0,0 @@
-import json
-
-from tests.utils import dump_sv
-
-
-def _write_csv(metric, filename, header=True):
-    with open(filename, "w", newline="") as csvobj:
-        dump_sv(csvobj, metric, delimiter=",", header=header)
-
-
-def _write_json(tmp_dir, metric, filename):
-    tmp_dir.gen(filename, json.dumps(metric, sort_keys=True))
diff --git a/tests/func/test_add.py b/tests/func/test_add.py
index 8e2040cddf..070207fe0b 100644
--- a/tests/func/test_add.py
+++ b/tests/func/test_add.py
@@ -55,7 +55,7 @@ def test_add(tmp_dir, dvc):
     assert stage.outs[0].hash_info == HashInfo("md5", md5)
     assert stage.md5 is None
 
-    assert load_yaml("foo.dvc") == {
+    assert (tmp_dir / "foo.dvc").parse() == {
         "outs": [
             {
                 "md5": "acbd18db4cc2f85cedef654fccc4a4d8",
@@ -73,7 +73,7 @@ def test_add_executable(tmp_dir, dvc):
     os.chmod("foo", st.st_mode | stat.S_IEXEC)
     dvc.add("foo")
 
-    assert load_yaml("foo.dvc") == {
+    assert (tmp_dir / "foo.dvc").parse() == {
         "outs": [
             {
                 "md5": "acbd18db4cc2f85cedef654fccc4a4d8",
diff --git a/tests/func/test_checkout.py b/tests/func/test_checkout.py
index 23809d9aca..1fa4fce82d 100644
--- a/tests/func/test_checkout.py
+++ b/tests/func/test_checkout.py
@@ -867,9 +867,9 @@ def test_checkout_external_modified_file(tmp_dir, dvc, scm, mocker, workspace):
 def test_checkout_executable(tmp_dir, dvc):
     tmp_dir.dvc_gen("foo", "foo")
 
-    contents = load_yaml("foo.dvc")
+    contents = (tmp_dir / "foo.dvc").parse()
     contents["outs"][0]["isexec"] = True
-    dump_yaml("foo.dvc", contents)
+    (tmp_dir / "foo.dvc").dump(contents)
 
     dvc.checkout("foo")
 
diff --git a/tests/func/test_commit.py b/tests/func/test_commit.py
index b507a552f1..afa7bec221 100644
--- a/tests/func/test_commit.py
+++ b/tests/func/test_commit.py
@@ -6,7 +6,6 @@
 from dvc.dvcfile import PIPELINE_FILE
 from dvc.output import OutputDoesNotExistError
 from dvc.stage.exceptions import StageCommitError
-from dvc.utils.serialize import dump_yaml, load_yaml
 from tests.utils import clean_staging
 
 
@@ -65,9 +64,9 @@ def test_commit_changed_md5(tmp_dir, dvc):
     tmp_dir.gen({"file": "file content"})
     (stage,) = dvc.add("file", no_commit=True)
 
-    stage_file_content = load_yaml(stage.path)
+    stage_file_content = (tmp_dir / stage.path).parse()
     stage_file_content["md5"] = "1111111111"
-    dump_yaml(stage.path, stage_file_content)
+    (tmp_dir / stage.path).dump(stage_file_content)
 
     clean_staging()
 
@@ -75,7 +74,7 @@ def test_commit_changed_md5(tmp_dir, dvc):
         dvc.commit(stage.path)
 
     dvc.commit(stage.path, force=True)
-    assert "md5" not in load_yaml(stage.path)
+    assert "md5" not in (tmp_dir / stage.path).parse()
 
 
 def test_commit_no_exec(tmp_dir, dvc):
diff --git a/tests/func/test_data_cloud.py b/tests/func/test_data_cloud.py
index bb2afe8719..82a0f44d8e 100644
--- a/tests/func/test_data_cloud.py
+++ b/tests/func/test_data_cloud.py
@@ -10,7 +10,6 @@
 from dvc.main import main
 from dvc.stage.exceptions import StageNotFound
 from dvc.utils.fs import move, remove
-from dvc.utils.serialize import dump_yaml, load_yaml
 
 all_clouds = [
     pytest.lazy_fixture(cloud)
@@ -204,9 +203,9 @@ def test_warn_on_outdated_stage(tmp_dir, dvc, local_remote, caplog):
     assert main(["push"]) == 0
 
     stage_file_path = stage.relpath
-    content = load_yaml(stage_file_path)
+    content = (tmp_dir / stage_file_path).parse()
     del content["outs"][0]["md5"]
-    dump_yaml(stage_file_path, content)
+    (tmp_dir / stage_file_path).dump(content)
 
     with caplog.at_level(logging.WARNING, logger="dvc"):
         caplog.clear()
@@ -612,8 +611,6 @@ def test_pull_no_00_prefix(tmp_dir, dvc, remote, monkeypatch):
 
 
 def test_output_remote(tmp_dir, dvc, make_remote):
-    from dvc.utils.serialize import modify_yaml
-
     make_remote("default", default=True)
     make_remote("for_foo", default=False)
     make_remote("for_data", default=False)
@@ -622,10 +619,10 @@ def test_output_remote(tmp_dir, dvc, make_remote):
     tmp_dir.dvc_gen("bar", "bar")
     tmp_dir.dvc_gen("data", {"one": "one", "two": "two"})
 
-    with modify_yaml("foo.dvc") as d:
+    with (tmp_dir / "foo.dvc").modify() as d:
         d["outs"][0]["remote"] = "for_foo"
 
-    with modify_yaml("data.dvc") as d:
+    with (tmp_dir / "data.dvc").modify() as d:
         d["outs"][0]["remote"] = "for_data"
 
     dvc.push()
diff --git a/tests/func/test_dvcfile.py b/tests/func/test_dvcfile.py
index 6710c04cb6..d487655558 100644
--- a/tests/func/test_dvcfile.py
+++ b/tests/func/test_dvcfile.py
@@ -15,7 +15,6 @@
     StageFileFormatError,
 )
 from dvc.stage.loader import StageNotFound
-from dvc.utils.serialize import dump_yaml
 
 
 def test_run_load_one_for_multistage(tmp_dir, dvc):
@@ -245,14 +244,14 @@ def test_remove_stage_on_lockfile_format_error(tmp_dir, dvc, run_copy):
     lock_data = lock_file.load()
     lock_data["gibberish"] = True
     data["gibberish"] = True
-    dump_yaml(lock_file.relpath, lock_data)
+    (tmp_dir / lock_file.relpath).dump(lock_data)
     with pytest.raises(StageFileFormatError):
         dvc_file.remove_stage(stage)
 
     lock_file.remove()
     dvc_file.dump(stage, update_pipeline=False)
 
-    dump_yaml(dvc_file.relpath, data)
+    (tmp_dir / dvc_file.relpath).dump(data)
     with pytest.raises(StageFileFormatError):
         dvc_file.remove_stage(stage)
 
@@ -335,7 +334,7 @@ def test_dvcfile_dump_preserves_desc(tmp_dir, dvc, run_copy):
 
     data = dvcfile._load()[0]
     data["stages"]["run_copy"]["outs"][0] = {"bar": {"desc": out_desc}}
-    dump_yaml(dvcfile.path, data)
+    (tmp_dir / dvcfile.path).dump(data)
 
     assert stage.desc == stage_desc
     stage.outs[0].desc = out_desc
@@ -376,7 +375,7 @@ 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"}]})
+    (tmp_dir / "dvc.yaml").dump({"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 eadc48effa..aafb5ef091 100644
--- a/tests/func/test_lockfile.py
+++ b/tests/func/test_lockfile.py
@@ -9,12 +9,7 @@
 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 (
-    dump_yaml,
-    dumps_yaml,
-    load_yaml,
-    parse_yaml_for_update,
-)
+from dvc.utils.serialize import dumps_yaml, parse_yaml_for_update
 from tests.func.test_run_multistage import supported_params
 
 FS_STRUCTURE = {
@@ -183,7 +178,7 @@ def v1_repo_lock(tmp_dir, dvc):
     }
     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)
+    (tmp_dir / "dvc.lock").dump(v1_lockdata)
     yield v1_lockdata
 
 
@@ -202,7 +197,7 @@ def test_migrates_v1_lockfile_to_v2_during_dump(
         assert dvc.reproduce()
 
     assert "Migrating lock file 'dvc.lock' from v1 to v2" in caplog.messages
-    d = load_yaml(tmp_dir / "dvc.lock")
+    d = (tmp_dir / "dvc.lock").parse()
     assert d == {"stages": v1_repo_lock, "schema": "2.0"}
 
 
@@ -211,7 +206,7 @@ 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"}}}
-    dump_yaml("dvc.lock", lockdata)
+    (tmp_dir / "dvc.lock").dump(lockdata)
     with pytest.raises(LockfileCorruptedError) as exc_info:
         Lockfile(dvc, tmp_dir / "dvc.lock").load()
 
diff --git a/tests/func/test_move.py b/tests/func/test_move.py
index 468a9421e8..6973c8577a 100644
--- a/tests/func/test_move.py
+++ b/tests/func/test_move.py
@@ -6,7 +6,7 @@
 from dvc.dvcfile import DVC_FILE_SUFFIX
 from dvc.exceptions import DvcException, MoveNotDataSourceError
 from dvc.main import main
-from dvc.utils.serialize import dump_yaml, load_yaml
+from dvc.utils.serialize import load_yaml
 from tests.basic_env import TestDvc, TestDvcGit
 from tests.func.test_repro import TestRepro
 from tests.utils import cd
@@ -249,9 +249,9 @@ def test_move_output_overlap(tmp_dir, dvc):
 
 def test_move_meta(tmp_dir, dvc):
     (stage,) = tmp_dir.dvc_gen("foo", "foo")
-    data = load_yaml(stage.path)
+    data = (tmp_dir / stage.path).parse()
     data["meta"] = {"custom_key": 42}
-    dump_yaml(stage.path, data)
+    (tmp_dir / stage.path).dump(data)
 
     dvc.move("foo", "bar")
     res = (tmp_dir / "bar.dvc").read_text()
diff --git a/tests/func/test_repo_index.py b/tests/func/test_repo_index.py
index 79cebfd0f7..31c981a6cf 100644
--- a/tests/func/test_repo_index.py
+++ b/tests/func/test_repo_index.py
@@ -7,7 +7,6 @@
 from dvc.repo.index import Index
 from dvc.stage import PipelineStage, Stage
 from dvc.utils import relpath
-from tests.func.plots.utils import _write_json
 
 
 def test_index(tmp_dir, scm, dvc, run_copy):
@@ -103,7 +102,9 @@ def test_deps_outs_getters(tmp_dir, dvc, run_copy_metrics):
         params=["param"],
         name="copy-metrics",
     )
-    _write_json(tmp_dir, [{"a": 1, "b": 2}, {"a": 2, "b": 3}], "metric_t.json")
+    (tmp_dir / "metric_t.json").dump_json(
+        [{"a": 1, "b": 2}, {"a": 2, "b": 3}], sort_keys=True
+    )
     run_stage2 = run_copy_metrics(
         "metric_t.json",
         "metric.json",
diff --git a/tests/func/test_repro.py b/tests/func/test_repro.py
index 46b4801231..4481ca00a3 100644
--- a/tests/func/test_repro.py
+++ b/tests/func/test_repro.py
@@ -1244,7 +1244,7 @@ def test_repro_when_cmd_changes(tmp_dir, dvc, run_copy, mocker):
 
     data = SingleStageFile(dvc, stage.path)._load()[0]
     data["cmd"] = "  ".join(stage.cmd.split())  # change cmd spacing by two
-    dump_yaml(stage.path, data)
+    (tmp_dir / stage.path).dump(data)
 
     assert dvc.status([stage.addressing]) == {
         stage.addressing: ["changed checksum"]
diff --git a/tests/func/test_repro_multistage.py b/tests/func/test_repro_multistage.py
index 306f3329a2..c700115e74 100644
--- a/tests/func/test_repro_multistage.py
+++ b/tests/func/test_repro_multistage.py
@@ -9,7 +9,6 @@
 from dvc.exceptions import CyclicGraphError, ReproductionError
 from dvc.main import main
 from dvc.stage import PipelineStage
-from dvc.utils.serialize import dump_yaml, load_yaml
 from tests.func import test_repro
 
 COPY_SCRIPT_FORMAT = dedent(
@@ -317,7 +316,7 @@ def test_repro_when_new_deps_is_added_in_dvcfile(tmp_dir, dvc, run_copy):
     dvcfile = Dvcfile(dvc, stage.path)
     data, _ = dvcfile._load()
     data["stages"]["copy-file"]["deps"] += ["copy.py"]
-    dump_yaml(stage.path, data)
+    (tmp_dir / stage.path).dump(data)
 
     assert dvc.reproduce(target)[0] == stage
 
@@ -339,7 +338,7 @@ def test_repro_when_new_outs_is_added_in_dvcfile(tmp_dir, dvc):
     dvcfile = Dvcfile(dvc, stage.path)
     data, _ = dvcfile._load()
     data["stages"]["copy-file"]["outs"] = ["foobar"]
-    dump_yaml(stage.path, data)
+    (tmp_dir / stage.path).dump(data)
 
     assert dvc.reproduce(target)[0] == stage
 
@@ -366,7 +365,7 @@ def test_repro_when_new_deps_is_moved(tmp_dir, dvc):
     dvcfile = Dvcfile(dvc, stage.path)
     data, _ = dvcfile._load()
     data["stages"]["copy-file"]["deps"] = ["bar"]
-    dump_yaml(stage.path, data)
+    (tmp_dir / stage.path).dump(data)
 
     assert dvc.reproduce(target)[0] == stage
 
@@ -376,8 +375,7 @@ def test_repro_when_new_out_overlaps_others_stage_outs(tmp_dir, dvc):
 
     tmp_dir.gen({"dir": {"file1": "file1"}, "foo": "foo"})
     dvc.add("dir")
-    dump_yaml(
-        PIPELINE_FILE,
+    (tmp_dir / PIPELINE_FILE).dump(
         {
             "stages": {
                 "run-copy": {
@@ -395,8 +393,7 @@ def test_repro_when_new_out_overlaps_others_stage_outs(tmp_dir, dvc):
 def test_repro_when_new_deps_added_does_not_exist(tmp_dir, dvc):
     tmp_dir.gen("copy.py", COPY_SCRIPT)
     tmp_dir.gen("foo", "foo")
-    dump_yaml(
-        PIPELINE_FILE,
+    (tmp_dir / PIPELINE_FILE).dump(
         {
             "stages": {
                 "run-copy": {
@@ -414,8 +411,7 @@ def test_repro_when_new_deps_added_does_not_exist(tmp_dir, dvc):
 def test_repro_when_new_outs_added_does_not_exist(tmp_dir, dvc):
     tmp_dir.gen("copy.py", COPY_SCRIPT)
     tmp_dir.gen("foo", "foo")
-    dump_yaml(
-        PIPELINE_FILE,
+    (tmp_dir / PIPELINE_FILE).dump(
         {
             "stages": {
                 "run-copy": {
@@ -433,8 +429,7 @@ def test_repro_when_new_outs_added_does_not_exist(tmp_dir, dvc):
 def test_repro_when_lockfile_gets_deleted(tmp_dir, dvc):
     tmp_dir.gen("copy.py", COPY_SCRIPT)
     tmp_dir.gen("foo", "foo")
-    dump_yaml(
-        PIPELINE_FILE,
+    (tmp_dir / PIPELINE_FILE).dump(
         {
             "stages": {
                 "run-copy": {
@@ -464,13 +459,13 @@ def test_cyclic_graph_error(tmp_dir, dvc, run_copy):
     run_copy("bar", "baz", name="copy-bar-baz")
     run_copy("baz", "foobar", name="copy-baz-foobar")
 
-    data = load_yaml(PIPELINE_FILE)
+    data = (tmp_dir / PIPELINE_FILE).parse()
     data["stages"]["copy-baz-foo"] = {
         "cmd": "echo baz > foo",
         "deps": ["baz"],
         "outs": ["foo"],
     }
-    dump_yaml(PIPELINE_FILE, data)
+    (tmp_dir / PIPELINE_FILE).dump(data)
     with pytest.raises(CyclicGraphError):
         dvc.reproduce(":copy-baz-foo")
 
@@ -479,8 +474,8 @@ def test_repro_multiple_params(tmp_dir, dvc):
     from dvc.stage.utils import split_params_deps
     from tests.func.test_run_multistage import supported_params
 
-    dump_yaml(tmp_dir / "params2.yaml", supported_params)
-    dump_yaml(tmp_dir / "params.yaml", supported_params)
+    (tmp_dir / "params2.yaml").dump(supported_params)
+    (tmp_dir / "params.yaml").dump(supported_params)
 
     (tmp_dir / "foo").write_text("foo")
     stage = dvc.run(
@@ -522,7 +517,7 @@ def test_repro_multiple_params(tmp_dir, dvc):
     assert not dvc.reproduce(stage.addressing)
     params = deepcopy(supported_params)
     params["answer"] = 43
-    dump_yaml(tmp_dir / "params.yaml", params)
+    (tmp_dir / "params.yaml").dump(params)
 
     assert dvc.reproduce(stage.addressing) == [stage]
 
@@ -533,7 +528,7 @@ def test_repro_list_of_commands_in_order(tmp_dir, dvc, multiline):
     if multiline:
         cmd = "\n".join(cmd)
 
-    dump_yaml("dvc.yaml", {"stages": {"multi": {"cmd": cmd}}})
+    (tmp_dir / "dvc.yaml").dump({"stages": {"multi": {"cmd": cmd}}})
 
     (tmp_dir / "dvc.yaml").write_text(
         dedent(
@@ -559,7 +554,7 @@ def test_repro_list_of_commands_raise_and_stops_after_failure(
     if multiline:
         cmd = "\n".join(cmd)
 
-    dump_yaml("dvc.yaml", {"stages": {"multi": {"cmd": cmd}}})
+    (tmp_dir / "dvc.yaml").dump({"stages": {"multi": {"cmd": cmd}}})
 
     with pytest.raises(ReproductionError):
         dvc.reproduce(targets=["multi"])
diff --git a/tests/func/test_run_multistage.py b/tests/func/test_run_multistage.py
index c5f855a0ee..a34e085ac7 100644
--- a/tests/func/test_run_multistage.py
+++ b/tests/func/test_run_multistage.py
@@ -5,7 +5,6 @@
 
 from dvc.exceptions import InvalidArgumentError
 from dvc.stage.exceptions import DuplicateStageName, InvalidStageName
-from dvc.utils.serialize import dump_yaml, parse_yaml_for_update
 
 
 def test_run_with_name(tmp_dir, dvc, run_copy):
@@ -232,7 +231,7 @@ def test_run_already_exists(tmp_dir, dvc, run_copy):
 def test_run_params_default(tmp_dir, dvc):
     from dvc.dependency import ParamsDependency
 
-    dump_yaml(tmp_dir / "params.yaml", supported_params)
+    (tmp_dir / "params.yaml").dump(supported_params)
     stage = dvc.run(
         name="read_params",
         params=["nested.nested1.nested2"],
@@ -255,7 +254,7 @@ def test_run_params_default(tmp_dir, dvc):
 def test_run_params_custom_file(tmp_dir, dvc):
     from dvc.dependency import ParamsDependency
 
-    dump_yaml(tmp_dir / "params2.yaml", supported_params)
+    (tmp_dir / "params2.yaml").dump(supported_params)
     stage = dvc.run(
         name="read_params",
         params=["params2.yaml:lists"],
@@ -278,7 +277,7 @@ def test_run_params_custom_file(tmp_dir, dvc):
 def test_run_params_no_exec(tmp_dir, dvc):
     from dvc.dependency import ParamsDependency
 
-    dump_yaml(tmp_dir / "params2.yaml", supported_params)
+    (tmp_dir / "params2.yaml").dump(supported_params)
     stage = dvc.run(
         name="read_params",
         params=["params2.yaml:lists"],
@@ -318,9 +317,7 @@ def test_run_overwrite_order(tmp_dir, dvc, run_copy):
 
     run_copy("foo1", "bar1", name="copy-foo-bar", force=True)
 
-    data = parse_yaml_for_update(
-        (tmp_dir / PIPELINE_FILE).read_text(), PIPELINE_FILE
-    )
+    data = (tmp_dir / PIPELINE_FILE).parse()
     assert list(data["stages"].keys()) == ["copy-foo-bar", "copy-bar-foobar"]
 
 
diff --git a/tests/func/test_stage.py b/tests/func/test_stage.py
index 12a708be7f..34612d2522 100644
--- a/tests/func/test_stage.py
+++ b/tests/func/test_stage.py
@@ -168,34 +168,34 @@ def test_meta_is_preserved(tmp_dir, dvc):
     (stage,) = tmp_dir.dvc_gen("foo", "foo content")
 
     # Add meta to DVC-file
-    data = load_yaml(stage.path)
+    data = (tmp_dir / stage.path).parse()
     data["meta"] = {"custom_key": 42}
-    dump_yaml(stage.path, data)
+    (tmp_dir / stage.path).dump(data)
 
     # Loading and dumping to test that it works and meta is retained
     dvcfile = SingleStageFile(dvc, stage.path)
     new_stage = dvcfile.stage
     dvcfile.dump(new_stage)
 
-    new_data = load_yaml(stage.path)
+    new_data = (tmp_dir / stage.path).parse()
     assert new_data["meta"] == data["meta"]
 
 
 def test_desc_is_preserved(tmp_dir, dvc):
     (stage,) = tmp_dir.dvc_gen("foo", "foo content")
 
-    data = load_yaml(stage.path)
+    data = (tmp_dir / stage.path).parse()
     stage_desc = "test stage description"
     out_desc = "test out description"
     data["desc"] = stage_desc
     data["outs"][0]["desc"] = out_desc
-    dump_yaml(stage.path, data)
+    (tmp_dir / stage.path).dump(data)
 
     dvcfile = SingleStageFile(dvc, stage.path)
     new_stage = dvcfile.stage
     dvcfile.dump(new_stage)
 
-    new_data = load_yaml(stage.path)
+    new_data = (tmp_dir / stage.path).parse()
     assert new_data["desc"] == stage_desc
     assert new_data["outs"][0]["desc"] == out_desc
 
diff --git a/tests/func/test_stage_load.py b/tests/func/test_stage_load.py
index 5264e15bf1..47e0f26672 100644
--- a/tests/func/test_stage_load.py
+++ b/tests/func/test_stage_load.py
@@ -16,7 +16,6 @@
 )
 from dvc.utils import relpath
 from dvc.utils.fs import remove
-from dvc.utils.serialize import dump_yaml
 
 
 def test_collect(tmp_dir, scm, dvc, run_copy):
@@ -149,7 +148,7 @@ def test_collect_generated(tmp_dir, dvc):
             "build": {"foreach": "${vars}", "do": {"cmd": "echo ${item}"}}
         },
     }
-    dump_yaml("dvc.yaml", d)
+    (tmp_dir / "dvc.yaml").dump(d)
 
     all_stages = set(dvc.index.stages)
     assert len(all_stages) == 5
@@ -443,7 +442,7 @@ def test_collect_repo_callback(tmp_dir, dvc, mocker):
     dvc.stage_collection_error_handler = mock
 
     (stage,) = tmp_dir.dvc_gen("foo", "foo")
-    dump_yaml(tmp_dir / PIPELINE_FILE, {"stages": {"cmd": "echo hello world"}})
+    (tmp_dir / PIPELINE_FILE).dump({"stages": {"cmd": "echo hello world"}})
 
     dvc._reset()
     assert dvc.index.stages == [stage]
diff --git a/tests/unit/render/test_vega.py b/tests/unit/render/test_vega.py
index 16700c5fd4..41002e38b6 100644
--- a/tests/unit/render/test_vega.py
+++ b/tests/unit/render/test_vega.py
@@ -305,7 +305,7 @@ def test_raise_on_no_template(tmp_dir, dvc):
 
 def test_bad_template(tmp_dir, dvc):
     metric = [{"val": 2}, {"val": 3}]
-    tmp_dir.gen("template.json", json.dumps({"a": "b", "c": "d"}))
+    (tmp_dir / "template.json").dump({"a": "b", "c": "d"})
     props = {"template": "template.json"}
     data = {
         "workspace": {"data": {"file.json": {"data": metric, "props": props}}}
diff --git a/tests/unit/test_context.py b/tests/unit/test_context.py
index ea8957504d..33661d557d 100644
--- a/tests/unit/test_context.py
+++ b/tests/unit/test_context.py
@@ -17,7 +17,7 @@
     recurse_not_a_node,
 )
 from dvc.utils import relpath
-from dvc.utils.serialize import dump_yaml, dumps_yaml
+from dvc.utils.serialize import dumps_yaml
 
 
 def test_context():
@@ -290,7 +290,7 @@ def test_track(tmp_dir):
     }
     fs = LocalFileSystem()
     path = tmp_dir / "params.yaml"
-    dump_yaml(path, d, fs)
+    path.dump(d, fs=fs)
 
     context = Context.load_from(fs, path)
 
@@ -326,8 +326,8 @@ def test_track_from_multiple_files(tmp_dir):
     fs = LocalFileSystem()
     path1 = tmp_dir / "params.yaml"
     path2 = tmp_dir / "params2.yaml"
-    dump_yaml(path1, d1, fs)
-    dump_yaml(path2, d2, fs)
+    path1.dump(d1, fs=fs)
+    path2.dump(d2, fs=fs)
 
     context = Context.load_from(fs, path1)
     c = Context.load_from(fs, path2)
diff --git a/tests/unit/test_dvcfile.py b/tests/unit/test_dvcfile.py
index dc578a687c..099d104436 100644
--- a/tests/unit/test_dvcfile.py
+++ b/tests/unit/test_dvcfile.py
@@ -15,7 +15,6 @@
     StageFileIsNotDvcFileError,
 )
 from dvc.utils.fs import remove
-from dvc.utils.serialize import dump_yaml
 
 
 @pytest.mark.parametrize(
@@ -66,7 +65,7 @@ def test_stage_load_on_non_file(tmp_dir, dvc, file):
 @pytest.mark.parametrize("file", ["stage.dvc", "dvc.yaml"])
 def test_stage_load_on_invalid_data(tmp_dir, dvc, file):
     data = {"is_this_a_valid_dvcfile": False}
-    dump_yaml(file, data)
+    (tmp_dir / file).dump(data)
     dvcfile = Dvcfile(dvc, file)
     with pytest.raises(StageFileFormatError):
         assert dvcfile.stages
diff --git a/tests/unit/test_lockfile.py b/tests/unit/test_lockfile.py
index 32055f8054..6c194e8637 100644
--- a/tests/unit/test_lockfile.py
+++ b/tests/unit/test_lockfile.py
@@ -2,7 +2,6 @@
 
 from dvc.dvcfile import FileIsGitIgnored, Lockfile, LockfileCorruptedError
 from dvc.stage import PipelineStage
-from dvc.utils.serialize import dump_yaml
 
 
 def test_stage_dump_no_outs_deps(tmp_dir, dvc):
@@ -17,7 +16,7 @@ def test_stage_dump_no_outs_deps(tmp_dir, dvc):
 
 def test_stage_dump_when_already_exists(tmp_dir, dvc):
     data = {"s1": {"cmd": "command", "deps": [], "outs": []}}
-    dump_yaml("path.lock", data)
+    (tmp_dir / "path.lock").dump(data)
     stage = PipelineStage(name="s2", repo=dvc, path="path", cmd="command2")
     lockfile = Lockfile(dvc, "path.lock")
     lockfile.dump(stage)
@@ -35,7 +34,7 @@ def test_stage_dump_with_deps_and_outs(tmp_dir, dvc):
             "outs": [{"md5": "2.txt", "path": "checksum"}],
         }
     }
-    dump_yaml("path.lock", data)
+    (tmp_dir / "path.lock").dump(data)
     lockfile = Lockfile(dvc, "path.lock")
     stage = PipelineStage(name="s2", repo=dvc, path="path", cmd="command2")
     lockfile.dump(stage)
@@ -78,7 +77,7 @@ def test_load_when_lockfile_does_not_exist(tmp_dir, dvc):
     ],
 )
 def test_load_when_lockfile_is_corrupted(tmp_dir, dvc, corrupt_data):
-    dump_yaml("Dvcfile.lock", corrupt_data)
+    (tmp_dir / "Dvcfile.lock").dump(corrupt_data)
     lockfile = Lockfile(dvc, "Dvcfile.lock")
     with pytest.raises(LockfileCorruptedError) as exc_info:
         lockfile.load()