diff --git a/.changes/next-release/enhancement-Output-51826.json b/.changes/next-release/enhancement-Output-51826.json
new file mode 100644
index 000000000000..685f6a4c976c
--- /dev/null
+++ b/.changes/next-release/enhancement-Output-51826.json
@@ -0,0 +1,5 @@
+{
+ "type": "enhancement",
+ "category": "Output",
+ "description": "Add support for ``--output off`` to suppress all stdout output while preserving stderr for errors and warnings."
+}
diff --git a/.changes/next-release/feature-Output-59989.json b/.changes/next-release/feature-Output-59989.json
new file mode 100644
index 000000000000..115a7ad09e17
--- /dev/null
+++ b/.changes/next-release/feature-Output-59989.json
@@ -0,0 +1,5 @@
+{
+ "type": "feature",
+ "category": "Output",
+ "description": "Add structured error output with configurable formats. AWS service errors now display additional fields in the configured format (legacy, json, yaml, text, table, or enhanced). Configure via ``--cli-error-format``, ``cli_error_format`` config variable, or ``AWS_CLI_ERROR_FORMAT`` environment variable. The new enhanced format is the default. Set ``cli_error_format=legacy`` to preserve the original error format."
+}
diff --git a/awscli/clidriver.py b/awscli/clidriver.py
index ae8a3af68845..a049f03e0c03 100644
--- a/awscli/clidriver.py
+++ b/awscli/clidriver.py
@@ -30,6 +30,7 @@
ScopedConfigProvider,
)
from botocore.context import start_as_current_context
+from botocore.exceptions import ClientError
from botocore.history import get_global_history_recorder
from awscli import __version__
@@ -119,7 +120,7 @@ def create_clidriver(args=None):
session.full_config.get('plugins', {}),
event_hooks=session.get_component('event_emitter'),
)
- error_handlers_chain = construct_cli_error_handlers_chain()
+ error_handlers_chain = construct_cli_error_handlers_chain(session)
driver = CLIDriver(
session=session, error_handler=error_handlers_chain, debug=debug
)
@@ -246,7 +247,9 @@ def __init__(self, session=None, error_handler=None, debug=False):
self.session = session
self._error_handler = error_handler
if self._error_handler is None:
- self._error_handler = construct_cli_error_handlers_chain()
+ self._error_handler = construct_cli_error_handlers_chain(
+ self.session
+ )
if debug:
self._set_logging(debug)
self._update_config_chain()
@@ -275,6 +278,9 @@ def _update_config_chain(self):
config_store.set_config_provider(
'cli_help_output', self._construct_cli_help_output_chain()
)
+ config_store.set_config_provider(
+ 'cli_error_format', self._construct_cli_error_format_chain()
+ )
def _construct_cli_region_chain(self):
providers = [
@@ -368,6 +374,20 @@ def _construct_cli_auto_prompt_chain(self):
]
return ChainProvider(providers=providers)
+ def _construct_cli_error_format_chain(self):
+ providers = [
+ EnvironmentProvider(
+ name='AWS_CLI_ERROR_FORMAT',
+ env=os.environ,
+ ),
+ ScopedConfigProvider(
+ config_var_name='cli_error_format',
+ session=self.session,
+ ),
+ ConstantProvider(value='enhanced'),
+ ]
+ return ChainProvider(providers=providers)
+
@property
def subcommand_table(self):
return self._get_command_table()
@@ -516,12 +536,16 @@ def main(self, args=None):
command_table = self._get_command_table()
parser = self.create_parser(command_table)
self._add_aliases(command_table, parser)
+
try:
# Because _handle_top_level_args emits events, it's possible
# that exceptions can be raised, which should have the same
# general exception handling logic as calling into the
# command table. This is why it's in the try/except clause.
parsed_args, remaining = parser.parse_known_args(args)
+ self._error_handler = construct_cli_error_handlers_chain(
+ self.session, parsed_args
+ )
self._handle_top_level_args(parsed_args)
validate_preferred_output_encoding()
self._emit_session_event(parsed_args)
@@ -1028,9 +1052,7 @@ def _make_client_call(
paginator = client.get_paginator(py_operation_name)
response = paginator.paginate(**parameters)
else:
- response = getattr(client, xform_name(operation_name))(
- **parameters
- )
+ response = getattr(client, py_operation_name)(**parameters)
return response
def _display_response(self, command_name, response, parsed_globals):
diff --git a/awscli/data/cli.json b/awscli/data/cli.json
index 27d32410393f..312d46b9e192 100644
--- a/awscli/data/cli.json
+++ b/awscli/data/cli.json
@@ -26,7 +26,8 @@
"text",
"table",
"yaml",
- "yaml-stream"
+ "yaml-stream",
+ "off"
],
"help": "The formatting style for command output.
"
},
@@ -85,6 +86,17 @@
"no-cli-auto-prompt": {
"action": "store_true",
"help": "Disable automatically prompt for CLI input parameters.
"
+ },
+ "cli-error-format": {
+ "choices": [
+ "legacy",
+ "json",
+ "yaml",
+ "text",
+ "table",
+ "enhanced"
+ ],
+ "help": "The formatting style for error output. By default, errors are displayed in enhanced format.
"
}
}
}
diff --git a/awscli/errorhandler.py b/awscli/errorhandler.py
index 51d1c31f9514..2cdf759b938f 100644
--- a/awscli/errorhandler.py
+++ b/awscli/errorhandler.py
@@ -10,6 +10,7 @@
# distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF
# ANY KIND, either express or implied. See the License for the specific
# language governing permissions and limitations under the License.
+import argparse
import logging
import signal
@@ -36,10 +37,64 @@
ConfigurationError,
ParamValidationError,
)
+from awscli.formatter import get_formatter
from awscli.utils import PagerInitializationException
LOG = logging.getLogger(__name__)
+VALID_ERROR_FORMATS = ['legacy', 'json', 'yaml', 'text', 'table', 'enhanced']
+# Maximum number of items to display inline for collections
+MAX_INLINE_ITEMS = 5
+
+
+class EnhancedErrorFormatter:
+ def format_error(self, error_info, formatted_message, stream):
+ stream.write(f'{formatted_message}\n')
+
+ additional_fields = self._get_additional_fields(error_info)
+
+ if not additional_fields:
+ return
+
+ stream.write('\nAdditional error details:\n')
+ for key, value in additional_fields.items():
+ if self._is_simple_value(value):
+ stream.write(f'{key}: {value}\n')
+ elif self._is_small_collection(value):
+ stream.write(f'{key}: {self._format_inline(value)}\n')
+ else:
+ stream.write(
+ f'{key}: \n'
+ f'(Use --cli-error-format with json or yaml '
+ f'to see full details)\n'
+ )
+
+ def _is_simple_value(self, value):
+ return isinstance(value, (str, int, float, bool, type(None)))
+
+ def _is_small_collection(self, value):
+ if isinstance(value, list):
+ return len(value) < MAX_INLINE_ITEMS and all(
+ self._is_simple_value(item) for item in value
+ )
+ elif isinstance(value, dict):
+ return len(value) < MAX_INLINE_ITEMS and all(
+ self._is_simple_value(v) for v in value.values()
+ )
+ return False
+
+ def _format_inline(self, value):
+ if isinstance(value, list):
+ return f"[{', '.join(str(item) for item in value)}]"
+ elif isinstance(value, dict):
+ items = ', '.join(f'{k}: {v}' for k, v in value.items())
+ return f'{{{items}}}'
+ return str(value)
+
+ def _get_additional_fields(self, error_info):
+ standard_keys = {'Code', 'Message'}
+ return {k: v for k, v in error_info.items() if k not in standard_keys}
+
def construct_entry_point_handlers_chain():
handlers = [
@@ -51,7 +106,7 @@ def construct_entry_point_handlers_chain():
return ChainedExceptionHandler(exception_handlers=handlers)
-def construct_cli_error_handlers_chain():
+def construct_cli_error_handlers_chain(session=None, parsed_globals=None):
handlers = [
ParamValidationErrorsHandler(),
UnknownArgumentErrorHandler(),
@@ -60,7 +115,7 @@ def construct_cli_error_handlers_chain():
NoCredentialsErrorHandler(),
PagerErrorHandler(),
InterruptExceptionHandler(),
- ClientErrorHandler(),
+ ClientErrorHandler(session, parsed_globals),
GeneralExceptionHandler(),
]
return ChainedExceptionHandler(exception_handlers=handlers)
@@ -108,6 +163,111 @@ class ClientErrorHandler(FilteredExceptionHandler):
EXCEPTIONS_TO_HANDLE = ClientError
RC = CLIENT_ERROR_RC
+ def __init__(self, session=None, parsed_globals=None):
+ self._session = session
+ self._parsed_globals = parsed_globals
+ self._enhanced_formatter = None
+
+ def _do_handle_exception(self, exception, stdout, stderr):
+ displayed_structured = False
+ if self._session:
+ displayed_structured = self._try_display_structured_error(
+ exception, stderr
+ )
+
+ if not displayed_structured:
+ return super()._do_handle_exception(exception, stdout, stderr)
+
+ return self.RC
+
+ def _resolve_error_format(self, parsed_globals):
+ if parsed_globals:
+ error_format = getattr(parsed_globals, 'cli_error_format', None)
+ if error_format:
+ return error_format.lower()
+ try:
+ error_format = self._session.get_config_variable(
+ 'cli_error_format'
+ )
+ if error_format:
+ return error_format.lower()
+ except (KeyError, AttributeError) as e:
+ LOG.debug(
+ 'Failed to get cli_error_format from config: %s', e
+ )
+
+ return 'enhanced'
+
+ def _try_display_structured_error(self, exception, stderr):
+ try:
+ error_response = self._extract_error_response(exception)
+ if not error_response or 'Error' not in error_response:
+ return False
+
+ error_info = error_response['Error']
+ error_format = self._resolve_error_format(self._parsed_globals)
+
+ if error_format not in VALID_ERROR_FORMATS:
+ LOG.warning(
+ f"Invalid cli_error_format: '{error_format}'. "
+ f"Using 'enhanced' format."
+ )
+ error_format = 'enhanced'
+
+ if error_format == 'legacy':
+ return False
+
+ formatted_message = str(exception)
+
+ if error_format == 'enhanced':
+ if self._enhanced_formatter is None:
+ self._enhanced_formatter = EnhancedErrorFormatter()
+ self._enhanced_formatter.format_error(
+ error_info, formatted_message, stderr
+ )
+ return True
+
+ temp_parsed_globals = argparse.Namespace()
+ temp_parsed_globals.output = error_format
+ temp_parsed_globals.query = None
+ temp_parsed_globals.color = (
+ getattr(self._parsed_globals, 'color', 'auto')
+ if self._parsed_globals
+ else 'auto'
+ )
+
+ formatter = get_formatter(error_format, temp_parsed_globals)
+ formatter('error', error_info, stderr)
+ return True
+
+ except Exception as e:
+ LOG.debug(
+ 'Failed to display structured error: %s', e, exc_info=True
+ )
+ return False
+
+ @staticmethod
+ def _extract_error_response(exception):
+ if not isinstance(exception, ClientError):
+ return None
+
+ if hasattr(exception, 'response') and 'Error' in exception.response:
+ error_dict = dict(exception.response['Error'])
+
+ # AWS services return modeled error fields
+ # at the top level of the error response,
+ # not nested under an Error key. Botocore preserves this structure.
+ # Include these fields to provide complete error information.
+ # Exclude response metadata and avoid duplicates.
+ excluded_keys = {'Error', 'ResponseMetadata', 'Code', 'Message'}
+ for key, value in exception.response.items():
+ if key not in excluded_keys and key not in error_dict:
+ error_dict[key] = value
+
+ return {'Error': error_dict}
+
+ return None
+
class ConfigurationErrorHandler(FilteredExceptionHandler):
EXCEPTIONS_TO_HANDLE = ConfigurationError
diff --git a/awscli/examples/global_options.rst b/awscli/examples/global_options.rst
index cec1f5529736..20354555fe3c 100644
--- a/awscli/examples/global_options.rst
+++ b/awscli/examples/global_options.rst
@@ -29,6 +29,8 @@
* yaml-stream
+ * off
+
``--query`` (string)
@@ -96,3 +98,21 @@
Disable automatically prompt for CLI input parameters.
+``--cli-error-format`` (string)
+
+ The formatting style for error output. By default, errors are displayed in enhanced format.
+
+
+ * legacy
+
+ * json
+
+ * yaml
+
+ * text
+
+ * table
+
+ * enhanced
+
+
diff --git a/awscli/examples/global_synopsis.rst b/awscli/examples/global_synopsis.rst
index 1ca332c717ff..3e603348debb 100644
--- a/awscli/examples/global_synopsis.rst
+++ b/awscli/examples/global_synopsis.rst
@@ -16,3 +16,4 @@
[--no-cli-pager]
[--cli-auto-prompt]
[--no-cli-auto-prompt]
+[--cli-error-format ]
diff --git a/awscli/formatter.py b/awscli/formatter.py
index 60d80d8db0c6..6cbd46369081 100644
--- a/awscli/formatter.py
+++ b/awscli/formatter.py
@@ -227,9 +227,9 @@ def _format_response(self, command_name, response, stream):
try:
self.table.render(stream)
except OSError:
- # If they're piping stdout to another process which exits before
- # we're done writing all of our output, we'll get an error about a
- # closed pipe which we can safely ignore.
+ # If they're piping stdout to another process which exits
+ # before we're done writing all of our output, we'll get an
+ # error about a closed pipe which we can safely ignore.
pass
def _build_table(self, title, current, indent_level=0):
@@ -368,12 +368,26 @@ def _format_response(self, response, stream):
text.format_text(response, stream)
+class OffFormatter(Formatter):
+ """Formatter that suppresses all output.
+ Only stdout is suppressed; stderr (error messages) remains visible.
+ """
+
+ def __call__(self, command_name, response, stream=None):
+ if is_response_paginated(response):
+ for page in response:
+ self._get_transformed_response_for_output(page)
+ else:
+ self._get_transformed_response_for_output(response)
+
+
CLI_OUTPUT_FORMATS = {
'json': JSONFormatter,
'text': TextFormatter,
'table': TableFormatter,
'yaml': YAMLFormatter,
'yaml-stream': StreamedYAMLFormatter,
+ 'off': OffFormatter,
}
diff --git a/awscli/topics/config-vars.rst b/awscli/topics/config-vars.rst
index 8e291725c166..28828bc88fba 100644
--- a/awscli/topics/config-vars.rst
+++ b/awscli/topics/config-vars.rst
@@ -82,6 +82,9 @@ max_attempts N/A max_attempts AWS_MAX_ATTEMPTS
retry_mode N/A retry_mode AWS_RETRY_MODE Type of retries performed
-------------------- -------------- --------------------- --------------------- --------------------------------
cli_pager --no-cli-pager cli_pager AWS_PAGER Redirect/Disable output to pager
+-------------------- -------------- --------------------- --------------------- --------------------------------
+cli_error_format --cli-error- cli_error_format AWS_CLI_ERROR_FORMAT Format for error output
+ format
==================== ============== ===================== ===================== ================================
The third column, Config Entry, is the value you would specify in the AWS CLI
@@ -94,6 +97,29 @@ The valid values of the ``output`` configuration variable are:
* json
* table
* text
+* yaml
+* yaml-stream
+* off
+
+The ``off`` value suppresses all stdout output while preserving stderr for
+errors and warnings.
+
+``cli_error_format`` controls how AWS service errors are displayed. The valid
+values of the ``cli_error_format`` configuration variable are:
+
+* enhanced - AWS service errors display additional modeled fields in a
+ human-readable format with inline display for simple values and small
+ collections. This is the default behavior.
+* json - AWS service errors are formatted as JSON, showing all modeled fields
+ in the error response.
+* yaml - AWS service errors are formatted as YAML, showing all modeled fields
+ in the error response.
+* text - AWS service errors are formatted as text with key-value pairs, showing
+ all modeled fields in the error response.
+* table - AWS service errors are formatted as a table, showing all modeled
+ fields in the error response.
+* legacy - AWS service errors are written to stderr as unstructured text,
+ displaying only the error code and message without additional modeled fields.
``cli_timestamp_format`` controls the format of timestamps displayed by the AWS CLI.
The valid values of the ``cli_timestamp_format`` configuration variable are:
diff --git a/tests/functional/autocomplete/test_completer.py b/tests/functional/autocomplete/test_completer.py
index 0e9e5dbc2a4c..568d38566509 100644
--- a/tests/functional/autocomplete/test_completer.py
+++ b/tests/functional/autocomplete/test_completer.py
@@ -483,7 +483,7 @@ def test_return_suggestions_for_global_arg_with_choices(self):
suggestions = self.completer.complete(parsed)
names = [s.name for s in suggestions]
self.assertEqual(
- names, ['json', 'text', 'table', 'yaml', 'yaml-stream']
+ names, ['json', 'text', 'table', 'yaml', 'yaml-stream', 'off']
)
def test_not_return_suggestions_for_global_arg_wo_trailing_space(self):
diff --git a/tests/unit/test_formatter.py b/tests/unit/test_formatter.py
index db24f1946754..f4f35a5677c5 100644
--- a/tests/unit/test_formatter.py
+++ b/tests/unit/test_formatter.py
@@ -19,7 +19,12 @@
from botocore.paginate import PageIterator
from awscli.compat import StringIO, contextlib
-from awscli.formatter import JSONFormatter, StreamedYAMLFormatter, YAMLDumper
+from awscli.formatter import (
+ JSONFormatter,
+ OffFormatter,
+ StreamedYAMLFormatter,
+ YAMLDumper,
+)
from awscli.testutils import mock, unittest
@@ -180,3 +185,44 @@ def test_encoding_override(self, env_vars):
'}\n'
).encode()
)
+
+
+class TestOffFormatter:
+ def setup_method(self):
+ self.args = Namespace(query=None)
+ self.formatter = OffFormatter(self.args)
+ self.output = StringIO()
+
+ def test_suppresses_simple_response(self):
+ response = {'Key': 'Value'}
+ self.formatter('test-command', response, self.output)
+ assert self.output.getvalue() == ''
+
+ def test_suppresses_complex_response(self):
+ response = {
+ 'Items': [
+ {'Name': 'Item1', 'Value': 'data'},
+ {'Name': 'Item2', 'Value': 'more-data'}
+ ],
+ 'Count': 2
+ }
+ self.formatter('test-command', response, self.output)
+ assert self.output.getvalue() == ''
+
+ def test_suppresses_empty_response(self):
+ response = {}
+ self.formatter('test-command', response, self.output)
+ assert self.output.getvalue() == ''
+
+ def test_suppresses_paginated_response(self):
+ response = FakePageIterator([
+ {'Items': ['Item1']},
+ {'Items': ['Item2']}
+ ])
+ self.formatter('test-command', response, self.output)
+ assert self.output.getvalue() == ''
+
+ def test_works_without_stream(self):
+ response = {'Key': 'Value'}
+ # Should not raise an exception
+ self.formatter('test-command', response, None)
diff --git a/tests/unit/test_structured_error.py b/tests/unit/test_structured_error.py
new file mode 100644
index 000000000000..cdd811daa550
--- /dev/null
+++ b/tests/unit/test_structured_error.py
@@ -0,0 +1,577 @@
+# Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+#
+# Licensed under the Apache License, Version 2.0 (the "License"). You
+# may not use this file except in compliance with the License. A copy of
+# the License is located at
+#
+# http://aws.amazon.com/apache2.0/
+#
+# or in the "license" file accompanying this file. This file is
+# distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF
+# ANY KIND, either express or implied. See the License for the specific
+# language governing permissions and limitations under the License.
+import argparse
+import io
+from unittest import mock
+
+from botocore.exceptions import ClientError
+
+from awscli.constants import CLIENT_ERROR_RC
+from awscli.errorhandler import (
+ ClientErrorHandler,
+ EnhancedErrorFormatter,
+ construct_cli_error_handlers_chain,
+)
+from tests.unit.test_clidriver import FakeSession
+
+
+class TestClientErrorHandler:
+ def setup_method(self):
+ self.session = FakeSession()
+ self.handler = ClientErrorHandler(self.session)
+
+ def test_extract_error_response_from_client_error(self):
+ error_response = {
+ 'Error': {
+ 'Code': 'NoSuchBucket',
+ 'Message': 'The specified bucket does not exist',
+ 'BucketName': 'my-bucket',
+ },
+ 'ResponseMetadata': {'RequestId': '123'},
+ }
+ client_error = ClientError(error_response, 'GetObject')
+
+ result = ClientErrorHandler._extract_error_response(client_error)
+
+ assert result is not None
+ assert 'Error' in result
+ assert result['Error']['Code'] == 'NoSuchBucket'
+ assert result['Error']['BucketName'] == 'my-bucket'
+
+ def test_extract_error_response_with_top_level_fields(self):
+ error_response = {
+ 'Error': {
+ 'Code': 'TransactionCanceledException',
+ 'Message': 'Transaction cancelled',
+ },
+ 'CancellationReasons': [
+ {
+ 'Code': 'ConditionalCheckFailed',
+ 'Message': 'The conditional request failed',
+ 'Item': {'id': {'S': '123'}},
+ }
+ ],
+ 'ResponseMetadata': {'RequestId': '456'},
+ }
+ client_error = ClientError(error_response, 'TransactWriteItems')
+
+ result = ClientErrorHandler._extract_error_response(client_error)
+
+ assert result is not None
+ assert 'Error' in result
+ assert result['Error']['Code'] == 'TransactionCanceledException'
+ assert 'CancellationReasons' in result['Error']
+ assert len(result['Error']['CancellationReasons']) == 1
+ assert (
+ result['Error']['CancellationReasons'][0]['Code']
+ == 'ConditionalCheckFailed'
+ )
+
+ def test_extract_error_response_from_non_client_error(self):
+ result = ClientErrorHandler._extract_error_response(
+ ValueError('Some error')
+ )
+ assert result is None
+
+ def test_displays_structured_error_with_additional_members(self):
+ error_response = {
+ 'Error': {
+ 'Code': 'NoSuchBucket',
+ 'Message': 'Error',
+ 'BucketName': 'my-bucket',
+ },
+ 'ResponseMetadata': {'RequestId': '123'},
+ }
+ client_error = ClientError(error_response, 'GetObject')
+
+ stdout = io.StringIO()
+ stderr = io.StringIO()
+
+ rc = self.handler.handle_exception(client_error, stdout, stderr)
+
+ assert rc == CLIENT_ERROR_RC
+ stderr_output = stderr.getvalue()
+ assert 'NoSuchBucket' in stderr_output
+ assert 'my-bucket' in stderr_output
+ assert 'BucketName' in stderr_output
+
+ def test_displays_standard_error_without_additional_members(self):
+ error_response = {
+ 'Error': {
+ 'Code': 'AccessDenied',
+ 'Message': 'Access Denied',
+ },
+ 'ResponseMetadata': {'RequestId': '123'},
+ }
+ client_error = ClientError(error_response, 'GetObject')
+
+ stdout = io.StringIO()
+ stderr = io.StringIO()
+
+ rc = self.handler.handle_exception(client_error, stdout, stderr)
+
+ assert rc == CLIENT_ERROR_RC
+ stderr_output = stderr.getvalue()
+ assert 'AccessDenied' in stderr_output
+ assert 'Additional error details' not in stderr_output
+
+ def test_respects_legacy_format_config(self):
+ error_response = {
+ 'Error': {
+ 'Code': 'NoSuchBucket',
+ 'Message': 'Error',
+ 'BucketName': 'test',
+ },
+ 'ResponseMetadata': {'RequestId': '123'},
+ }
+ client_error = ClientError(error_response, 'GetObject')
+
+ self.session.session_vars['cli_error_format'] = 'legacy'
+
+ stdout = io.StringIO()
+ stderr = io.StringIO()
+
+ rc = self.handler.handle_exception(client_error, stdout, stderr)
+
+ assert rc == CLIENT_ERROR_RC
+ stderr_output = stderr.getvalue()
+ assert 'NoSuchBucket' in stderr_output
+ assert 'BucketName' not in stderr_output
+
+ def test_error_format_case_insensitive(self):
+ """Test that error format config is case-insensitive."""
+ error_response = {
+ 'Error': {
+ 'Code': 'NoSuchBucket',
+ 'Message': 'Error',
+ 'BucketName': 'test',
+ },
+ 'ResponseMetadata': {'RequestId': '123'},
+ }
+ client_error = ClientError(error_response, 'GetObject')
+
+ self.session.config_store.set_config_provider(
+ 'cli_error_format', mock.Mock(provide=lambda: 'Enhanced')
+ )
+
+ stdout = io.StringIO()
+ stderr = io.StringIO()
+
+ rc = self.handler.handle_exception(client_error, stdout, stderr)
+
+ assert rc == CLIENT_ERROR_RC
+ stderr_output = stderr.getvalue()
+ assert 'NoSuchBucket' in stderr_output
+ assert 'test' in stderr_output
+
+
+class TestEnhancedErrorFormatter:
+ def setup_method(self):
+ self.formatter = EnhancedErrorFormatter()
+
+ def test_is_simple_value(self):
+ assert self.formatter._is_simple_value('string')
+ assert self.formatter._is_simple_value(42)
+ assert self.formatter._is_simple_value(3.14)
+ assert self.formatter._is_simple_value(True)
+ assert self.formatter._is_simple_value(None)
+ assert not self.formatter._is_simple_value([1, 2, 3])
+ assert not self.formatter._is_simple_value({'key': 'value'})
+
+ def test_is_small_collection_list(self):
+ assert self.formatter._is_small_collection(['a', 'b'])
+ assert self.formatter._is_small_collection([1, 2, 3, 4])
+
+ assert not self.formatter._is_small_collection([1, 2, 3, 4, 5])
+
+ assert not self.formatter._is_small_collection([1, [2, 3]])
+ assert not self.formatter._is_small_collection([{'key': 'value'}])
+
+ def test_is_small_collection_dict(self):
+ assert self.formatter._is_small_collection({'key': 'value'})
+ assert self.formatter._is_small_collection(
+ {'a': 1, 'b': 2, 'c': 3, 'd': 4}
+ )
+
+ assert not self.formatter._is_small_collection(
+ {'a': 1, 'b': 2, 'c': 3, 'd': 4, 'e': 5}
+ )
+
+ assert not self.formatter._is_small_collection({'key': [1, 2]})
+ assert not self.formatter._is_small_collection({'key': {'nested': 1}})
+
+ def test_format_inline_list(self):
+ result = self.formatter._format_inline([1, 2, 3])
+ assert result == '[1, 2, 3]'
+
+ result = self.formatter._format_inline(['a', 'b', 'c'])
+ assert result == '[a, b, c]'
+
+ def test_format_inline_dict(self):
+ result = self.formatter._format_inline({'a': 1, 'b': 2})
+ assert 'a: 1' in result
+ assert 'b: 2' in result
+ assert result.startswith('{')
+ assert result.endswith('}')
+
+ def test_get_additional_fields(self):
+ error_info = {
+ 'Code': 'NoSuchBucket',
+ 'Message': 'The bucket does not exist',
+ 'BucketName': 'my-bucket',
+ 'Region': 'us-east-1',
+ }
+
+ additional = self.formatter._get_additional_fields(error_info)
+ assert 'Code' not in additional
+ assert 'Message' not in additional
+ assert additional['BucketName'] == 'my-bucket'
+ assert additional['Region'] == 'us-east-1'
+
+ def test_format_error_with_no_additional_fields(self):
+ error_info = {
+ 'Code': 'AccessDenied',
+ 'Message': 'Access Denied',
+ }
+ formatted_message = 'An error occurred (AccessDenied): Access Denied'
+
+ stream = io.StringIO()
+ self.formatter.format_error(error_info, formatted_message, stream)
+
+ output = stream.getvalue()
+ assert formatted_message in output
+ assert 'Additional error details' not in output
+
+ def test_format_error_with_simple_fields(self):
+ error_info = {
+ 'Code': 'NoSuchBucket',
+ 'Message': 'The bucket does not exist',
+ 'BucketName': 'my-bucket',
+ 'Region': 'us-east-1',
+ }
+ formatted_message = (
+ 'An error occurred (NoSuchBucket): The bucket does not exist'
+ )
+
+ stream = io.StringIO()
+ self.formatter.format_error(error_info, formatted_message, stream)
+
+ output = stream.getvalue()
+ assert formatted_message in output
+ assert 'Additional error details' in output
+ assert 'BucketName: my-bucket' in output
+ assert 'Region: us-east-1' in output
+
+ def test_format_error_with_small_list(self):
+ error_info = {
+ 'Code': 'ValidationError',
+ 'Message': 'Validation failed',
+ 'AllowedValues': ['value1', 'value2', 'value3'],
+ }
+ formatted_message = (
+ 'An error occurred (ValidationError): Validation failed'
+ )
+
+ stream = io.StringIO()
+ self.formatter.format_error(error_info, formatted_message, stream)
+
+ output = stream.getvalue()
+ assert formatted_message in output
+ assert 'Additional error details' in output
+ assert 'AllowedValues: [value1, value2, value3]' in output
+
+ def test_format_error_with_small_dict(self):
+ error_info = {
+ 'Code': 'ValidationError',
+ 'Message': 'Validation failed',
+ 'Metadata': {'key1': 'value1', 'key2': 'value2'},
+ }
+ formatted_message = (
+ 'An error occurred (ValidationError): Validation failed'
+ )
+
+ stream = io.StringIO()
+ self.formatter.format_error(error_info, formatted_message, stream)
+
+ output = stream.getvalue()
+ assert formatted_message in output
+ assert 'Additional error details' in output
+ assert 'Metadata:' in output
+ assert 'key1: value1' in output
+ assert 'key2: value2' in output
+
+ def test_format_error_with_complex_object(self):
+ error_info = {
+ 'Code': 'ValidationError',
+ 'Message': 'Validation failed',
+ 'Details': [1, 2, 3, 4, 5, 6],
+ }
+ formatted_message = (
+ 'An error occurred (ValidationError): Validation failed'
+ )
+
+ stream = io.StringIO()
+ self.formatter.format_error(error_info, formatted_message, stream)
+
+ output = stream.getvalue()
+ assert formatted_message in output
+ assert 'Additional error details' in output
+ assert 'Details: ' in output
+ assert '--cli-error-format with json or yaml' in output
+
+ def test_format_error_with_nested_dict(self):
+ """Test formatting with nested dictionary structures."""
+ error_info = {
+ 'Code': 'ValidationError',
+ 'Message': 'Validation failed',
+ 'FieldErrors': {
+ 'email': {'pattern': 'invalid', 'required': True},
+ 'age': {'min': 0, 'max': 120},
+ },
+ }
+ formatted_message = (
+ 'An error occurred (ValidationError): Validation failed'
+ )
+
+ stream = io.StringIO()
+ self.formatter.format_error(error_info, formatted_message, stream)
+
+ output = stream.getvalue()
+ assert formatted_message in output
+ assert 'FieldErrors: ' in output
+
+ def test_format_error_with_list_of_dicts(self):
+ """Test formatting with list containing dictionaries."""
+ error_info = {
+ 'Code': 'TransactionCanceledException',
+ 'Message': 'Transaction cancelled',
+ 'CancellationReasons': [
+ {'Code': 'ConditionalCheckFailed', 'Message': 'Check failed'},
+ {'Code': 'ItemCollectionSizeLimitExceeded', 'Message': 'Too large'},
+ ],
+ }
+ formatted_message = (
+ 'An error occurred (TransactionCanceledException): '
+ 'Transaction cancelled'
+ )
+
+ stream = io.StringIO()
+ self.formatter.format_error(error_info, formatted_message, stream)
+
+ output = stream.getvalue()
+ assert formatted_message in output
+ assert 'CancellationReasons: ' in output
+
+ def test_format_error_with_mixed_types(self):
+ """Test formatting with various data types."""
+ error_info = {
+ 'Code': 'ComplexError',
+ 'Message': 'Complex error occurred',
+ 'StringField': 'test-value',
+ 'IntField': 42,
+ 'FloatField': 3.14,
+ 'BoolField': True,
+ 'NoneField': None,
+ }
+ formatted_message = (
+ 'An error occurred (ComplexError): Complex error occurred'
+ )
+
+ stream = io.StringIO()
+ self.formatter.format_error(error_info, formatted_message, stream)
+
+ output = stream.getvalue()
+ assert formatted_message in output
+ assert 'Additional error details' in output
+ assert 'StringField: test-value' in output
+ assert 'IntField: 42' in output
+ assert 'FloatField: 3.14' in output
+ assert 'BoolField: True' in output
+ assert 'NoneField: None' in output
+
+ def test_format_error_with_unicode_and_special_chars(self):
+ """Test formatting with unicode and special characters."""
+ error_info = {
+ 'Code': 'InvalidInput',
+ 'Message': 'Invalid input provided',
+ 'UserName': 'José García',
+ 'Description': 'Error with "quotes" and \'apostrophes\'',
+ 'Path': '/path/to/file.txt',
+ }
+ formatted_message = (
+ 'An error occurred (InvalidInput): Invalid input provided'
+ )
+
+ stream = io.StringIO()
+ self.formatter.format_error(error_info, formatted_message, stream)
+
+ output = stream.getvalue()
+ assert formatted_message in output
+ assert 'José García' in output
+ assert 'quotes' in output
+ assert 'apostrophes' in output
+
+ def test_format_error_with_large_list(self):
+ """Test that large lists are marked as complex."""
+ error_info = {
+ 'Code': 'LargeList',
+ 'Message': 'Large list error',
+ 'Items': list(range(10)),
+ }
+ formatted_message = (
+ 'An error occurred (LargeList): Large list error'
+ )
+
+ stream = io.StringIO()
+ self.formatter.format_error(error_info, formatted_message, stream)
+
+ output = stream.getvalue()
+ assert formatted_message in output
+ assert 'Items: ' in output
+
+
+class TestRealWorldErrorScenarios:
+ """Test real-world AWS error response structures."""
+
+ def setup_method(self):
+ self.session = FakeSession()
+ self.handler = ClientErrorHandler(self.session)
+
+ def test_dynamodb_transaction_cancelled_error(self):
+ """Test DynamoDB TransactionCanceledException with CancellationReasons."""
+ error_response = {
+ 'Error': {
+ 'Code': 'TransactionCanceledException',
+ 'Message': (
+ 'Transaction cancelled, please refer to '
+ 'CancellationReasons for specific reasons'
+ ),
+ },
+ 'CancellationReasons': [
+ {
+ 'Code': 'ConditionalCheckFailed',
+ 'Message': 'The conditional request failed',
+ 'Item': {
+ 'id': {'S': 'item-123'},
+ 'status': {'S': 'active'},
+ },
+ },
+ {
+ 'Code': 'None',
+ 'Message': None,
+ },
+ ],
+ 'ResponseMetadata': {
+ 'RequestId': 'abc-123',
+ 'HTTPStatusCode': 400,
+ },
+ }
+ client_error = ClientError(error_response, 'TransactWriteItems')
+
+ stdout = io.StringIO()
+ stderr = io.StringIO()
+
+ rc = self.handler.handle_exception(client_error, stdout, stderr)
+
+ assert rc == CLIENT_ERROR_RC
+ stderr_output = stderr.getvalue()
+ assert 'TransactionCanceledException' in stderr_output
+ assert 'CancellationReasons' in stderr_output
+
+ def test_throttling_error_with_retry_info(self):
+ """Test throttling error with retry information."""
+ error_response = {
+ 'Error': {
+ 'Code': 'ThrottlingException',
+ 'Message': 'Rate exceeded',
+ },
+ 'RetryAfterSeconds': 30,
+ 'RequestsPerSecond': 100,
+ 'CurrentRate': 150,
+ 'ResponseMetadata': {'RequestId': 'throttle-123'},
+ }
+ client_error = ClientError(error_response, 'DescribeInstances')
+
+ stdout = io.StringIO()
+ stderr = io.StringIO()
+
+ rc = self.handler.handle_exception(client_error, stdout, stderr)
+
+ assert rc == CLIENT_ERROR_RC
+ stderr_output = stderr.getvalue()
+ assert 'ThrottlingException' in stderr_output
+ assert '30' in stderr_output
+ assert '100' in stderr_output
+
+
+class TestParsedGlobalsPassthrough:
+ def test_error_handler_receives_parsed_globals_from_clidriver(self):
+ session = FakeSession()
+
+ parsed_globals = argparse.Namespace()
+ parsed_globals.cli_error_format = 'json'
+ parsed_globals.command = 's3'
+ parsed_globals.color = 'auto'
+
+ error_handler = construct_cli_error_handlers_chain(
+ session, parsed_globals
+ )
+
+ error_response = {
+ 'Error': {
+ 'Code': 'NoSuchBucket',
+ 'Message': 'The specified bucket does not exist',
+ 'BucketName': 'test-bucket',
+ },
+ 'ResponseMetadata': {'RequestId': '123'},
+ }
+ client_error = ClientError(error_response, 'GetObject')
+
+ stdout = io.StringIO()
+ stderr = io.StringIO()
+
+ rc = error_handler.handle_exception(client_error, stdout, stderr)
+
+ assert rc == CLIENT_ERROR_RC
+
+ stderr_output = stderr.getvalue()
+ assert '"Code"' in stderr_output or '"code"' in stderr_output.lower()
+ assert 'NoSuchBucket' in stderr_output
+ assert 'test-bucket' in stderr_output
+
+ def test_error_handler_without_parsed_globals_uses_default(self):
+ session = FakeSession()
+
+ error_handler = construct_cli_error_handlers_chain(session, None)
+
+ error_response = {
+ 'Error': {
+ 'Code': 'NoSuchBucket',
+ 'Message': 'The specified bucket does not exist',
+ 'BucketName': 'test-bucket',
+ },
+ 'ResponseMetadata': {'RequestId': '123'},
+ }
+ client_error = ClientError(error_response, 'GetObject')
+
+ stdout = io.StringIO()
+ stderr = io.StringIO()
+
+ rc = error_handler.handle_exception(client_error, stdout, stderr)
+
+ assert rc == CLIENT_ERROR_RC
+
+ stderr_output = stderr.getvalue()
+ assert 'NoSuchBucket' in stderr_output
+ assert 'test-bucket' in stderr_output
+ assert 'BucketName' in stderr_output