Merge "Add logging details of the http response in the unit tests" into stable/2023.1

Author: Zuul

neutron/tests/unit/db/test_db_base_plugin_v2.py

@@ -246,6 +246,14 @@ def _req(self, method, resource, data=None, fmt=None, id=None, params=None,
                                           query_string=params, context=context,
                                           headers=headers)
 
+    def _check_http_response(self, res):
+        # Things can go wrong - raise HTTP exc with res code only
+        # so it can be caught by unit tests
+        if res.status_int >= webob.exc.HTTPClientError.code:
+            res.charset = 'utf8'
+            raise webob.exc.HTTPClientError(explanation=str(res),
+                                            code=res.status_int)
+
     def new_create_request(self, resource, data, fmt=None, id=None,
                            subresource=None, context=None):
         return self._req('POST', resource, data, fmt, id=id,
@@ -523,10 +531,7 @@ def _make_network(self, fmt, name, admin_state_up, **kwargs):
         res = self._create_network(fmt, name, admin_state_up, **kwargs)
         # TODO(salvatore-orlando): do exception handling in this test module
         # in a uniform way (we do it differently for ports, subnets, and nets
-        # Things can go wrong - raise HTTP exc with res code only
-        # so it can be caught by unit tests
-        if res.status_int >= webob.exc.HTTPClientError.code:
-            raise webob.exc.HTTPClientError(code=res.status_int)
+        self._check_http_response(res)
         return self.deserialize(fmt, res)
 
     def _make_subnet(self, fmt, network, gateway, cidr, subnetpool_id=None,
@@ -551,10 +556,7 @@ def _make_subnet(self, fmt, network, gateway, cidr, subnetpool_id=None,
                                   ipv6_ra_mode=ipv6_ra_mode,
                                   ipv6_address_mode=ipv6_address_mode,
                                   set_context=set_context)
-        # Things can go wrong - raise HTTP exc with res code only
-        # so it can be caught by unit tests
-        if res.status_int >= webob.exc.HTTPClientError.code:
-            raise webob.exc.HTTPClientError(code=res.status_int)
+        self._check_http_response(res)
         return self.deserialize(fmt, res)
 
     def _make_v6_subnet(self, network, ra_addr_mode, ipv6_pd=False):
@@ -580,16 +582,14 @@ def _make_subnetpool(self, fmt, prefixes, admin=False, **kwargs):
                                       **kwargs)
         # Things can go wrong - raise HTTP exc with res code only
         # so it can be caught by unit tests
-        if res.status_int >= webob.exc.HTTPClientError.code:
-            raise webob.exc.HTTPClientError(code=res.status_int)
+        self._check_http_response(res)
         return self.deserialize(fmt, res)
 
     def _make_port(self, fmt, net_id, expected_res_status=None, **kwargs):
         res = self._create_port(fmt, net_id, expected_res_status, **kwargs)
         # Things can go wrong - raise HTTP exc with res code only
         # so it can be caught by unit tests
-        if res.status_int >= webob.exc.HTTPClientError.code:
-            raise webob.exc.HTTPClientError(code=res.status_int)
+        self._check_http_response(res)
         return self.deserialize(fmt, res)
 
     def _create_qos_rule(self, fmt, qos_policy_id, rule_type, max_kbps=None,

neutron/tests/unit/db/test_l3_db.py

@@ -32,7 +32,6 @@
 from neutron_lib.plugins import utils as plugin_utils
 from oslo_utils import uuidutils
 import testtools
-import webob.exc
 
 from neutron.db import extraroute_db
 from neutron.db import l3_db
@@ -951,8 +950,7 @@ def _create_external_network(self, name=None, **kwargs):
             res = self._create_network(
                 self.fmt, name, True,
                 arg_list=(extnet_apidef.EXTERNAL,), **kwargs)
-            if res.status_int >= webob.exc.HTTPClientError.code:
-                raise webob.exc.HTTPClientError(code=res.status_int)
+            self._check_http_response(res)
             return self.deserialize(self.fmt, res)
 
     def test_update_router_gw_notify(self):

neutron/tests/unit/extensions/test_address_group.py

@@ -52,8 +52,7 @@ def _create_address_group(self, **kwargs):
                                                          self._tenant_id))
         req.environ['neutron.context'] = neutron_context
         res = req.get_response(self.ext_api)
-        if res.status_int >= webob.exc.HTTPClientError.code:
-            raise webob.exc.HTTPClientError(code=res.status_int)
+        self._check_http_response(res)
         return res
 
     def _test_create_address_group(self, expected=None, **kwargs):

neutron/tests/unit/extensions/test_address_scope.py

