openstack/neutron
Code Issues Proposed changes

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

Browse Source
This commit is contained in:
Zuul 2024-11-11 09:42:29 +00:00 committed by Gerrit Code Review
commit 4899ef09fb
9 changed files with 32 additions and 54 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

26
neutron/tests/unit/db/test_db_base_plugin_v2.py
View File

@ -290,6 +290,14 @@ class NeutronDbPluginV2TestCase(testlib_api.WebTestCase):
'', tenant_id, roles=['reader']) '', tenant_id, roles=['reader'])
return req return req
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, def new_create_request(self, resource, data, fmt=None, id=None,
subresource=None, context=None, tenant_id=None, subresource=None, context=None, tenant_id=None,
as_admin=False, as_service=False): as_admin=False, as_service=False):
@ -590,10 +598,7 @@ class NeutronDbPluginV2TestCase(testlib_api.WebTestCase):
as_admin=as_admin, **kwargs) as_admin=as_admin, **kwargs)
# TODO(salvatore-orlando): do exception handling in this test module # TODO(salvatore-orlando): do exception handling in this test module
# in a uniform way (we do it differently for ports, subnets, and nets # in a uniform way (we do it differently for ports, subnets, and nets
# Things can go wrong - raise HTTP exc with res code only self._check_http_response(res)
# so it can be caught by unit tests
if res.status_int >= webob.exc.HTTPClientError.code:
raise webob.exc.HTTPClientError(code=res.status_int)
return self.deserialize(fmt, res) return self.deserialize(fmt, res)
def _make_subnet(self, fmt, network, gateway, cidr, subnetpool_id=None, def _make_subnet(self, fmt, network, gateway, cidr, subnetpool_id=None,
@ -618,10 +623,7 @@ class NeutronDbPluginV2TestCase(testlib_api.WebTestCase):
ipv6_ra_mode=ipv6_ra_mode, ipv6_ra_mode=ipv6_ra_mode,
ipv6_address_mode=ipv6_address_mode, ipv6_address_mode=ipv6_address_mode,
as_admin=as_admin) as_admin=as_admin)
# Things can go wrong - raise HTTP exc with res code only self._check_http_response(res)
# so it can be caught by unit tests
if res.status_int >= webob.exc.HTTPClientError.code:
raise webob.exc.HTTPClientError(code=res.status_int)
return self.deserialize(fmt, res) return self.deserialize(fmt, res)
def _make_v6_subnet(self, network, ra_addr_mode, ipv6_pd=False): def _make_v6_subnet(self, network, ra_addr_mode, ipv6_pd=False):
@ -649,18 +651,14 @@ class NeutronDbPluginV2TestCase(testlib_api.WebTestCase):
**kwargs) **kwargs)
# Things can go wrong - raise HTTP exc with res code only # Things can go wrong - raise HTTP exc with res code only
# so it can be caught by unit tests # so it can be caught by unit tests
if res.status_int >= webob.exc.HTTPClientError.code: self._check_http_response(res)
raise webob.exc.HTTPClientError(code=res.status_int)
return self.deserialize(fmt, res) return self.deserialize(fmt, res)
def _make_port(self, fmt, net_id, expected_res_status=None, def _make_port(self, fmt, net_id, expected_res_status=None,
as_admin=False, **kwargs): as_admin=False, **kwargs):
res = self._create_port(fmt, net_id, expected_res_status, res = self._create_port(fmt, net_id, expected_res_status,
is_admin=as_admin, **kwargs) is_admin=as_admin, **kwargs)
# Things can go wrong - raise HTTP exc with res code only self._check_http_response(res)
# so it can be caught by unit tests
if res.status_int >= webob.exc.HTTPClientError.code:
raise webob.exc.HTTPClientError(code=res.status_int)
return self.deserialize(fmt, res) return self.deserialize(fmt, res)
def _create_qos_rule(self, fmt, qos_policy_id, rule_type, max_kbps=None, def _create_qos_rule(self, fmt, qos_policy_id, rule_type, max_kbps=None,

4
neutron/tests/unit/db/test_l3_db.py
View File

@ -33,7 +33,6 @@ from neutron_lib.plugins import directory
from neutron_lib.plugins import utils as plugin_utils from neutron_lib.plugins import utils as plugin_utils
from oslo_utils import uuidutils from oslo_utils import uuidutils
import testtools import testtools
import webob.exc
from neutron.db import extraroute_db from neutron.db import extraroute_db
from neutron.db import l3_db from neutron.db import l3_db
@ -1049,8 +1048,7 @@ class L3TestCase(test_db_base_plugin_v2.NeutronDbPluginV2TestCase):
self.fmt, name, True, self.fmt, name, True,
arg_list=(extnet_apidef.EXTERNAL,), arg_list=(extnet_apidef.EXTERNAL,),
as_admin=True, **kwargs) as_admin=True, **kwargs)
if res.status_int >= webob.exc.HTTPClientError.code: self._check_http_response(res)
raise webob.exc.HTTPClientError(code=res.status_int)
return self.deserialize(self.fmt, res) return self.deserialize(self.fmt, res)
def test_update_router_gw_notify(self): def test_update_router_gw_notify(self):

