[stable-2.19] Move display internals into _internal (#85889) (#85896)

(cherry picked from commit 58bad71859)

changelogs/fragments/display_internals.yml

@@ -0,0 +1,3 @@
+bugfixes:
+  - display - Fixed reference to undefined `_DeferredWarningContext` when issuing early warnings during startup.
+    (https://github.com/ansible/ansible/issues/85886)

lib/ansible/_internal/_display_utils.py

@@ -0,0 +1,145 @@
+from __future__ import annotations
+
+import dataclasses
+
+from ansible.module_utils._internal import _ambient_context, _messages
+from . import _event_formatting
+
+
+class DeferredWarningContext(_ambient_context.AmbientContextBase):
+    """
+    Calls to `Display.warning()` and `Display.deprecated()` within this context will cause the resulting warnings to be captured and not displayed.
+    The intended use is for task-initiated warnings to be recorded with the task result, which makes them visible to registered results, callbacks, etc.
+    The active display callback is responsible for communicating any warnings to the user.
+    """
+
+    # DTFIX-FUTURE: once we start implementing nested scoped contexts for our own bookkeeping, this should be an interface facade that forwards to the nearest
+    #               context that actually implements the warnings collection capability
+
+    def __init__(self, *, variables: dict[str, object]) -> None:
+        self._variables = variables  # DTFIX-FUTURE: move this to an AmbientContext-derived TaskContext (once it exists)
+        self._deprecation_warnings: list[_messages.DeprecationSummary] = []
+        self._warnings: list[_messages.WarningSummary] = []
+        self._seen: set[_messages.WarningSummary] = set()
+
+    def capture(self, warning: _messages.WarningSummary) -> None:
+        """Add the warning/deprecation to the context if it has not already been seen by this context."""
+        if warning in self._seen:
+            return
+
+        self._seen.add(warning)
+
+        if isinstance(warning, _messages.DeprecationSummary):
+            self._deprecation_warnings.append(warning)
+        else:
+            self._warnings.append(warning)
+
+    def get_warnings(self) -> list[_messages.WarningSummary]:
+        """Return a list of the captured non-deprecation warnings."""
+        # DTFIX-FUTURE: return a read-only list proxy instead
+        return self._warnings
+
+    def get_deprecation_warnings(self) -> list[_messages.DeprecationSummary]:
+        """Return a list of the captured deprecation warnings."""
+        # DTFIX-FUTURE: return a read-only list proxy instead
+        return self._deprecation_warnings
+
+
+def format_message(summary: _messages.SummaryBase, include_traceback: bool) -> str:
+    if isinstance(summary, _messages.DeprecationSummary):
+        deprecation_message = get_deprecation_message_with_plugin_info(
+            msg=summary.event.msg,
+            version=summary.version,
+            date=summary.date,
+            deprecator=summary.deprecator,
+        )
+
+        event = dataclasses.replace(summary.event, msg=deprecation_message)
+    else:
+        event = summary.event
+
+    return _event_formatting.format_event(event, include_traceback)
+
+
+def get_deprecation_message_with_plugin_info(
+    *,
+    msg: str,
+    version: str | None,
+    removed: bool = False,
+    date: str | None,
+    deprecator: _messages.PluginInfo | None,
+) -> str:
+    """Internal use only. Return a deprecation message and help text for display."""
+    # DTFIX-FUTURE: the logic for omitting date/version doesn't apply to the payload, so it shows up in vars in some cases when it should not
+
+    if removed:
+        removal_fragment = 'This feature was removed'
+    else:
+        removal_fragment = 'This feature will be removed'
+
+    if not deprecator or not deprecator.type:
+        # indeterminate has no resolved_name or type
+        # collections have a resolved_name but no type
+        collection = deprecator.resolved_name if deprecator else None
+        plugin_fragment = ''
+    elif deprecator.resolved_name == 'ansible.builtin':
+        # core deprecations from base classes (the API) have no plugin name, only 'ansible.builtin'
+        plugin_type_name = str(deprecator.type) if deprecator.type is _messages.PluginType.MODULE else f'{deprecator.type} plugin'
+
+        collection = deprecator.resolved_name
+        plugin_fragment = f'the {plugin_type_name} API'
+    else:
+        parts = deprecator.resolved_name.split('.')
+        plugin_name = parts[-1]
+        plugin_type_name = str(deprecator.type) if deprecator.type is _messages.PluginType.MODULE else f'{deprecator.type} plugin'
+
+        collection = '.'.join(parts[:2]) if len(parts) > 2 else None
+        plugin_fragment = f'{plugin_type_name} {plugin_name!r}'
+
+    if collection and plugin_fragment:
+        plugin_fragment += ' in'
+
+    if collection == 'ansible.builtin':
+        collection_fragment = 'ansible-core'
+    elif collection:
+        collection_fragment = f'collection {collection!r}'
+    else:
+        collection_fragment = ''
+
+    if not collection:
+        when_fragment = 'in the future' if not removed else ''
+    elif date:
+        when_fragment = f'in a release after {date}'
+    elif version:
+        when_fragment = f'version {version}'
+    else:
+        when_fragment = 'in a future release' if not removed else ''
+
+    if plugin_fragment or collection_fragment:
+        from_fragment = 'from'
+    else:
+        from_fragment = ''
+
+    deprecation_msg = ' '.join(f for f in [removal_fragment, from_fragment, plugin_fragment, collection_fragment, when_fragment] if f) + '.'
+
+    return join_sentences(msg, deprecation_msg)
+
+
+def join_sentences(first: str | None, second: str | None) -> str:
+    """Join two sentences together."""
+    first = (first or '').strip()
+    second = (second or '').strip()
+
+    if first and first[-1] not in ('!', '?', '.'):
+        first += '.'
+
+    if second and second[-1] not in ('!', '?', '.'):
+        second += '.'
+
+    if first and not second:
+        return first
+
+    if not first and second:
+        return second
+
+    return ' '.join((first, second))

lib/ansible/executor/task_executor.py

@@ -19,6 +19,8 @@ from ansible.errors import (
     AnsibleError, AnsibleParserError, AnsibleUndefinedVariable, AnsibleTaskError,
     AnsibleValueOmittedError,
 )
+
+from ansible._internal import _display_utils
 from ansible.executor.task_result import _RawTaskResult
 from ansible._internal._datatag import _utils
 from ansible.module_utils._internal import _messages
@@ -35,7 +37,7 @@ from ansible._internal._templating._jinja_plugins import _invoke_lookup, _Direct
 from ansible._internal._templating._engine import TemplateEngine
 from ansible.template import Templar
 from ansible.utils.collection_loader import AnsibleCollectionConfig
-from ansible.utils.display import Display, _DeferredWarningContext
+from ansible.utils.display import Display
 from ansible.utils.vars import combine_vars
 from ansible.vars.clean import namespace_facts, clean_facts
 from ansible.vars.manager import _deprecate_top_level_fact
@@ -416,7 +418,7 @@ class TaskExecutor:
     def _execute(self, templar: TemplateEngine, variables: dict[str, t.Any]) -> dict[str, t.Any]:
         result: dict[str, t.Any]
 
-        with _DeferredWarningContext(variables=variables) as warning_ctx:
+        with _display_utils.DeferredWarningContext(variables=variables) as warning_ctx:
             try:
                 # DTFIX-FUTURE: improve error handling to prioritize the earliest exception, turning the remaining ones into warnings
                 result = self._execute_internal(templar, variables)
@@ -431,7 +433,7 @@ class TaskExecutor:
 
             self._task.update_result_no_log(templar, result)
 
-        # The warnings/deprecations in the result have already been captured in the _DeferredWarningContext by _apply_task_result_compat.
+        # The warnings/deprecations in the result have already been captured in the DeferredWarningContext by _apply_task_result_compat.
         # The captured warnings/deprecations are a superset of the ones from the result, and may have been converted from a dict to a dataclass.
         # These are then used to supersede the entries in the result.
 
@@ -788,7 +790,7 @@ class TaskExecutor:
         return result
 
     @staticmethod
-    def _apply_task_result_compat(result: dict[str, t.Any], warning_ctx: _DeferredWarningContext) -> None:
+    def _apply_task_result_compat(result: dict[str, t.Any], warning_ctx: _display_utils.DeferredWarningContext) -> None:
         """Apply backward-compatibility mutations to the supplied task result."""
         if warnings := result.get('warnings'):
             if isinstance(warnings, list):

lib/ansible/plugins/loader.py

@@ -36,6 +36,7 @@ from ansible.utils.collection_loader._collection_finder import _AnsibleCollectio
 from ansible.utils.display import Display
 from ansible.utils.plugin_docs import add_fragments
 from ansible._internal._datatag import _tags
+from ansible._internal import _display_utils
 
 from . import _AnsiblePluginInfoMixin
 from .filter import AnsibleJinja2Filter
@@ -606,7 +607,7 @@ class PluginLoader:
                 warning_text = tombstone.get('warning_text') or ''
                 warning_plugin_type = "module" if self.type == "modules" else f'{self.type} plugin'
                 warning_text = f'The {fq_name!r} {warning_plugin_type} has been removed.{" " if warning_text else ""}{warning_text}'
-                removed_msg = display._get_deprecation_message_with_plugin_info(
+                removed_msg = _display_utils.get_deprecation_message_with_plugin_info(
                     msg=warning_text,
                     version=removal_version,
                     date=removal_date,
@@ -1411,7 +1412,7 @@ class Jinja2Loader(PluginLoader):
                 removal_version = tombstone_entry.get('removal_version')
                 warning_text = f'The {key!r} {self.type} plugin has been removed.{" " if warning_text else ""}{warning_text}'
 
-                exc_msg = display._get_deprecation_message_with_plugin_info(
+                exc_msg = _display_utils.get_deprecation_message_with_plugin_info(
                     msg=warning_text,
                     version=removal_version,
                     date=removal_date,

lib/ansible/utils/display.py

@@ -18,7 +18,6 @@
 from __future__ import annotations
 
 import contextlib
-import dataclasses
 
 try:
     import curses
@@ -52,8 +51,8 @@ from ansible import constants as C
 from ansible.constants import config
 from ansible.errors import AnsibleAssertionError, AnsiblePromptInterrupt, AnsiblePromptNoninteractive, AnsibleError
 from ansible._internal._errors import _error_utils, _error_factory
-from ansible._internal import _event_formatting
-from ansible.module_utils._internal import _ambient_context, _deprecator, _messages
+from ansible._internal import _display_utils
+from ansible.module_utils._internal import _deprecator, _messages
 from ansible.module_utils.common.text.converters import to_bytes, to_text
 from ansible.module_utils.datatag import deprecator_from_collection_name
 from ansible._internal._datatag._tags import TrustedAsTemplate
@@ -100,6 +99,17 @@ def _is_controller_traceback_enabled(event: _traceback.TracebackEvent) -> bool:
 _traceback._is_traceback_enabled = _is_controller_traceback_enabled
 
 
+def _deprecation_warnings_enabled() -> bool:
+    """Return True if deprecation warnings are enabled for the current calling context, otherwise False."""
+    # DTFIX-FUTURE: move this capability into config using an AmbientContext-derived TaskContext (once it exists)
+    if warning_ctx := _display_utils.DeferredWarningContext.current(optional=True):
+        variables = warning_ctx._variables
+    else:
+        variables = None
+
+    return C.config.get_config_value('DEPRECATION_WARNINGS', variables=variables)
+
+
 def get_text_width(text: str) -> int:
     """Function that utilizes ``wcswidth`` or ``wcwidth`` to determine the
     number of columns used to display a text string.
@@ -582,7 +592,7 @@ class Display(metaclass=Singleton):
             version="2.23",
         )
 
-        msg = self._get_deprecation_message_with_plugin_info(
+        msg = _display_utils.get_deprecation_message_with_plugin_info(
             msg=msg,
             version=version,
             removed=removed,
@@ -597,70 +607,6 @@ class Display(metaclass=Singleton):
 
         return msg
 
-    def _get_deprecation_message_with_plugin_info(
-        self,
-        *,
-        msg: str,
-        version: str | None,
-        removed: bool = False,
-        date: str | None,
-        deprecator: _messages.PluginInfo | None,
-    ) -> str:
-        """Internal use only. Return a deprecation message and help text for display."""
-        # DTFIX-FUTURE: the logic for omitting date/version doesn't apply to the payload, so it shows up in vars in some cases when it should not
-
-        if removed:
-            removal_fragment = 'This feature was removed'
-        else:
-            removal_fragment = 'This feature will be removed'
-
-        if not deprecator or not deprecator.type:
-            # indeterminate has no resolved_name or type
-            # collections have a resolved_name but no type
-            collection = deprecator.resolved_name if deprecator else None
-            plugin_fragment = ''
-        elif deprecator.resolved_name == 'ansible.builtin':
-            # core deprecations from base classes (the API) have no plugin name, only 'ansible.builtin'
-            plugin_type_name = str(deprecator.type) if deprecator.type is _messages.PluginType.MODULE else f'{deprecator.type} plugin'
-
-            collection = deprecator.resolved_name
-            plugin_fragment = f'the {plugin_type_name} API'
-        else:
-            parts = deprecator.resolved_name.split('.')
-            plugin_name = parts[-1]
-            plugin_type_name = str(deprecator.type) if deprecator.type is _messages.PluginType.MODULE else f'{deprecator.type} plugin'
-
-            collection = '.'.join(parts[:2]) if len(parts) > 2 else None
-            plugin_fragment = f'{plugin_type_name} {plugin_name!r}'
-
-        if collection and plugin_fragment:
-            plugin_fragment += ' in'
-
-        if collection == 'ansible.builtin':
-            collection_fragment = 'ansible-core'
-        elif collection:
-            collection_fragment = f'collection {collection!r}'
-        else:
-            collection_fragment = ''
-
-        if not collection:
-            when_fragment = 'in the future' if not removed else ''
-        elif date:
-            when_fragment = f'in a release after {date}'
-        elif version:
-            when_fragment = f'version {version}'
-        else:
-            when_fragment = 'in a future release' if not removed else ''
-
-        if plugin_fragment or collection_fragment:
-            from_fragment = 'from'
-        else:
-            from_fragment = ''
-
-        deprecation_msg = ' '.join(f for f in [removal_fragment, from_fragment, plugin_fragment, collection_fragment, when_fragment] if f) + '.'
-
-        return _join_sentences(msg, deprecation_msg)
-
     @staticmethod
     def _deduplicate(msg: str, messages: set[str]) -> bool:
         """
@@ -729,7 +675,7 @@ class Display(metaclass=Singleton):
         _skip_stackwalk = True
 
         if removed:
-            formatted_msg = self._get_deprecation_message_with_plugin_info(
+            formatted_msg = _display_utils.get_deprecation_message_with_plugin_info(
                 msg=msg,
                 version=version,
                 removed=removed,
@@ -756,7 +702,7 @@ class Display(metaclass=Singleton):
             deprecator=deprecator,
         )
 
-        if warning_ctx := _DeferredWarningContext.current(optional=True):
+        if warning_ctx := _display_utils.DeferredWarningContext.current(optional=True):
             warning_ctx.capture(deprecation)
             return
 
@@ -769,12 +715,12 @@ class Display(metaclass=Singleton):
         # This is the post-proxy half of the `deprecated` implementation.
         # Any logic that must occur in the primary controller process needs to be implemented here.
 
-        if not _DeferredWarningContext.deprecation_warnings_enabled():
+        if not _deprecation_warnings_enabled():
             return
 
         self.warning('Deprecation warnings can be disabled by setting `deprecation_warnings=False` in ansible.cfg.')
 
-        msg = _format_message(warning, _traceback.is_traceback_enabled(_traceback.TracebackEvent.DEPRECATED))
+        msg = _display_utils.format_message(warning, _traceback.is_traceback_enabled(_traceback.TracebackEvent.DEPRECATED))
         msg = f'[DEPRECATION WARNING]: {msg}'
 
         if self._deduplicate(msg, self._deprecations):
@@ -812,7 +758,7 @@ class Display(metaclass=Singleton):
             ),
         )
 
-        if warning_ctx := _DeferredWarningContext.current(optional=True):
+        if warning_ctx := _display_utils.DeferredWarningContext.current(optional=True):
             warning_ctx.capture(warning)
             return
 
@@ -825,7 +771,7 @@ class Display(metaclass=Singleton):
         # This is the post-proxy half of the `warning` implementation.
         # Any logic that must occur in the primary controller process needs to be implemented here.
 
-        msg = _format_message(warning, _traceback.is_traceback_enabled(_traceback.TracebackEvent.WARNING))
+        msg = _display_utils.format_message(warning, _traceback.is_traceback_enabled(_traceback.TracebackEvent.WARNING))
         msg = f"[WARNING]: {msg}"
 
         if self._deduplicate(msg, self._warns):
@@ -915,7 +861,7 @@ class Display(metaclass=Singleton):
             event=event,
         )
 
-        if warning_ctx := _DeferredWarningContext.current(optional=True):
+        if warning_ctx := _display_utils.DeferredWarningContext.current(optional=True):
             warning_ctx.capture(warning)
             return
 
@@ -952,7 +898,7 @@ class Display(metaclass=Singleton):
         # This is the post-proxy half of the `error` implementation.
         # Any logic that must occur in the primary controller process needs to be implemented here.
 
-        msg = _format_message(error, _traceback.is_traceback_enabled(_traceback.TracebackEvent.ERROR))
+        msg = _display_utils.format_message(error, _traceback.is_traceback_enabled(_traceback.TracebackEvent.ERROR))
         msg = f'[ERROR]: {msg}'
 
         if self._deduplicate(msg, self._errors):
@@ -1173,92 +1119,6 @@ class Display(metaclass=Singleton):
 _display = Display()
 
 
-class _DeferredWarningContext(_ambient_context.AmbientContextBase):
-    """
-    Calls to `Display.warning()` and `Display.deprecated()` within this context will cause the resulting warnings to be captured and not displayed.
-    The intended use is for task-initiated warnings to be recorded with the task result, which makes them visible to registered results, callbacks, etc.
-    The active display callback is responsible for communicating any warnings to the user.
-    """
-
-    # DTFIX-FUTURE: once we start implementing nested scoped contexts for our own bookkeeping, this should be an interface facade that forwards to the nearest
-    #               context that actually implements the warnings collection capability
-
-    def __init__(self, *, variables: dict[str, object]) -> None:
-        self._variables = variables  # DTFIX-FUTURE: move this to an AmbientContext-derived TaskContext (once it exists)
-        self._deprecation_warnings: list[_messages.DeprecationSummary] = []
-        self._warnings: list[_messages.WarningSummary] = []
-        self._seen: set[_messages.WarningSummary] = set()
-
-    @classmethod
-    def deprecation_warnings_enabled(cls) -> bool:
-        """Return True if deprecation warnings are enabled for the current calling context, otherwise False."""
-        # DTFIX-FUTURE: move this capability into config using an AmbientContext-derived TaskContext (once it exists)
-        if warning_ctx := cls.current(optional=True):
-            variables = warning_ctx._variables
-        else:
-            variables = None
-
-        return C.config.get_config_value('DEPRECATION_WARNINGS', variables=variables)
-
-    def capture(self, warning: _messages.WarningSummary) -> None:
-        """Add the warning/deprecation to the context if it has not already been seen by this context."""
-        if warning in self._seen:
-            return
-
-        self._seen.add(warning)
-
-        if isinstance(warning, _messages.DeprecationSummary):
-            self._deprecation_warnings.append(warning)
-        else:
-            self._warnings.append(warning)
-
-    def get_warnings(self) -> list[_messages.WarningSummary]:
-        """Return a list of the captured non-deprecation warnings."""
-        # DTFIX-FUTURE: return a read-only list proxy instead
-        return self._warnings
-
-    def get_deprecation_warnings(self) -> list[_messages.DeprecationSummary]:
-        """Return a list of the captured deprecation warnings."""
-        # DTFIX-FUTURE: return a read-only list proxy instead
-        return self._deprecation_warnings
-
-
-def _join_sentences(first: str | None, second: str | None) -> str:
-    """Join two sentences together."""
-    first = (first or '').strip()
-    second = (second or '').strip()
-
-    if first and first[-1] not in ('!', '?', '.'):
-        first += '.'
-
-    if second and second[-1] not in ('!', '?', '.'):
-        second += '.'
-
-    if first and not second:
-        return first
-
-    if not first and second:
-        return second
-
-    return ' '.join((first, second))
-
-
-def _format_message(summary: _messages.SummaryBase, include_traceback: bool) -> str:
-    if isinstance(summary, _messages.DeprecationSummary):
-        deprecation_message = _display._get_deprecation_message_with_plugin_info(
-            msg=summary.event.msg,
-            version=summary.version,
-            date=summary.date,
-            deprecator=summary.deprecator,
-        )
-
-        event = dataclasses.replace(summary.event, msg=deprecation_message)
-    else:
-        event = summary.event
-
-    return _event_formatting.format_event(event, include_traceback)
-
-
 def _report_config_warnings(deprecator: _messages.PluginInfo) -> None:
     """Called by config to report warnings/deprecations collected during a config parse."""
     while config._errors:

test/units/_internal/templating/test_jinja_bits.py

@@ -9,6 +9,7 @@ from contextlib import nullcontext
 import pytest
 import pytest_mock
 
+from ansible._internal import _display_utils
 from ansible._internal._templating._access import NotifiableAccessContextBase
 from ansible.errors import AnsibleUndefinedVariable, AnsibleTemplateError
 from ansible._internal._templating._errors import AnsibleTemplatePluginRuntimeError
@@ -22,8 +23,6 @@ from ansible._internal._templating import _jinja_plugins
 from ansible._internal._templating._engine import TemplateEngine, TemplateOptions
 from jinja2.loaders import DictLoader
 
-from ansible.utils.display import _DeferredWarningContext
-
 if t.TYPE_CHECKING:
     import unittest.mock
 
@@ -79,7 +78,7 @@ def test_templatemodule_ignore(template_context):
     templar = TemplateEngine()
     templar.environment.loader = DictLoader(dict(foo=TRUST.tag('{{ undefined_in_import }}')))
 
-    with _DeferredWarningContext(variables=templar.available_variables) as warnings:
+    with _display_utils.DeferredWarningContext(variables=templar.available_variables) as warnings:
         result = templar.template(template)
 
     assert not warnings.get_warnings()

test/units/_internal/templating/test_templar.py

@@ -31,6 +31,7 @@ from jinja2.loaders import DictLoader
 
 import unittest
 
+from ansible._internal import _display_utils
 from ansible._internal._templating._datatag import _JinjaConstTemplate
 from ansible.errors import (
     AnsibleError, AnsibleUndefinedVariable, AnsibleTemplateSyntaxError, AnsibleBrokenConditionalError, AnsibleTemplateError, AnsibleTemplateTransformLimitError,
@@ -52,7 +53,7 @@ from ansible._internal._templating._jinja_bits import AnsibleEnvironment, Ansibl
 from ansible._internal._templating._marker_behaviors import ReplacingMarkerBehavior
 from ansible._internal._templating._utils import TemplateContext
 from ansible.module_utils._internal import _event_utils
-from ansible.utils.display import Display, _DeferredWarningContext
+from ansible.utils.display import Display
 from units.mock.loader import DictDataLoader
 from units.test_utils.controller.display import emits_warnings
 
@@ -1033,7 +1034,7 @@ def test_deprecated_dedupe_and_source():
 
     templar = TemplateEngine(variables=variables)
 
-    with _DeferredWarningContext(variables=variables) as dwc:
+    with _display_utils.DeferredWarningContext(variables=variables) as dwc:
         # The indirect access summary occurs first.
         # The two following direct access summaries get deduped to a single one by the warning context (but unique template value keeps distinct from indirect).
         # The accesses with the shared tag instance values are internally deduped by the audit context.
@@ -1049,14 +1050,14 @@ def test_deprecated_dedupe_and_source():
 
 def test_jinja_const_template_leak(template_context: TemplateContext) -> None:
     """Verify that _JinjaConstTemplate is present during internal templating."""
-    with _DeferredWarningContext(variables={}):  # suppress warning from usage of embedded template
+    with _display_utils.DeferredWarningContext(variables={}):  # suppress warning from usage of embedded template
         with unittest.mock.patch.object(_TemplateConfig, 'allow_embedded_templates', True):
             assert _JinjaConstTemplate.is_tagged_on(TemplateEngine().template(TRUST.tag("{{ '{{ 1 }}' }}")))
 
 
 def test_jinja_const_template_finalized() -> None:
     """Verify that _JinjaConstTemplate is not present in finalized template results."""
-    with _DeferredWarningContext(variables={}):  # suppress warning from usage of embedded template
+    with _display_utils.DeferredWarningContext(variables={}):  # suppress warning from usage of embedded template
         with unittest.mock.patch.object(_TemplateConfig, 'allow_embedded_templates', True):
             assert not _JinjaConstTemplate.is_tagged_on(TemplateEngine().template(TRUST.tag("{{ '{{ 1 }}' }}")))
 

test/units/errors/test_utils.py

@@ -5,9 +5,9 @@ import pytest
 from ansible._internal._errors import _error_factory
 
 from ansible.errors import AnsibleError
+from ansible._internal import _display_utils
 from ansible._internal._datatag._tags import Origin
 from ansible._internal._errors._error_utils import format_exception_message
-from ansible.utils.display import _format_message
 from ansible.module_utils._internal import _messages
 from units.mock.error_helper import raise_exceptions
 
@@ -186,7 +186,7 @@ def test_error_messages(exceptions: list[BaseException], expected_message_chain:
     event = _error_factory.ControllerEventFactory.from_exception(error.value, False)
 
     message_chain = format_exception_message(error.value)
-    formatted_message = _format_message(_messages.ErrorSummary(event=event), False)
+    formatted_message = _display_utils.format_message(_messages.ErrorSummary(event=event), False)
 
     assert message_chain == expected_message_chain
     assert formatted_message.strip() == (expected_formatted_message or expected_message_chain)

test/units/parsing/yaml/test_objects.py

@@ -4,17 +4,17 @@ import typing as t
 
 import pytest
 
+from ansible._internal import _display_utils
 from ansible._internal._datatag._tags import Origin
 from ansible.module_utils._internal._datatag import AnsibleTagHelper
 from ansible.parsing.vault import EncryptedString
 from ansible.parsing.yaml.objects import _AnsibleMapping, _AnsibleUnicode, _AnsibleSequence
-from ansible.utils.display import _DeferredWarningContext
 from ansible.parsing.yaml import objects
 
 
 @pytest.fixture(autouse=True, scope='function')
 def suppress_warnings() -> t.Generator[None]:
-    with _DeferredWarningContext(variables={}):
+    with _display_utils.DeferredWarningContext(variables={}):
         yield
 
 

test/units/test_utils/controller/display.py

@@ -4,8 +4,8 @@ import contextlib
 import re
 import typing as t
 
+from ansible._internal import _display_utils
 from ansible.module_utils._internal import _messages
-from ansible.utils.display import _DeferredWarningContext
 
 
 @contextlib.contextmanager
@@ -16,7 +16,7 @@ def emits_warnings(
     allow_unmatched_message: bool = False,
 ) -> t.Iterator[None]:
     """Assert that the code within the context manager body emits a warning or deprecation warning whose formatted output matches the supplied regex."""
-    with _DeferredWarningContext(variables=dict(ansible_deprecation_warnings=True)) as ctx:
+    with _display_utils.DeferredWarningContext(variables=dict(ansible_deprecation_warnings=True)) as ctx:
         yield
 
     deprecations = ctx.get_deprecation_warnings()

test/units/utils/test_display.py

@@ -16,8 +16,9 @@ import pytest
 
 from ansible.module_utils.datatag import deprecator_from_collection_name
 from ansible.module_utils._internal import _deprecator, _errors, _messages
-from ansible.utils.display import _LIBC, _MAX_INT, Display, get_text_width, _format_message
+from ansible.utils.display import _LIBC, _MAX_INT, Display, get_text_width
 from ansible.utils.multiprocessing import context as multiprocessing_context
+from ansible._internal import _display_utils
 
 
 @pytest.fixture
@@ -164,7 +165,7 @@ def test_format_message_deprecation_with_multiple_details() -> None:
         ),
     )
 
-    result = _format_message(_messages.DeprecationSummary(event=event), False)
+    result = _display_utils.format_message(_messages.DeprecationSummary(event=event), False)
 
     assert result == '''Ignoring ExceptionX. This feature will be removed in the future: Something went wrong.
 
@@ -236,7 +237,7 @@ def test_get_deprecation_message_with_plugin_info(kwargs: dict[str, t.Any], expe
     for kwarg in ('version', 'date', 'deprecator'):
         kwargs.setdefault(kwarg, None)
 
-    msg = Display()._get_deprecation_message_with_plugin_info(**kwargs)
+    msg = _display_utils.get_deprecation_message_with_plugin_info(**kwargs)
 
     assert msg == expected