Repository: ansible/ansible-navigator (language: python)

Issue:
Switch from Named to mkstemp
https://github.com/ansible/ansible-navigator/blob/b7d6fd223f9f55fb5d176b60a687782135a9ae4b/ansible_navigator/actions/open_file.py#L91


- credit @abadger 


So one issue is portability, there's a race where the filehandle opened by `tempfile.NamedTemporaryFile()` hasn't been closed yet when `open(filename, "w")` is called.  On systems where the file can't be opened twice at the same time, that will fail.

The other one is security.  Because of the same race, sometimes the filename will be deleted before `open(filename, "w")` is called.  That gives an attacker a brief window to put their own file, symlink, etc, in place.  IIRC, a symlink would allow the attacker to overwrite one of the user's file with the data, corrupting that file.


Discussion / hints:
This issue may address the pylint violations listed @ https://github.com/ansible/ansible-navigator/issues/539.
This function operates exactly as TemporaryFile() does, except that the file is guaranteed to have a visible name in the file system (on Unix, the directory entry is not unlinked). That name can be retrieved from the name attribute of the returned file-like object. Whether the name can be used to open the file a second time, while the named temporary file is still open, varies across platforms (it can be so used on Unix; it cannot on Windows NT or later). If delete is true (the default), the file is deleted as soon as it is closed. The returned object is always a file-like object whose file attribute is the underlying true file object. This file-like object can be used in a with statement, just like a normal file.

https://docs.python.org/3/library/tempfile.html#tempfile.NamedTemporaryFile

Fix patch:
diff --git a/docs/conf.py b/docs/conf.py
index a06aa7155..ef2b61738 100644
--- a/docs/conf.py
+++ b/docs/conf.py
@@ -147,7 +147,7 @@ def parse_with_fetch(*args, **kwargs) -> str:
     ("py:class", "State"),
     ("py:class", "WhileRule"),
     ("py:class", "Window"),
-    ("py:class", "yaml.cyaml.CDumper"),
+    ("py:class", "yaml.cyaml.CSafeDumper"),
     ("py:class", "yaml.nodes.ScalarNode"),
     ("py:obj", "ansible_navigator.steps.T"),
     ("py:obj", "ansible_navigator.tm_tokenize.fchainmap.TKey"),
diff --git a/src/ansible_navigator/actions/open_file.py b/src/ansible_navigator/actions/open_file.py
index aa799a046..d6d84bcb9 100644
--- a/src/ansible_navigator/actions/open_file.py
+++ b/src/ansible_navigator/actions/open_file.py
@@ -4,18 +4,21 @@
 import os
 import tempfile
 
+from pathlib import Path
 from typing import Any
 from typing import Callable
 from typing import Dict
 from typing import List
+from typing import Optional
 
 from ..app_public import AppPublic
 from ..configuration_subsystem import ApplicationConfiguration
+from ..ui_framework import ContentView
 from ..ui_framework import Interaction
 from ..ui_framework import Menu
 from ..utils.functions import remove_dbl_un
-from ..utils.serialize import human_dump
-from ..utils.serialize import json_dump
+from ..utils.serialize import SerializationFormat
+from ..utils.serialize import serialize_write_temp_file
 from . import _actions as actions
 
 