3
neutron/tests/unit/extensions/test_address_group.py
View File

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

3
neutron/tests/unit/extensions/test_address_scope.py
View File

@ -72,8 +72,7 @@ class AddressScopeTestCase(test_db_base_plugin_v2.NeutronDbPluginV2TestCase):
res = self._create_address_scope(fmt, ip_version, res = self._create_address_scope(fmt, ip_version,
admin=admin, tenant_id=tenant_id, admin=admin, tenant_id=tenant_id,
**kwargs) **kwargs)
if res.status_int >= webob.exc.HTTPClientError.code: self._check_http_response(res)
raise webob.exc.HTTPClientError(code=res.status_int)
return self.deserialize(fmt, res) return self.deserialize(fmt, res)
@contextlib.contextmanager @contextlib.contextmanager

9
neutron/tests/unit/extensions/test_local_ip.py
View File

@ -48,16 +48,14 @@ class LocalIPTestBase(test_db_base_plugin_v2.NeutronDbPluginV2TestCase):
req = self.new_create_request('local-ips', local_ip, req = self.new_create_request('local-ips', local_ip,
tenant_id=self._tenant_id, as_admin=True) tenant_id=self._tenant_id, as_admin=True)
res = req.get_response(self.ext_api) res = req.get_response(self.ext_api)
if res.status_int >= webob.exc.HTTPClientError.code: self._check_http_response(res)
raise webob.exc.HTTPClientError(code=res.status_int)
return self.deserialize(self.fmt, res) return self.deserialize(self.fmt, res)
def _update_local_ip(self, lip_id, data): def _update_local_ip(self, lip_id, data):
update_req = self.new_update_request( update_req = self.new_update_request(
'local-ips', data, lip_id, tenant_id=self._tenant_id) 'local-ips', data, lip_id, tenant_id=self._tenant_id)
res = update_req.get_response(self.ext_api) res = update_req.get_response(self.ext_api)
if res.status_int >= webob.exc.HTTPClientError.code: self._check_http_response(res)
raise webob.exc.HTTPClientError(code=res.status_int)
return self.deserialize(self.fmt, res) return self.deserialize(self.fmt, res)
def _create_local_ip_association(self, local_ip_id, fixed_port_id, def _create_local_ip_association(self, local_ip_id, fixed_port_id,
@ -71,8 +69,7 @@ class LocalIPTestBase(test_db_base_plugin_v2.NeutronDbPluginV2TestCase):
subresource='port_associations', subresource='port_associations',
tenant_id=self._tenant_id) tenant_id=self._tenant_id)
res = req.get_response(self.ext_api) res = req.get_response(self.ext_api)
if res.status_int >= webob.exc.HTTPClientError.code: self._check_http_response(res)
raise webob.exc.HTTPClientError(code=res.status_int)
return self.deserialize(self.fmt, res) return self.deserialize(self.fmt, res)
@contextlib.contextmanager @contextlib.contextmanager

18
neutron/tests/unit/extensions/test_network_segment_range.py
View File