@@ -71,8 +71,7 @@ def _create_address_scope(self, fmt, ip_version=constants.IP_VERSION_4,
     def _make_address_scope(self, fmt, ip_version, admin=False, **kwargs):
         res = self._create_address_scope(fmt, ip_version,
                                          admin=admin, **kwargs)
-        if res.status_int >= webob.exc.HTTPClientError.code:
-            raise webob.exc.HTTPClientError(code=res.status_int)
+        self._check_http_response(res)
         return self.deserialize(fmt, res)
 
     @contextlib.contextmanager

neutron/tests/unit/extensions/test_local_ip.py

@@ -51,8 +51,7 @@ def _create_local_ip(self, **kwargs):
             '', kwargs.get('project_id', self._tenant_id), is_admin=True)
         req.environ['neutron.context'] = neutron_context
         res = req.get_response(self.ext_api)
-        if res.status_int >= webob.exc.HTTPClientError.code:
-            raise webob.exc.HTTPClientError(code=res.status_int)
+        self._check_http_response(res)
         return self.deserialize(self.fmt, res)
 
     def _update_local_ip(self, lip_id, data):
@@ -61,8 +60,7 @@ def _update_local_ip(self, lip_id, data):
         update_req.environ['neutron.context'] = context.Context(
             '', self._tenant_id)
         res = update_req.get_response(self.ext_api)