@@ -78,14 +81,15 @@ def run(self, interaction: Interaction, app: AppPublic) -> None:
         """
         self._logger.debug("open requested")
 
-        filename = None
+        filename: Optional[Path] = None
         line_number = 0
 
         something = interaction.action.match.groupdict()["something"]
         if something:
             parts = something.rsplit(":", 1)
-            if os.path.isfile(parts[0]):
-                filename = parts[0]
+            possible_filename = Path(parts[0])
+            if possible_filename.is_file():
+                filename = possible_filename
                 line_number = parts[1:][0] if parts[1:] else 0
             else:
                 obj = something
@@ -99,24 +103,33 @@ def run(self, interaction: Interaction, app: AppPublic) -> None:
 
         if not filename:
             if interaction.ui.serialization_format() == "text.html.markdown":
-                with tempfile.NamedTemporaryFile(suffix=".md", delete=False) as temp_file:
-                    filename = temp_file.name
-                    with open(filename, "w", encoding="utf-8") as fh:
-                        fh.write(obj)
+                with tempfile.NamedTemporaryFile(
+                    suffix=".md",
+                    delete=False,
+                    mode="w+t",
+                ) as file_like:
+                    filename = Path(file_like.name)
+                    file_like.write(obj)
             elif interaction.ui.serialization_format() == "source.yaml":
-                with tempfile.NamedTemporaryFile(suffix=".yaml", delete=False) as temp_file:
-                    filename = temp_file.name
-                    human_dump(obj=obj, filename=filename)
+                filename = serialize_write_temp_file(
+                    content=obj,
+                    content_view=ContentView.NORMAL,
+                    serialization_format=SerializationFormat.YAML,
+                )
             elif interaction.ui.serialization_format() == "source.json":
-                with tempfile.NamedTemporaryFile(suffix=".json", delete=False) as temp_file:
-                    filename = temp_file.name
-                    with open(filename, "w", encoding="utf-8") as fh:
-                        json_dump(obj, fh)
+                filename = serialize_write_temp_file(
+                    content=obj,
+                    content_view=ContentView.NORMAL,
+                    serialization_format=SerializationFormat.JSON,
+                )
             else:
-                with tempfile.NamedTemporaryFile(suffix=".txt", delete=False) as temp_file:
-                    filename = temp_file.name
-                    with open(filename, "w", encoding="utf-8") as fh:
-                        fh.write(obj)
+                with tempfile.NamedTemporaryFile(
+                    suffix=".txt",
+                    delete=False,
+                    mode="w+t",
+                ) as file_like:
+                    filename = Path(file_like.name)
+                    file_like.write(obj)
 
         command = self._args.editor_command.format(filename=filename, line_number=line_number)
         is_console = self._args.editor_console
diff --git a/src/ansible_navigator/actions/run.py b/src/ansible_navigator/actions/run.py
index eebf71bf7..0098d61cf 100644
--- a/src/ansible_navigator/actions/run.py
+++ b/src/ansible_navigator/actions/run.py
@@ -12,6 +12,7 @@
 import uuid
 
 from math import floor
+from pathlib import Path
 from queue import Queue
 from typing import Any
 from typing import Callable
@@ -20,6 +21,7 @@
 from typing import Optional
 from typing import Tuple
 
+from ansible_navigator.ui_framework.content_defs import ContentView
 from ..action_base import ActionBase
 from ..action_defs import RunStdoutReturn
 from ..app_public import AppPublic
@@ -38,7 +40,8 @@
 from ..utils.functions import human_time
 from ..utils.functions import remove_ansi
 from ..utils.functions import round_half_up
-from ..utils.serialize import json_dump
+from ..utils.serialize import SerializationFormat
+from ..utils.serialize import serialize_write_file
 from . import _actions as actions
 from . import run_action
 
@@ -849,16 +852,21 @@ def write_artifact(self, filename: Optional[str] = None) -> None:
 
             try:
                 os.makedirs(os.path.dirname(filename), exist_ok=True)
-                with open(filename, "w", encoding="utf-8") as fh:
-                    artifact = {
-                        "version": "1.0.0",
-                        "plays": self._plays.value,
-                        "stdout": self.stdout,
-                        "status": status,
-                        "status_color": status_color,
-                    }
-                    json_dump(artifact, fh)
-                    self._logger.info("Saved artifact as %s", filename)
+                artifact = {
+                    "version": "1.0.0",
+                    "plays": self._plays.value,
+                    "stdout": self.stdout,
+                    "status": status,
+                    "status_color": status_color,
+                }
+                serialize_write_file(
+                    content=artifact,
+                    content_view=ContentView.NORMAL,
+                    file_mode="w",
+                    file=Path(filename),
+                    serialization_format=SerializationFormat.JSON,
+                )
+                self._logger.info("Saved artifact as %s", filename)
 
             except (IOError, OSError) as exc:
                 error = (
diff --git a/src/ansible_navigator/actions/write_file.py b/src/ansible_navigator/actions/write_file.py
index 38bf2d58a..3569c7fad 100644
--- a/src/ansible_navigator/actions/write_file.py
+++ b/src/ansible_navigator/actions/write_file.py
@@ -3,12 +3,15 @@
 import os
 import re
 
+from pathlib import Path
+
+from ansible_navigator.ui_framework.content_defs import ContentView
 from ..app_public import AppPublic
 from ..configuration_subsystem import ApplicationConfiguration
 from ..ui_framework import Interaction
 from ..utils.functions import remove_dbl_un
-from ..utils.serialize import human_dump
-from ..utils.serialize import json_dump
+from ..utils.serialize import SerializationFormat
+from ..utils.serialize import serialize_write_file
 from . import _actions as actions
 
 
@@ -83,11 +86,20 @@ def run(self, interaction: Interaction, app: AppPublic) -> None:
             with open(os.path.abspath(filename), file_mode, encoding="utf-8") as fh:
                 fh.write(obj)
         elif write_as == "yaml":
-            human_dump(obj=obj, filename=filename, file_mode=file_mode)
+            serialize_write_file(
+                content=obj,
+                content_view=ContentView.NORMAL,
+                file=Path(filename),
+                file_mode=file_mode,
+                serialization_format=SerializationFormat.YAML,
+            )
         elif write_as == "json":
-            with open(os.path.abspath(filename), file_mode, encoding="utf-8") as fh:
-                json_dump(obj, fh)
-                fh.write("\n")
-
+            serialize_write_file(
+                content=obj,
+                content_view=ContentView.NORMAL,
+                file=Path(filename),
+                file_mode=file_mode,
+                serialization_format=SerializationFormat.JSON,
+            )
         self._logger.info("Wrote to '%s' with mode '%s' as '%s'", filename, file_mode, write_as)
         return None
diff --git a/src/ansible_navigator/ui_framework/ui.py b/src/ansible_navigator/ui_framework/ui.py
index 9611dee63..f778e615e 100644
--- a/src/ansible_navigator/ui_framework/ui.py
+++ b/src/ansible_navigator/ui_framework/ui.py
@@ -22,10 +22,11 @@
 from typing import Union
 
 from ..utils.functions import templar
-from ..utils.serialize import human_dump
-from ..utils.serialize import json_dumps
+from ..utils.serialize import SerializationFormat
+from ..utils.serialize import serialize
 from .colorize import Colorize
 from .colorize import rgb_to_ansi
+from .content_defs import ContentView
 from .curses_defs import CursesLine
 from .curses_defs import CursesLinePart
 from .curses_defs import CursesLines
@@ -510,10 +511,20 @@ def _serialize_color(self, obj: Any) -> CursesLines:
 
         if self.serialization_format() == "source.ansi":
             return self._colorizer.render_ansi(doc=obj)
+
+        content_view = ContentView.NORMAL if self._hide_keys else ContentView.FULL
         if self.serialization_format() == "source.yaml":
-            string = human_dump(obj)
+            string = serialize(
+                content_view=content_view,
+                content=obj,
+                serialization_format=SerializationFormat.YAML,
+            )
         elif self.serialization_format() == "source.json":
-            string = json_dumps(obj)
+            string = serialize(
+                content_view=content_view,
+                content=obj,
+                serialization_format=SerializationFormat.JSON,
+            )
         else:
             string = obj
 
diff --git a/src/ansible_navigator/utils/serialize.py b/src/ansible_navigator/utils/serialize.py
index 667417ffd..a603c1886 100644
--- a/src/ansible_navigator/utils/serialize.py
+++ b/src/ansible_navigator/utils/serialize.py
@@ -1,14 +1,18 @@
 """Abstractions for common serialization formats."""
 
 import json
+import logging
 import re
+import tempfile
 
+from dataclasses import is_dataclass
 from enum import Enum
 from pathlib import Path
 from typing import IO
 from typing import TYPE_CHECKING
 from typing import Any
 from typing import Dict
+from typing import List
 from typing import NamedTuple
 from typing import Optional
 from typing import Union
@@ -16,11 +20,14 @@
 import yaml
 
 
+logger = logging.getLogger(__name__)
+
+
 class SerializationFormat(Enum):
     """The serialization format."""
 
-    YAML = "yaml"
-    JSON = "json"
+    YAML = "YAML"
+    JSON = "JSON"
 
 
 if TYPE_CHECKING:
@@ -30,9 +37,9 @@ class SerializationFormat(Enum):
 
 # pylint: disable=unused-import
 try:
-    from yaml import CDumper as Dumper
+    from yaml import CSafeDumper as SafeDumper
 except ImportError:
-    from yaml import Dumper  # type: ignore[misc] # noqa: F401
+    from yaml import SafeDumper  # type: ignore[misc] # noqa: F401
 
 try:
     from yaml import CLoader as Loader
@@ -42,36 +49,127 @@ class SerializationFormat(Enum):
     from yaml import SafeLoader  # type: ignore[misc] # noqa: F401
 # pylint: enable=unused-import
 
+ContentType = Union["ContentBase", bool, Dict[str, Any], float, int, List[Any], str]
+
 
 def serialize(
+    content: ContentType,
     content_view: "ContentView",
-    content: "ContentBase",
     serialization_format: SerializationFormat,
-    file_mode: str = "w",
-    filename: Optional[Path] = None,
-) -> Optional[str]:
+) -> str:
     """Serialize a dataclass based on format and view.
 