@ -69,8 +69,7 @@ class NetworkSegmentRangeTestBase(test_db_base_plugin_v2.
def network_segment_range(self, **kwargs): def network_segment_range(self, **kwargs):
res = self._create_network_segment_range(self.fmt, **kwargs) res = self._create_network_segment_range(self.fmt, **kwargs)
if res.status_int >= webob.exc.HTTPClientError.code: self._check_http_response(res)
raise webob.exc.HTTPClientError(code=res.status_int)
return self.deserialize(self.fmt, res) return self.deserialize(self.fmt, res)
def _test_create_network_segment_range(self, expected=None, **kwargs): def _test_create_network_segment_range(self, expected=None, **kwargs):
@ -165,7 +164,7 @@ class TestNetworkSegmentRange(NetworkSegmentRangeTestBase):
self._test_create_network_segment_range, self._test_create_network_segment_range,
network_type='foo-network-type') network_type='foo-network-type')
self.assertEqual(webob.exc.HTTPClientError.code, exc.code) 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) exc.explanation)
def test_create_network_segment_range_no_physical_network(self): def test_create_network_segment_range_no_physical_network(self):
@ -215,16 +214,14 @@ class TestNetworkSegmentRange(NetworkSegmentRangeTestBase):
self._test_create_network_segment_range, self._test_create_network_segment_range,
minimum=0) minimum=0)
self.assertEqual(webob.exc.HTTPClientError.code, exc.code) self.assertEqual(webob.exc.HTTPClientError.code, exc.code)
self.assertIn('The server could not comply with the request', self.assertIn('Invalid input for minimum', exc.explanation)
exc.explanation)
def test_create_network_segment_range_failed_with_vlan_maximum_id(self): def test_create_network_segment_range_failed_with_vlan_maximum_id(self):
exc = self.assertRaises(webob.exc.HTTPClientError, exc = self.assertRaises(webob.exc.HTTPClientError,
self._test_create_network_segment_range, self._test_create_network_segment_range,
minimum=4095) minimum=4095)
self.assertEqual(webob.exc.HTTPServerError.code, exc.code) self.assertEqual(webob.exc.HTTPServerError.code, exc.code)
self.assertIn('The server could not comply with the request', self.assertIn('Invalid network VLAN range', exc.explanation)
exc.explanation)
def test_create_network_segment_range_failed_with_tunnel_minimum_id(self): def test_create_network_segment_range_failed_with_tunnel_minimum_id(self):
tunnel_type = [constants.TYPE_VXLAN, tunnel_type = [constants.TYPE_VXLAN,
@ -237,8 +234,7 @@ class TestNetworkSegmentRange(NetworkSegmentRangeTestBase):
physical_network=None, physical_network=None,
minimum=0) minimum=0)
self.assertEqual(webob.exc.HTTPClientError.code, exc.code) self.assertEqual(webob.exc.HTTPClientError.code, exc.code)
self.assertIn('The server could not comply with the request', self.assertIn('Invalid input for minimum', exc.explanation)
exc.explanation)
def test_create_network_segment_range_failed_with_tunnel_maximum_id(self): def test_create_network_segment_range_failed_with_tunnel_maximum_id(self):
expected_res = [(constants.TYPE_VXLAN, 2 ** 24), expected_res = [(constants.TYPE_VXLAN, 2 ** 24),
@ -252,10 +248,10 @@ class TestNetworkSegmentRange(NetworkSegmentRangeTestBase):
maximum=max_id) maximum=max_id)
if network_type == constants.TYPE_GRE: if network_type == constants.TYPE_GRE:
self.assertEqual(webob.exc.HTTPClientError.code, exc.code) self.assertEqual(webob.exc.HTTPClientError.code, exc.code)
self.assertIn('Invalid input for maximum', exc.explanation)
else: else:
self.assertEqual(webob.exc.HTTPServerError.code, exc.code) self.assertEqual(webob.exc.HTTPServerError.code, exc.code)
self.assertIn('The server could not comply with the request', self.assertIn('Invalid network tunnel range', exc.explanation)
exc.explanation)
def test_update_network_segment_range_set_name(self): def test_update_network_segment_range_set_name(self):
network_segment_range = self._test_create_network_segment_range() network_segment_range = self._test_create_network_segment_range()

5
neutron/tests/unit/extensions/test_segment.py
View File

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

9
neutron/tests/unit/plugins/ml2/test_port_binding.py
View File

