File: _event_formatting.py

package info (click to toggle)
ansible-core 2.19.0~beta6-1
  • links: PTS, VCS
  • area: main
  • in suites: trixie
  • size: 32,628 kB
  • sloc: python: 180,313; cs: 4,929; sh: 4,601; xml: 34; makefile: 21
file content (127 lines) | stat: -rw-r--r-- 4,222 bytes parent folder | download | duplicates (3)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
from __future__ import annotations as _annotations

import collections.abc as _c
import textwrap as _textwrap

from ansible.module_utils._internal import _event_utils, _messages


def format_event(event: _messages.Event, include_traceback: bool) -> str:
    """Format an event into a verbose message and traceback."""
    msg = format_event_verbose_message(event)

    if include_traceback:
        msg += '\n' + format_event_traceback(event)

    msg = msg.strip()

    if '\n' in msg:
        msg += '\n\n'
    else:
        msg += '\n'

    return msg


def format_event_traceback(event: _messages.Event) -> str:
    """Format an event into a traceback."""
    segments: list[str] = []

    while event:
        segment = event.formatted_traceback or '(traceback missing)\n'

        if event.events:
            child_tracebacks = [format_event_traceback(child) for child in event.events]
            segment += _format_event_children("Sub-Traceback", child_tracebacks)

        segments.append(segment)

        if event.chain:
            segments.append(f'\n{event.chain.traceback_reason}\n\n')

            event = event.chain.event
        else:
            event = None

    return ''.join(reversed(segments))


def format_event_verbose_message(event: _messages.Event) -> str:
    """
    Format an event into a verbose message.
    Help text, contextual information and sub-events will be included.
    """
    segments: list[str] = []
    original_event = event

    while event:
        messages = [event.msg]
        chain: _messages.EventChain = event.chain

        while chain and chain.follow:
            if chain.event.events:
                break  # do not collapse a chained event with sub-events, since they would be lost

            if chain.event.formatted_source_context or chain.event.help_text:
                if chain.event.formatted_source_context != event.formatted_source_context or chain.event.help_text != event.help_text:
                    break  # do not collapse a chained event with different details, since they would be lost

            if chain.event.chain and chain.msg_reason != chain.event.chain.msg_reason:
                break  # do not collapse a chained event which has a chain with a different msg_reason

            messages.append(chain.event.msg)

            chain = chain.event.chain

        msg = _event_utils.deduplicate_message_parts(messages)
        segment = '\n'.join(_get_message_lines(msg, event.help_text, event.formatted_source_context)) + '\n'

        if event.events:
            child_msgs = [format_event_verbose_message(child) for child in event.events]
            segment += _format_event_children("Sub-Event", child_msgs)

        segments.append(segment)

        if chain and chain.follow:
            segments.append(f'\n{chain.msg_reason}\n\n')

            event = chain.event
        else:
            event = None

    if len(segments) > 1:
        segments.insert(0, _event_utils.format_event_brief_message(original_event) + '\n\n')

    return ''.join(segments)


def _format_event_children(label: str, children: _c.Iterable[str]) -> str:
    """Format the given list of child messages into a single string."""
    items = list(children)
    count = len(items)
    lines = ['\n']

    for idx, item in enumerate(items):
        lines.append(f'+--[ {label} {idx + 1} of {count} ]---\n')
        lines.append(_textwrap.indent(f"\n{item}\n", "| ", lambda value: True))

    lines.append(f'+--[ End {label} ]---\n')

    return ''.join(lines)


def _get_message_lines(message: str, help_text: str | None, formatted_source_context: str | None) -> list[str]:
    """Return a list of message lines constructed from the given message, help text and formatted source context."""
    if help_text and not formatted_source_context and '\n' not in message and '\n' not in help_text:
        return [f'{message} {help_text}']  # prefer a single-line message with help text when there is no source context

    message_lines = [message]

    if formatted_source_context:
        message_lines.append(formatted_source_context)

    if help_text:
        message_lines.append('')
        message_lines.append(help_text)

    return message_lines