-    :param content_view: The content view
     :param content: The content dataclass to serialize
+    :param content_view: The content view
     :param serialization_format: The serialization format
-    :param file_mode: The mode for the file operation
-    :param filename: A filename to write to
+    :raises ValueError: When serialization format is not recognized
     :returns: The serialized content
     """
-    content_as_dict = content.asdict(
+    dumpable = _prepare_content(
+        content=content,
         content_view=content_view,
         serialization_format=serialization_format,
     )
     if serialization_format == SerializationFormat.YAML:
-        if filename is None:
-            return yaml_dumps(obj=content_as_dict)
-        return yaml_dump(obj=content_as_dict, filename=filename, file_mode=file_mode)
+        return _yaml_dumps(dumpable=dumpable)
     if serialization_format == SerializationFormat.JSON:
-        if filename is None:
-            return json_dumps(content_as_dict)
-        return _json_dump(dumpable=content_as_dict, filename=filename, file_mode=file_mode)
-    return None
+        return _json_dumps(dumpable=dumpable)
+    raise ValueError("Unknown serialization format")
+
+
+def serialize_write_file(
+    content: ContentType,
+    content_view: "ContentView",
+    file_mode: str,
+    file: Path,
+    serialization_format: SerializationFormat,
+):
+    """Serialize and write content to a file.
+
+    :param content: The content to serialize
+    :param content_view: The content view
+    :param file_mode: The file mode for the file
+    :param file: The file to write to
+    :param serialization_format: The serialization format
+    :raises ValueError: When serialization format is not recognized
+    """
+    dumpable = _prepare_content(
+        content=content,
+        content_view=content_view,
+        serialization_format=serialization_format,
+    )
+    with file.open(mode=file_mode, encoding="utf-8") as file_handle:
+        if serialization_format == SerializationFormat.JSON:
+            _json_dump(dumpable=dumpable, file_handle=file_handle)
+            return
+        if serialization_format == SerializationFormat.YAML:
+            _yaml_dump(dumpable=dumpable, file_handle=file_handle)
+            return
+    raise ValueError("Unknown serialization format")
+
+
+def serialize_write_temp_file(
+    content: ContentType,
+    content_view: "ContentView",
+    serialization_format: SerializationFormat,
+) -> Path:
+    """Serialize and write content to a premanent temporary file.
+
+    :param content: The content to serialize
+    :param content_view: The content view
+    :param serialization_format: The serialization format
+    :raises ValueError: When serialization format is not recognized
+    :returns: A ``Path`` to the file written to
+    """
+    dumpable = _prepare_content(
+        content=content,
+        content_view=content_view,
+        serialization_format=serialization_format,
+    )
+    suffix = f".{serialization_format.value!s}"
+    with tempfile.NamedTemporaryFile(suffix=suffix, delete=False, mode="w+t") as file_like:
+        if serialization_format == SerializationFormat.JSON:
+            _json_dump(dumpable=dumpable, file_handle=file_like)
+            return Path(file_like.name)
+        if serialization_format == SerializationFormat.YAML:
+            _yaml_dump(dumpable=dumpable, file_handle=file_like)
+            return Path(file_like.name)
+    raise ValueError("Unknown serialization format")
+
+
+SERIALIZATION_FAILURE_MSG = (
+    "The requested content could not be converted to {serialization_format}.\n"
+    "The content was {content}\n"
+    "Please log an issue for this, it should not have happened\n"
+    "Error details: {exception_str}\n"
+)
+
+
+def _prepare_content(
+    content: ContentType,
+    content_view: "ContentView",
+    serialization_format: SerializationFormat,
+) -> ContentType:
+    if isinstance(content, (bool, dict, float, int, list, str)):
+        return content
+    if is_dataclass(content):
+        return content.asdict(
+            content_view=content_view,
+            serialization_format=serialization_format,
+        )
+
+    # This is a big chance, it suggests all content is
+    # one of ``ContentType``. the thinking is it's better
+    # to error here early, then return a str(content)
+    # Ideally this get caught in testing
+    value_error = "Content type not recognized"
+    error = SERIALIZATION_FAILURE_MSG.format(
+        content=str(content),
+        exception_str=value_error,
+        serialization_format=serialization_format.value,
+    )
+    error += f"Content view: {content_view}\n"
+    return error
 
 
 class JsonParams(NamedTuple):
@@ -82,36 +180,41 @@ class JsonParams(NamedTuple):
     ensure_ascii: bool = False
 
 
-def _json_dump(dumpable: Dict, filename: Path, file_mode: str):
-    """Create a file handle and dump json.
+def _json_dump(dumpable: ContentType, file_handle: IO) -> None:
+    """Serialize the dumpable to json and write to a file.
 
     :param dumpable: The object to dump
-    :param filename: The file name for the file
-    :param file_mode: The file mode for the operation
-    """
-    with filename.open(mode=file_mode, encoding="utf-8") as file_handle:
-        json_dump(dumpable=dumpable, file_handle=file_handle)
-
-
-def json_dump(dumpable: Any, file_handle: IO, params: NamedTuple = JsonParams()) -> None:
-    """Serialize and write the dumpable to a file.
-
-    :param dumpable: The object to serialize
     :param file_handle: The file handle to write to
-    :param params: Parameters to override the defaults
     """
