Isolate query compatible changes into function

botocore/parsers.py

@@ -701,10 +701,6 @@ def _do_error_parse(self, response, shape):
         # if the message did not contain an error code
         # include the response status code
         response_code = response.get('status_code')
-        # Error response may contain an x-amzn-query-error header for json
-        # we need to fetch the error code from this header in that case
-        query_error = headers.get('x-amzn-query-error', '')
-        query_error_components = query_error.split(';')
 
         code = body.get('__type', response_code and str(response_code))
         if code is not None:
@@ -713,14 +709,29 @@ def _do_error_parse(self, response, shape):
             # * "ResourceNotFoundException"
             if '#' in code:
                 code = code.rsplit('#', 1)[1]
+            if 'x-amzn-query-error' in headers:
+                code = self._do_query_compatible_error_parse(
+                    code, headers, error
+                )
             error['Error']['Code'] = code
-            if len(query_error_components) == 2 and query_error_components[0]:
-                error['Error']['QueryErrorCode'] = code
-                error['Error']['Code'] = query_error_components[0]
-                error['Error']['Type'] = query_error_components[1]
         self._inject_response_metadata(error, response['headers'])
         return error
 
+    def _do_query_compatible_error_parse(self, code, headers, error):
+        """
+        Error response may contain an x-amzn-query-error header to translate
+        errors codes from former `query` services into `json`. We use this to
+        do our lookup in the errorfactory for modeled errors.
+        """
+        query_error = headers['x-amzn-query-error']
+        query_error_components = query_error.split(';')
+
+        if len(query_error_components) == 2 and query_error_components[0]:
+            error['Error']['QueryErrorCode'] = code
+            error['Error']['Type'] = query_error_components[1]
+            return query_error_components[0]
+        return code
+
     def _inject_response_metadata(self, parsed, headers):
         if 'x-amzn-requestid' in headers:
             parsed.setdefault('ResponseMetadata', {})['RequestId'] = headers[

tests/functional/test_sqs.py

@@ -0,0 +1,48 @@
+# Copyright 2023 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 pytest
+
+from tests import BaseSessionTest, ClientHTTPStubber
+
+
+class BaseSQSOperationTest(BaseSessionTest):
+    def setUp(self):
+        super().setUp()
+        self.region = "us-west-2"
+        self.client = self.session.create_client("sqs", self.region)
+        self.http_stubber = ClientHTTPStubber(self.client)
+
+
+class SQSQueryCompatibleTest(BaseSQSOperationTest):
+    def test_query_compatible_error_parsing(self):
+        """When migrating SQS from the ``query`` protocol to ``json``,
+        we unintentionally moved from modeled errors to a general ``ClientError``.
+        This ensures we're not silently regressing that behavior.
+        """
+
+        error_body = (
+            b'{"__type":"com.amazonaws.sqs#QueueDoesNotExist",'
+            b'"message":"The specified queue does not exist."}'
+        )
+        error_headers = {
+            "x-amzn-query-error": "AWS.SimpleQueueService.NonExistentQueue;Sender",
+        }
+        with self.http_stubber as stub:
+            stub.add_response(
+                status=400, body=error_body, headers=error_headers
+            )
+            with pytest.raises(self.client.exceptions.QueueDoesNotExist):
+                self.client.delete_queue(
+                    QueueUrl="not-a-real-queue-botocore",
+                )

tests/unit/test_parsers.py

@@ -1158,7 +1158,7 @@ def test_response_with_invalid_query_error_for_json_protocol(self):
         self.assertIn('Error', parsed)
         self.assertEqual(parsed['Error']['Message'], 'this is a message')
         self.assertEqual(parsed['Error']['Code'], 'ValidationException')
-        assert 'QueryErrorCode' not in parsed['Error']
+        self.assertNotIn('QueryErrorCode', parsed['Error'])
         self.assertNotIn('Type', parsed['Error'])
 
     def test_response_with_incomplete_query_error_for_json_protocol(self):