-        if res.status_int >= webob.exc.HTTPClientError.code:
-            raise webob.exc.HTTPClientError(code=res.status_int)
+        self._check_http_response(res)
         return self.deserialize(self.fmt, res)
 
     def _create_local_ip_association(self, local_ip_id, fixed_port_id,
@@ -77,8 +75,7 @@ def _create_local_ip_association(self, local_ip_id, fixed_port_id,
         neutron_context = context.Context('', self._tenant_id)
         req.environ['neutron.context'] = neutron_context
         res = req.get_response(self.ext_api)
-        if res.status_int >= webob.exc.HTTPClientError.code:
-            raise webob.exc.HTTPClientError(code=res.status_int)
+        self._check_http_response(res)
         return self.deserialize(self.fmt, res)
 
     @contextlib.contextmanager

neutron/tests/unit/extensions/test_network_segment_range.py

@@ -67,8 +67,7 @@ def _create_network_segment_range(self, fmt, expected_res_status=None,
 
     def network_segment_range(self, **kwargs):
         res = self._create_network_segment_range(self.fmt, **kwargs)
-        if res.status_int >= webob.exc.HTTPClientError.code:
-            raise webob.exc.HTTPClientError(code=res.status_int)
+        self._check_http_response(res)
         return self.deserialize(self.fmt, res)
 
     def _test_create_network_segment_range(self, expected=None, **kwargs):
@@ -162,7 +161,7 @@ def test_create_network_segment_range_unsupported_network_type(self):
                                 self._test_create_network_segment_range,
                                 network_type='foo-network-type')
         self.assertEqual(webob.exc.HTTPClientError.code, exc.code)
-        self.assertIn('The server could not comply with the request',
+        self.assertIn('foo-network-type is not in valid_values',
                       exc.explanation)
 
     def test_create_network_segment_range_no_physical_network(self):
@@ -212,16 +211,14 @@ def test_create_network_segment_range_failed_with_vlan_minimum_id(self):
                                 self._test_create_network_segment_range,
                                 minimum=0)
         self.assertEqual(webob.exc.HTTPClientError.code, exc.code)
-        self.assertIn('The server could not comply with the request',
-                      exc.explanation)
+        self.assertIn('Invalid input for minimum', exc.explanation)
 
     def test_create_network_segment_range_failed_with_vlan_maximum_id(self):
         exc = self.assertRaises(webob.exc.HTTPClientError,
                                 self._test_create_network_segment_range,
                                 minimum=4095)
         self.assertEqual(webob.exc.HTTPServerError.code, exc.code)
-        self.assertIn('The server could not comply with the request',
-                      exc.explanation)
+        self.assertIn('Invalid network VLAN range', exc.explanation)
 
     def test_create_network_segment_range_failed_with_tunnel_minimum_id(self):
         tunnel_type = [constants.TYPE_VXLAN,
@@ -234,8 +231,7 @@ def test_create_network_segment_range_failed_with_tunnel_minimum_id(self):
                                     physical_network=None,
                                     minimum=0)
             self.assertEqual(webob.exc.HTTPClientError.code, exc.code)
-            self.assertIn('The server could not comply with the request',
-                          exc.explanation)
+            self.assertIn('Invalid input for minimum', exc.explanation)
 
     def test_create_network_segment_range_failed_with_tunnel_maximum_id(self):
         expected_res = [(constants.TYPE_VXLAN, 2 ** 24),
@@ -249,10 +245,10 @@ def test_create_network_segment_range_failed_with_tunnel_maximum_id(self):
                                     maximum=max_id)
             if network_type == constants.TYPE_GRE:
                 self.assertEqual(webob.exc.HTTPClientError.code, exc.code)
+                self.assertIn('Invalid input for maximum', exc.explanation)
             else:
                 self.assertEqual(webob.exc.HTTPServerError.code, exc.code)
-            self.assertIn('The server could not comply with the request',
-                          exc.explanation)
+                self.assertIn('Invalid network tunnel range', exc.explanation)
 
     def test_update_network_segment_range_set_name(self):
         network_segment_range = self._test_create_network_segment_range()

neutron/tests/unit/extensions/test_segment.py

@@ -121,10 +121,7 @@ def _create_segment(self, fmt, expected_res_status=None, **kwargs):
 
     def _make_segment(self, fmt, **kwargs):
         res = self._create_segment(fmt, **kwargs)
-        if res.status_int >= webob.exc.HTTPClientError.code:
-            res.charset = 'utf8'
-            raise webob.exc.HTTPClientError(
-                code=res.status_int, explanation=str(res))
+        self._check_http_response(res)
         return self.deserialize(fmt, res)
 
     def segment(self, **kwargs):

neutron/tests/unit/plugins/ml2/test_port_binding.py

@@ -374,8 +374,7 @@ def _create_port_binding(self, fmt, port_id, host, tenant_id=None,
 
     def _make_port_binding(self, fmt, port_id, host, **kwargs):
         res = self._create_port_binding(fmt, port_id, host, **kwargs)
-        if res.status_int >= webob.exc.HTTPClientError.code:
-            raise webob.exc.HTTPClientError(code=res.status_int)
+        self._check_http_response(res)
         return self.deserialize(fmt, res)
 
     def _update_port_binding(self, fmt, port_id, host, **kwargs):
@@ -387,8 +386,7 @@ def _update_port_binding(self, fmt, port_id, host, **kwargs):
 
     def _do_update_port_binding(self, fmt, port_id, host, **kwargs):
         res = self._update_port_binding(fmt, port_id, host, **kwargs)
-        if res.status_int >= webob.exc.HTTPClientError.code:
-            raise webob.exc.HTTPClientError(code=res.status_int)
+        self._check_http_response(res)
         return self.deserialize(fmt, res)
 
     def _activate_port_binding(self, port_id, host, raw_response=True):
@@ -401,8 +399,7 @@ def _activate_port_binding(self, port_id, host, raw_response=True):
     def _check_code_and_serialize(self, response, raw_response):
         if raw_response:
             return response
-        if response.status_int >= webob.exc.HTTPClientError.code:
-            raise webob.exc.HTTPClientError(code=response.status_int)
+        self._check_http_response(response)
         return self.deserialize(self.fmt, response)
 
     def _list_port_bindings(self, port_id, params=None, raw_response=True):

neutron/tests/unit/services/qos/test_qos_plugin.py

@@ -1929,26 +1929,23 @@ def _update_rule(self, rule_type, rule_id, **kwargs):
                                           rule_type.replace('_', '-'))
         request = self.new_update_request(resource, data, rule_id, self.fmt)
         res = request.get_response(self.ext_api)
-        if res.status_int >= webob.exc.HTTPClientError.code:
-            raise webob.exc.HTTPClientError(code=res.status_int)
+        self._check_http_response(res)
         return self.deserialize(self.fmt, res)
 
     def _show_rule(self, rule_type, rule_id):
         resource = '%s/alias-%s-rules' % (qos.ALIAS,
                                           rule_type.replace('_', '-'))
         request = self.new_show_request(resource, rule_id, self.fmt)
         res = request.get_response(self.ext_api)
-        if res.status_int >= webob.exc.HTTPClientError.code:
-            raise webob.exc.HTTPClientError(code=res.status_int)
+        self._check_http_response(res)
         return self.deserialize(self.fmt, res)
 
     def _delete_rule(self, rule_type, rule_id):
         resource = '%s/alias-%s-rules' % (qos.ALIAS,
                                           rule_type.replace('_', '-'))
         request = self.new_delete_request(resource, rule_id, self.fmt)
         res = request.get_response(self.ext_api)
-        if res.status_int >= webob.exc.HTTPClientError.code:
-            raise webob.exc.HTTPClientError(code=res.status_int)
+        self._check_http_response(res)
 
     @mock.patch.object(qos_plugin.QoSPlugin, "update_policy_rule")
     def test_update_rule(self, update_policy_rule_mock):