-    json.dump(dumpable, file_handle, **params._asdict())
-
-
-def json_dumps(dumpable: Any, params: NamedTuple = JsonParams()) -> str:
+    try:
+        json.dump(dumpable, file_handle, **JsonParams()._asdict())
+        file_handle.write("\n")  # Add newline json does not
+    except TypeError as exc:
+        error_message = SERIALIZATION_FAILURE_MSG.format(
+            content=str(dumpable),
+            exception_str=str(exc),
+            serialization_format="JSON",
+        )
+        file_handle.write(error_message)
+        logger.error(error_message)
+
+
+def _json_dumps(dumpable: ContentType) -> str:
     """Serialize the dumpable to json.
 
     :param dumpable: The object to serialize
-    :param params: Parameters to override the defaults
     :returns: The object serialized
     """
-    string = json.dumps(dumpable, **params._asdict())
-    return string
+    try:
+        return json.dumps(dumpable, **JsonParams()._asdict())
+    except TypeError as exc:
+        error_message = SERIALIZATION_FAILURE_MSG.format(
+            content=str(dumpable),
+            exception_str=str(exc),
+            serialization_format="JSON",
+        )
+        logger.error(error_message)
+        return error_message
 
 
 class YamlStyle(NamedTuple):
@@ -122,38 +225,43 @@ class YamlStyle(NamedTuple):
     allow_unicode: bool = True
 
 