@ -379,8 +379,7 @@ class ExtendedPortBindingTestCase(test_plugin.NeutronDbPluginV2TestCase):
def _make_port_binding(self, fmt, port_id, host, **kwargs): def _make_port_binding(self, fmt, port_id, host, **kwargs):
res = self._create_port_binding(fmt, port_id, host, **kwargs) res = self._create_port_binding(fmt, port_id, host, **kwargs)
if res.status_int >= webob.exc.HTTPClientError.code: self._check_http_response(res)
raise webob.exc.HTTPClientError(code=res.status_int)
return self.deserialize(fmt, res) return self.deserialize(fmt, res)
def _update_port_binding(self, fmt, port_id, host, **kwargs): def _update_port_binding(self, fmt, port_id, host, **kwargs):
@ -393,8 +392,7 @@ class ExtendedPortBindingTestCase(test_plugin.NeutronDbPluginV2TestCase):
def _do_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) res = self._update_port_binding(fmt, port_id, host, **kwargs)
if res.status_int >= webob.exc.HTTPClientError.code: self._check_http_response(res)
raise webob.exc.HTTPClientError(code=res.status_int)
return self.deserialize(fmt, res) return self.deserialize(fmt, res)
def _activate_port_binding(self, port_id, host, raw_response=True): def _activate_port_binding(self, port_id, host, raw_response=True):
@ -408,8 +406,7 @@ class ExtendedPortBindingTestCase(test_plugin.NeutronDbPluginV2TestCase):
def _check_code_and_serialize(self, response, raw_response): def _check_code_and_serialize(self, response, raw_response):
if raw_response: if raw_response:
return response return response
if response.status_int >= webob.exc.HTTPClientError.code: self._check_http_response(response)
raise webob.exc.HTTPClientError(code=response.status_int)
return self.deserialize(self.fmt, response) return self.deserialize(self.fmt, response)
def _list_port_bindings(self, port_id, params=None, raw_response=True): def _list_port_bindings(self, port_id, params=None, raw_response=True):

9
neutron/tests/unit/services/qos/test_qos_plugin.py
View File

@ -1929,8 +1929,7 @@ class TestQoSRuleAlias(test_db_base_plugin_v2.NeutronDbPluginV2TestCase):
request = self.new_update_request(resource, data, rule_id, self.fmt, request = self.new_update_request(resource, data, rule_id, self.fmt,
as_admin=True) as_admin=True)
res = request.get_response(self.ext_api) res = request.get_response(self.ext_api)
if res.status_int >= webob.exc.HTTPClientError.code: self._check_http_response(res)
raise webob.exc.HTTPClientError(code=res.status_int)
return self.deserialize(self.fmt, res) return self.deserialize(self.fmt, res)
def _show_rule(self, rule_type, rule_id): def _show_rule(self, rule_type, rule_id):
@ -1939,8 +1938,7 @@ class TestQoSRuleAlias(test_db_base_plugin_v2.NeutronDbPluginV2TestCase):
request = self.new_show_request(resource, rule_id, self.fmt, request = self.new_show_request(resource, rule_id, self.fmt,
as_admin=True) as_admin=True)
res = request.get_response(self.ext_api) res = request.get_response(self.ext_api)
if res.status_int >= webob.exc.HTTPClientError.code: self._check_http_response(res)
raise webob.exc.HTTPClientError(code=res.status_int)
return self.deserialize(self.fmt, res) return self.deserialize(self.fmt, res)
def _delete_rule(self, rule_type, rule_id): def _delete_rule(self, rule_type, rule_id):
@ -1949,8 +1947,7 @@ class TestQoSRuleAlias(test_db_base_plugin_v2.NeutronDbPluginV2TestCase):
request = self.new_delete_request(resource, rule_id, self.fmt, request = self.new_delete_request(resource, rule_id, self.fmt,
as_admin=True) as_admin=True)
res = request.get_response(self.ext_api) res = request.get_response(self.ext_api)
if res.status_int >= webob.exc.HTTPClientError.code: self._check_http_response(res)
raise webob.exc.HTTPClientError(code=res.status_int)
@mock.patch.object(qos_plugin.QoSPlugin, "update_policy_rule") @mock.patch.object(qos_plugin.QoSPlugin, "update_policy_rule")
def test_update_rule(self, update_policy_rule_mock): def test_update_rule(self, update_policy_rule_mock):