-def human_dump(
-    obj: Any,
-    filename: Optional[Union[Path, str]] = None,
-    file_mode: str = "w",
-) -> Optional[str]:
-    """Serialize an object to yaml.
-
-    This allows for the consistent representation across the application.
+def _yaml_dump(dumpable: ContentType, file_handle: IO):
+    """Serialize the dumpable to yaml and write to a file.
 
-    :param obj: The object to serialize
-    :param filename: The filename of the file in which the obj should be written
-    :param file_mode: The mode to use for file writing
-    :returns: Either the serialized obj or None if written to a file
+    :param dumpable: The object to serialize
+    :param file_handle: The file handle to write to
     """
-    dumper = HumanDumper
-    if filename is not None:
-        with open(filename, file_mode, encoding="utf-8") as fh:
-            yaml.dump(
-                obj,
-                fh,
-                Dumper=dumper,
-                **YamlStyle()._asdict(),
-            )
-        return None
-    return yaml.dump(obj, Dumper=dumper, **YamlStyle()._asdict())
-
+    try:
+        yaml.dump(dumpable, file_handle, Dumper=HumanDumper, **YamlStyle()._asdict())
+    except yaml.representer.RepresenterError as exc:
+        error_message = SERIALIZATION_FAILURE_MSG.format(
+            content=str(dumpable),
+            exception_str=str(exc),
+            serialization_format="YAML",
+        )
+        file_handle.write(error_message)
+        logger.error(error_message)
 
-yaml_dump = human_dump
-yaml_dumps = human_dump
 
+def _yaml_dumps(dumpable: ContentType):
+    """Serialize the dumpable to yaml.
 
-class HumanDumper(Dumper):
+    :param dumpable: The object to serialize
+    :return: The serialized object
+    """
+    try:
+        return yaml.dump(dumpable, Dumper=HumanDumper, **YamlStyle()._asdict())
+    except yaml.representer.RepresenterError as exc:
+        error_message = SERIALIZATION_FAILURE_MSG.format(
+            content=str(dumpable),
+            exception_str=str(exc),
+            serialization_format="YAML",
+        )
+        logger.error(error_message)
+        return error_message
+
+
+class HumanDumper(SafeDumper):
     # pylint: disable=too-many-ancestors
     """An instance of a pyyaml Dumper.
 


Test patch:
diff --git a/tests/unit/actions/test_run.py b/tests/unit/actions/test_run.py
index 742cd63a8..35de308d2 100644
--- a/tests/unit/actions/test_run.py
+++ b/tests/unit/actions/test_run.py
@@ -4,6 +4,7 @@
 import os
 
 from copy import deepcopy
+from pathlib import Path
 from queue import Queue
 from typing import Dict
 from typing import List
@@ -84,10 +85,9 @@ def id_from_data(value):
 
 @patch.dict("os.environ", {"HOME": "/home/test_user"})
 @patch("os.makedirs", return_value=True)
-@patch("builtins.open", new_callable=mock_open)
 @patch("ansible_navigator.actions.run.Action._get_status", return_value=(0, 0))
 @pytest.mark.parametrize("data", artifact_test_data, ids=id_from_data)
-def test_artifact_path(_mocked_get_status, mocked_open, _mocked_makedirs, caplog, data):
+def test_artifact_path(_mocked_get_status, _mocked_makedirs, caplog, data):
     """Test the building of the artifact filename given a filename or playbook"""
     caplog.set_level(logging.DEBUG)
 
@@ -105,13 +105,20 @@ def test_artifact_path(_mocked_get_status, mocked_open, _mocked_makedirs, caplog
     args.entry("playbook_artifact_enable").value.current = True
 
     run = action(args=args)
-    run.write_artifact(filename=data.filename)
+
+    opener = mock_open()
+
+    def mocked_open(self, *args, **kwargs):
+        return opener(self, *args, **kwargs)
+
+    with patch.object(Path, "open", mocked_open):
+        run.write_artifact(filename=data.filename)
 
     if data.help_playbook is not True:
-        open_filename = mocked_open.call_args[0][0]
+        open_filename = str(opener.call_args[0][0])
         assert open_filename.startswith(data.expected), caplog.text
     else:
-        mocked_open.assert_not_called()
+        opener.assert_not_called()
 
 
 class RunRunnerTstData(NamedTuple):
diff --git a/tests/unit/ui_framework/test_colorize.py b/tests/unit/ui_framework/test_colorize.py
index 39dd93354..b3fb8f6f4 100644
--- a/tests/unit/ui_framework/test_colorize.py
+++ b/tests/unit/ui_framework/test_colorize.py
@@ -1,12 +1,15 @@
 """tests for colorize
 """
-import json
 import os
 
+from typing import Dict
+from typing import NamedTuple
 from unittest.mock import patch
 
 from ansible_navigator.ui_framework.colorize import Colorize
-from ansible_navigator.utils.serialize import human_dump
+from ansible_navigator.ui_framework.content_defs import ContentView
+from ansible_navigator.utils.serialize import SerializationFormat
+from ansible_navigator.utils.serialize import serialize
 
 
 SHARE_DIR = os.path.abspath(
@@ -16,17 +19,31 @@
 GRAMMAR_DIR = os.path.join(SHARE_DIR, "grammar")
 
 
+class Sample(NamedTuple):
+    """Sample data for colorize tests."""
+
+    serialization_format: SerializationFormat
+    content: Dict[str, str] = {"test": "data"}
+    content_view: ContentView = ContentView.NORMAL
+
+
+SAMPLE_JSON = Sample(serialization_format=SerializationFormat.JSON)._asdict()
+SAMPLE_YAML = Sample(serialization_format=SerializationFormat.YAML)._asdict()
+
+
 def test_basic_success_json():
     """Ensure the json string is returned as 1 lines, 5 parts and can be reassembled
     to the json string"""
-    sample = json.dumps({"test": "data"})
-    result = Colorize(grammar_dir=GRAMMAR_DIR, theme_path=THEME_PATH).render(
+    sample = serialize(**SAMPLE_JSON)
+    colorized = Colorize(grammar_dir=GRAMMAR_DIR, theme_path=THEME_PATH).render(
         doc=sample,
         scope="source.json",
     )
-    assert len(result) == 1
-    assert len(result[0]) == 5
-    assert "".join(line_part.chars for line_part in result[0]) == sample
+    assert len(colorized) == 3
+    serialized_lines = '{\n    "test": "data"\n}'.splitlines()
+    colorized_lines = ["".join(part.chars for part in line) for line in colorized]
+    assert serialized_lines == colorized_lines
+    assert "\n".join(serialized_lines) == sample
 
 
 def test_basic_success_yaml():
@@ -34,7 +51,7 @@ def test_basic_success_yaml():
     respectively, ensure the parts of the second line can be reassembled to
     the second line of the yaml string
     """
-    sample = human_dump({"test": "data"})
+    sample = serialize(**SAMPLE_YAML)
     result = Colorize(grammar_dir=GRAMMAR_DIR, theme_path=THEME_PATH).render(
         doc=sample,
         scope="source.yaml",
@@ -66,16 +83,12 @@ def test_basic_success_log():
 
 def test_basic_success_no_color():
     """Ensure scope ``no_color`` return just lines."""
-    sample = json.dumps({"test": "data"})
-    result = Colorize(grammar_dir=GRAMMAR_DIR, theme_path=THEME_PATH).render(
+    sample = serialize(**SAMPLE_JSON)
+    colorized = Colorize(grammar_dir=GRAMMAR_DIR, theme_path=THEME_PATH).render(
         doc=sample,
         scope="no_color",
     )
-    assert len(result) == 1
-    assert len(result[0]) == 1
-    assert result[0][0].chars == sample
-    assert result[0][0].color is None
-    assert result[0][0].column == 0
+    assert not any(part.color for line in colorized for part in line)
 
 
 @patch("ansible_navigator.ui_framework.colorize.tokenize")
@@ -84,14 +97,10 @@ def test_graceful_failure(mocked_func, caplog):
     w/o color and the log reflects the critical error
     """
     mocked_func.side_effect = ValueError()
-    sample = json.dumps({"test": "data"})
-    result = Colorize(grammar_dir=GRAMMAR_DIR, theme_path=THEME_PATH).render(
+    sample = serialize(**SAMPLE_JSON)
+
+    _result = Colorize(grammar_dir=GRAMMAR_DIR, theme_path=THEME_PATH).render(
         doc=sample,
         scope="source.json",
     )
-    assert len(result) == 1
-    assert len(result[0]) == 1
-    assert result[0][0].chars == sample
-    assert result[0][0].color is None
-    assert result[0][0].column == 0
     assert "rendered without color" in caplog.text
diff --git a/tests/unit/utils/test_serialize_yaml.py b/tests/unit/utils/test_serialize_yaml.py
index b14c30461..076dfb8d9 100644
--- a/tests/unit/utils/test_serialize_yaml.py
+++ b/tests/unit/utils/test_serialize_yaml.py
@@ -4,14 +4,15 @@
 
 from yaml import Dumper
 
-import ansible_navigator.utils.serialize as yaml_import
+from ansible_navigator.ui_framework import ContentView
+from ansible_navigator.utils import serialize
 
 
 def test_check_yaml_imports():
     """Ensure yaml, Dumper, and Loader are imported without error."""
-    assert yaml_import.yaml is not None
-    assert yaml_import.Dumper is not None
-    assert yaml_import.Loader is not None
+    assert serialize.yaml is not None
+    assert serialize.SafeDumper is not None
+    assert serialize.Loader is not None
 
 
 @pytest.fixture(name="aliases_allowed")
@@ -43,7 +44,11 @@ def test_anchor_alias(aliases_allowed: bool):
     """
     data = {"integer": 42}
     many_data = [data, data, data]
-    yaml_string = yaml_import.human_dump(many_data)
+    yaml_string = serialize.serialize(
+        content=many_data,
+        content_view=ContentView.NORMAL,
+        serialization_format=serialize.SerializationFormat.YAML,
+    )
     assert isinstance(yaml_string, str), "Unexpected value from human_dump"
     found_alias = "&id" in yaml_string
     assert found_alias is aliases_allowed
diff --git a/tests/unit/utils/test_unserializable.py b/tests/unit/utils/test_unserializable.py
new file mode 100644
index 000000000..b862f5db7
--- /dev/null
+++ b/tests/unit/utils/test_unserializable.py
@@ -0,0 +1,75 @@
+"""Tests for content that cannot be serialized."""
+from typing import Tuple
+
+import pytest
+
+from ansible_navigator.ui_framework.content_defs import ContentView
+from ansible_navigator.utils.serialize import SerializationFormat
+from ansible_navigator.utils.serialize import serialize
+
+
+content_views = pytest.mark.parametrize(
+    argnames="content_view",
+    argvalues=ContentView.__members__.items(),
+    ids=ContentView.__members__,
+)
+
+serialization_formats = pytest.mark.parametrize(
+    argnames="serialization_format",
+    argvalues=SerializationFormat.__members__.items(),
+    ids=SerializationFormat.__members__,
+)
+
+
+@serialization_formats
+@content_views
+def test_custom_class(
+    content_view: Tuple[str, ContentView],
+    serialization_format: Tuple[str, SerializationFormat],
+):
+    """Ensure an error is provided when something can't be serialized.
+
+    A typing error does not exist here because the content is Dict[str, Any].
+
+    :param content_view: The content view
+    :param serialization_format: The serialization format
+    """
+
+    class CustomClass:
+        """An empty custom class."""
+
+    content = {"foo": CustomClass()}
+    serialized = serialize(
+        content=content,
+        content_view=content_view[1],
+        serialization_format=serialization_format[1],
+    )
+    assert (
+        f"The requested content could not be converted to {serialization_format[0]!s}."
+        in serialized
+    )
+
+
+@serialization_formats
+@content_views
+def test_tuple(
+    content_view: Tuple[str, ContentView],
+    serialization_format: Tuple[str, SerializationFormat],
+):
+    """Ensure an error is provided when something can't be serialized.
+
+    A typing error exists here, because tuple is not a ``utils.serialize.ContentType``.
+
+    :param content_view: The content view
+    :param serialization_format: The serialization format
+    """
+    content = (1, 2, 3)
+    serialized = serialize(
+        content=content,  # type:ignore[arg-type]
+        content_view=content_view[1],
+        serialization_format=serialization_format[1],
+    )
+    assert (
+        f"The requested content could not be converted to {serialization_format[0]!s}."
+        in serialized
+    )