openstack
/
designate
Code Issues Proposed changes

Improved V2 controller coverage 

- Removed code that was impossible to reach.

Change-Id: Ib3c11d792ff1a5c0110ab2263e7a8a6d5f6baae0
This commit is contained in:
Erik Olof Gunnar Andersson 2023-12-16 16:11:52 -08:00
parent 571d902c59
commit 39a8c8f116
13 changed files with 576 additions and 27 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

18
designate/api/v2/controllers/rest.py
View File

@ -24,6 +24,8 @@
# ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
# (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS
# SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
import inspect
import pecan
@ -63,9 +65,9 @@ class RestController(pecan.rest.RestController):
return criterion
def _handle_post(self, method, remainder, request=None):
'''
"""
Routes ``POST`` actions to the appropriate controller.
'''
"""
# route to a post_all or get if no additional parts are available
if not remainder or remainder == ['']:
controller = self._find_controller('post_all', 'post')
@ -86,9 +88,9 @@ class RestController(pecan.rest.RestController):
pecan.abort(405)
def _handle_patch(self, method, remainder, request=None):
'''
"""
Routes ``PATCH`` actions to the appropriate controller.
'''
"""
# route to a patch_all or get if no additional parts are available
if not remainder or remainder == ['']:
controller = self._find_controller('patch_all', 'patch')
@ -109,9 +111,9 @@ class RestController(pecan.rest.RestController):
pecan.abort(405)
def _handle_put(self, method, remainder, request=None):
'''
"""
Routes ``PUT`` actions to the appropriate controller.
'''
"""
# route to a put_all or get if no additional parts are available
if not remainder or remainder == ['']:
controller = self._find_controller('put_all', 'put')
@ -132,9 +134,9 @@ class RestController(pecan.rest.RestController):
pecan.abort(405)
def _handle_delete(self, method, remainder, request=None):
'''
"""
Routes ``DELETE`` actions to the appropriate controller.
'''
"""
# route to a delete_all or get if no additional parts are available
if not remainder or remainder == ['']:
controller = self._find_controller('delete_all', 'delete')

22
designate/api/v2/controllers/zones/__init__.py
View File

@ -53,7 +53,7 @@ class ZonesController(rest.RestController):
zone = self.central_api.get_zone(context, zone_id)
LOG.info("Retrieved %(zone)s", {'zone': zone})
LOG.info('Retrieved %(zone)s', {'zone': zone})
return DesignateAdapter.render('API_v2', zone, request=request)
@ -76,7 +76,7 @@ class ZonesController(rest.RestController):
zones = self.central_api.find_zones(
context, criterion, marker, limit, sort_key, sort_dir)
LOG.info("Retrieved %(zones)s", {'zones': zones})
LOG.info('Retrieved %(zones)s', {'zones': zones})
return DesignateAdapter.render('API_v2', zones, request=request)
@ -89,9 +89,8 @@ class ZonesController(rest.RestController):
zone = request.body_dict
if isinstance(zone, dict):
if 'type' not in zone:
zone['type'] = 'PRIMARY'
if 'type' not in zone:
zone['type'] = 'PRIMARY'
zone = DesignateAdapter.parse('API_v2', zone, objects.Zone())
zone.validate()
@ -107,11 +106,10 @@ class ZonesController(rest.RestController):
# new zone cannot yet be shared.
zone.shared = False
LOG.info("Created %(zone)s", {'zone': zone})
LOG.info('Created %(zone)s', {'zone': zone})
# Prepare the response headers
# If the zone has been created asynchronously
if zone['status'] == 'PENDING':
response.status_int = 202
else:
@ -141,7 +139,7 @@ class ZonesController(rest.RestController):
zone = self.central_api.get_zone(context, zone_id)
# Don't allow updates to zones that are being deleted
if zone.action == "DELETE":
if zone.action == 'DELETE':
raise exceptions.BadRequest('Can not update a deleting zone')
if request.content_type == 'application/json-patch+json':
@ -170,15 +168,11 @@ class ZonesController(rest.RestController):
# Update and persist the resource
if zone.type == 'SECONDARY' and 'email' in zone.obj_what_changed():
msg = "Changed email is not allowed."
raise exceptions.InvalidObject(msg)
increment_serial = zone.type == 'PRIMARY'
zone = self.central_api.update_zone(
context, zone, increment_serial=increment_serial)
LOG.info("Updated %(zone)s", {'zone': zone})
LOG.info('Updated %(zone)s', {'zone': zone})
if zone.status == 'PENDING':
response.status_int = 202
@ -198,6 +192,6 @@ class ZonesController(rest.RestController):
zone = self.central_api.delete_zone(context, zone_id)
response.status_int = 202
LOG.info("Deleted %(zone)s", {'zone': zone})
LOG.info('Deleted %(zone)s', {'zone': zone})
return DesignateAdapter.render('API_v2', zone, request=request)

10
designate/api/v2/controllers/zones/recordsets.py
View File

@ -70,7 +70,8 @@ class RecordSetsController(rest.RestController):
# SOA recordsets cannot be created manually
if recordset.type == 'SOA':
raise exceptions.BadRequest(
"Creating a SOA recordset is not allowed")
'Creating a SOA recordset is not allowed'
)
# Create the recordset
recordset = self.central_api.create_recordset(
@ -102,7 +103,6 @@ class RecordSetsController(rest.RestController):
# Fetch the existing recordset
recordset = self.central_api.get_recordset(context, zone_id,
recordset_id)
# TODO(graham): Move this further down the stack
if recordset.managed and not context.edit_managed_records:
raise exceptions.BadRequest('Managed records may not be updated')
@ -110,14 +110,16 @@ class RecordSetsController(rest.RestController):
# SOA recordsets cannot be updated manually
if recordset['type'] == 'SOA':
raise exceptions.BadRequest(
'Updating SOA recordsets is not allowed')
'Updating SOA recordsets is not allowed'
)
# NS recordsets at the zone root cannot be manually updated
if recordset['type'] == 'NS':
zone = self.central_api.get_zone(context, zone_id)
if recordset['name'] == zone['name']:
raise exceptions.BadRequest(
'Updating a root zone NS record is not allowed')
'Updating a root zone NS record is not allowed'
)
# Convert to APIv2 Format

3
designate/api/v2/patches.py
View File

@ -33,7 +33,8 @@ class Request(pecan.core.Request):
"""
if self.content_type not in JSON_TYPES:
raise exceptions.UnsupportedContentType(
'Content-type must be application/json')
'Content-type must be application/json'
)
try:
json_dict = jsonutils.load(self.body_file)

21
designate/tests/unit/api/test_root.py
View File

@ -74,6 +74,27 @@ class RootTest(oslotest.base.BaseTestCase):
namespace=mock.ANY, names=['v2'], invoke_on_load=True
)
@mock.patch('stevedore.named.NamedExtensionManager')
def test_v2_root_object_not_found(self, mock_manger):
mock_extension = mock.Mock()
mock_extension.obj.get_path.return_value = '..test.a'
mock_manger.return_value = [mock_extension]
CONF.set_override(
'enabled_extensions_v2',
['v2'],
'service:api'
)
self.assertRaisesRegex(
AttributeError,
"object has no attribute 'test'",
v2_root.RootController
)
mock_manger.assert_called_with(
namespace=mock.ANY, names=['v2'], invoke_on_load=True
)
@mock.patch('stevedore.named.NamedExtensionManager')
def test_v2_root_no_extensions(self, mock_manger):
mock_manger.return_value = []

48
designate/tests/unit/api/v2/test_abandon.py Normal file
View File

@ -0,0 +1,48 @@
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License 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.
from unittest import mock
import oslotest.base
from designate.api.v2.controllers.zones.tasks import abandon
from designate.central import rpcapi
from designate import objects
class TestAbandonAPI(oslotest.base.BaseTestCase):
def setUp(self):
super().setUp()
self.central_api = mock.Mock()
self.zone = objects.Zone(
id='1e8952a5-e5a4-426a-afab-4cd10131a351',
name='example.com.',
email='example@example.com'
)
mock.patch.object(rpcapi.CentralAPI, 'get_instance',
return_value=self.central_api).start()
self.controller = abandon.AbandonController()
@mock.patch('pecan.response')
@mock.patch('pecan.request')
def test_post_all_move_error(self, mock_request, mock_response):
mock_request.environ = {'context': mock.Mock()}
mock_delete_zone = mock.Mock()
mock_delete_zone.deleted_at = None
self.central_api.delete_zone.return_value = mock_delete_zone
self.controller.post_all(self.zone.id)
self.assertEqual(500, mock_response.status_int)

0
designate/tests/unit/api/test_api_v2.py → designate/tests/unit/api/v2/test_api_v2.py
View File

0
designate/tests/unit/api/test_floatingips.py → designate/tests/unit/api/v2/test_floatingips.py
View File

50
designate/tests/unit/api/v2/test_patches.py Normal file
View File

@ -0,0 +1,50 @@
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License 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.
from unittest import mock
import oslotest.base
import testtools
from designate.api.v2 import patches
from designate import exceptions
class TestPatches(oslotest.base.BaseTestCase):
def setUp(self):
super().setUp()
self.environ = {
'CONTENT_TYPE': 'application/json',
}
self.request = patches.Request(self.environ)
def test_unsupported_content_type(self):
self.environ['CONTENT_TYPE'] = 'invalid'
with testtools.ExpectedException(exceptions.UnsupportedContentType):
self.assertIsNone(self.request.body_dict)
@mock.patch('oslo_serialization.jsonutils.load')
def test_request_body_empty(self, mock_load):
mock_load.side_effect = ValueError()
with testtools.ExpectedException(exceptions.EmptyRequestBody):
self.assertIsNone(self.request.body_dict)
@mock.patch('oslo_serialization.jsonutils.load')
def test_invalid_json(self, mock_load):
mock_load.side_effect = ValueError()
self.request.body = b'invalid'
with testtools.ExpectedException(exceptions.InvalidJson):
self.assertIsNone(self.request.body_dict)

45
designate/tests/unit/api/v2/test_quotas.py Normal file
View File

@ -0,0 +1,45 @@
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License 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.
from unittest import mock
import oslotest.base
from designate.api.v2.controllers import quotas
from designate.central import rpcapi
from designate import exceptions
class TestQuotasAPI(oslotest.base.BaseTestCase):
def setUp(self):
super().setUp()
self.central_api = mock.Mock()
mock.patch.object(rpcapi.CentralAPI, 'get_instance',
return_value=self.central_api).start()
self.controller = quotas.QuotasController()
@mock.patch('pecan.response')
@mock.patch('pecan.request')
def test_post_all_move_error(self, mock_request, mock_response):
mock_context = mock.Mock()
mock_context.project_id = None
mock_context.all_tenants = False
mock_request.environ = {'context': mock_context}
self.assertRaisesRegex(
exceptions.MissingProjectID,
'The all-projects flag must be used when using non-project '
'scoped tokens.',
self.controller.patch_one, 'b0758367-4ac7-436d-917e-390d2b3df734'
)

157
designate/tests/unit/api/v2/test_recordsets.py Normal file
View File

@ -0,0 +1,157 @@
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License 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.
from unittest import mock
import oslotest.base
from designate.api.v2.controllers.zones import recordsets
from designate.central import rpcapi
from designate import exceptions
from designate import objects
class TestRecordsetAPI(oslotest.base.BaseTestCase):
def setUp(self):
super().setUp()
self.central_api = mock.Mock()
self.zone = objects.Zone(
id='1e8952a5-e5a4-426a-afab-4cd10131a351',
name='example.com.',
email='example@example.com'
)
mock.patch.object(rpcapi.CentralAPI, 'get_instance',
return_value=self.central_api).start()
self.controller = recordsets.RecordSetsController()
@mock.patch('pecan.response', mock.Mock())
@mock.patch('pecan.request')
def test_post_all_soa_not_allowed(self, mock_request):
mock_request.environ = {'context': mock.Mock()}
mock_request.body_dict = {
'name': 'soa.example.com.',
'type': 'SOA'
}
self.assertRaisesRegex(
exceptions.BadRequest,
'Creating a SOA recordset is not allowed',
self.controller.post_all, self.zone.id
)
@mock.patch('pecan.response')
@mock.patch('pecan.request')
def test_put_one(self, mock_request, mock_response):
mock_context = mock.Mock()
mock_context.edit_managed_records = False
mock_request.environ = {'context': mock_context}
record_set = objects.RecordSet(name='www.example.org.', type='NS')
record_set.records = objects.RecordList(
objects=[objects.Record(data='ns1.example.org.', action='NONE')]
)
self.central_api.get_recordset.return_value = record_set
self.central_api.update_recordset.return_value = record_set
self.central_api.get_zone.return_value = self.zone
self.controller.put_one(
self.zone.id, '99a60ad0-b9ac-4e83-9eee-859e99299bcf'
)
self.assertEqual(200, mock_response.status_int)
@mock.patch('pecan.response', mock.Mock())
@mock.patch('pecan.request')
def test_put_one_managed_not_allowed(self, mock_request):
mock_context = mock.Mock()
mock_context.edit_managed_records = False
mock_request.environ = {'context': mock_context}
record_set = objects.RecordSet(name='www.example.org.', type='A')
record_set.records = objects.RecordList(
objects=[objects.Record(data='192.0.2.1', managed=True)]
)
self.central_api.get_recordset.return_value = record_set
self.assertRaisesRegex(
exceptions.BadRequest,
'Managed records may not be updated',
self.controller.put_one, self.zone.id,
'3a2a2c3a-8f47-4788-8622-231f1c8f19c3'
)
@mock.patch('pecan.response', mock.Mock())
@mock.patch('pecan.request')
def test_put_one_soa_not_allowed(self, mock_request):
mock_context = mock.Mock()
mock_request.environ = {'context': mock_context}
record_set = objects.RecordSet(name='soa.example.org.', type='SOA')
record_set.records = objects.RecordList(
objects=[objects.Record(data='192.0.2.2', managed=True)]
)
self.central_api.get_recordset.return_value = record_set
self.assertRaisesRegex(
exceptions.BadRequest,
'Updating SOA recordsets is not allowed',
self.controller.put_one, self.zone.id,
'3a2a2c3a-8f47-4788-8622-231f1c8f19c3'
)
@mock.patch('pecan.response', mock.Mock())
@mock.patch('pecan.request')
def test_put_one_update_root_ns_not_allowed(self, mock_request):
mock_context = mock.Mock()
mock_request.environ = {'context': mock_context}
record_set = objects.RecordSet(name='example.com.', type='NS')
record_set.records = objects.RecordList(
objects=[objects.Record(data='192.0.2.3', managed=True)]
)
self.central_api.get_recordset.return_value = record_set
self.central_api.get_zone.return_value = self.zone
self.assertRaisesRegex(
exceptions.BadRequest,
'Updating a root zone NS record is not allowed',
self.controller.put_one, self.zone.id,
'3a2a2c3a-8f47-4788-8622-231f1c8f19c3'
)
@mock.patch('pecan.response', mock.Mock())
@mock.patch('pecan.request')
def test_delete_one_soa_not_allowed(self, mock_request):
mock_request.environ = {'context': mock.Mock()}
record_set = objects.RecordSet(name='soa.example.com.', type='SOA')
record_set.records = objects.RecordList(
objects=[objects.Record(data='192.0.2.4')]
)
self.central_api.get_recordset.return_value = record_set
self.assertRaisesRegex(
exceptions.BadRequest,
'Deleting a SOA recordset is not allowed',
self.controller.delete_one, self.zone.id,
'3a2a2c3a-8f47-4788-8622-231f1c8f19c3'
)

146
designate/tests/unit/api/v2/test_rest_controller.py Normal file
View File

@ -0,0 +1,146 @@
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License 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.
from unittest import mock
import oslotest.base
from webob import exc
from designate.api.v2.controllers import rest
class TestRestController(oslotest.base.BaseTestCase):
def setUp(self):
super().setUp()
self.controller = rest.RestController()
def test_handle_post(self):
self.controller._find_controller = mock.Mock()
self.controller._find_controller.return_value = mock.Mock()
self.assertEqual(
(mock.ANY, []), self.controller._handle_post(mock.Mock(), None)
)
def test_handle_patch(self):
self.controller._find_controller = mock.Mock()
self.controller._find_controller.return_value = mock.Mock()
self.assertEqual(
(mock.ANY, []), self.controller._handle_patch(mock.Mock(), None)
)
def test_handle_put(self):
self.controller._find_controller = mock.Mock()
self.controller._find_controller.return_value = mock.Mock()
self.assertEqual(
(mock.ANY, []), self.controller._handle_put(mock.Mock(), None)
)
def test_handle_delete(self):
self.controller._find_controller = mock.Mock()
self.controller._find_controller.return_value = mock.Mock()
self.assertEqual(
(mock.ANY, []), self.controller._handle_delete(mock.Mock(), None)
)
def test_handle_post_method_not_allowed(self):
self.controller._find_controller = mock.Mock()
self.controller._find_controller.return_value = None
self.assertRaisesRegex(
exc.HTTPMethodNotAllowed,
'The server could not comply with the request since it is either '
'malformed or otherwise incorrect.',
self.controller._handle_post, mock.Mock(), None
)
def test_handle_patch_method_not_allowed(self):
self.controller._find_controller = mock.Mock()
self.controller._find_controller.return_value = None
self.assertRaisesRegex(
exc.HTTPMethodNotAllowed,
'The server could not comply with the request since it is either '
'malformed or otherwise incorrect.',
self.controller._handle_patch, mock.Mock(), None
)
def test_handle_put_method_not_allowed(self):
self.controller._find_controller = mock.Mock()
self.controller._find_controller.return_value = None
self.assertRaisesRegex(
exc.HTTPMethodNotAllowed,
'The server could not comply with the request since it is either '
'malformed or otherwise incorrect.',
self.controller._handle_put, mock.Mock(), None
)
def test_handle_delete_method_not_allowed(self):
self.controller._find_controller = mock.Mock()
self.controller._find_controller.return_value = None
self.assertRaisesRegex(
exc.HTTPMethodNotAllowed,
'The server could not comply with the request since it is either '
'malformed or otherwise incorrect.',
self.controller._handle_delete, mock.Mock(), None
)
def test_handle_post_controller_not_found(self):
self.controller._find_controller = mock.Mock()
self.controller._find_controller.return_value = None
self.assertRaisesRegex(
exc.HTTPMethodNotAllowed,
'The server could not comply with the request since it is either '
'malformed or otherwise incorrect.',
self.controller._handle_post, mock.Mock(), ['fake']
)
def test_handle_patch_controller_not_found(self):
self.controller._find_controller = mock.Mock()
self.controller._find_controller.return_value = None
self.assertRaisesRegex(
exc.HTTPMethodNotAllowed,
'The server could not comply with the request since it is either '
'malformed or otherwise incorrect.',
self.controller._handle_patch, mock.Mock(), ['fake']
)
def test_handle_put_controller_not_found(self):
self.controller._find_controller = mock.Mock()
self.controller._find_controller.return_value = None
self.assertRaisesRegex(
exc.HTTPMethodNotAllowed,
'The server could not comply with the request since it is either '
'malformed or otherwise incorrect.',
self.controller._handle_put, mock.Mock(), ['fake']
)
def test_handle_delete_controller_not_found(self):
self.controller._find_controller = mock.Mock()
self.controller._find_controller.return_value = None
self.assertRaisesRegex(
exc.HTTPMethodNotAllowed,
'The server could not comply with the request since it is either '
'malformed or otherwise incorrect.',
self.controller._handle_delete, mock.Mock(), ['fake']
)

83
designate/tests/unit/api/v2/test_zones.py Normal file
View File

@ -0,0 +1,83 @@
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License 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.
from unittest import mock
import oslotest.base
from designate.api.v2.controllers import zones
from designate.central import rpcapi
from designate import objects
class TestZonesAPI(oslotest.base.BaseTestCase):
def setUp(self):
super().setUp()
self.central_api = mock.Mock()
self.zone = objects.Zone(
id='1e8952a5-e5a4-426a-afab-4cd10131a351',
name='example.com.',
email='example@example.com',
masters=objects.ZoneMasterList(),
attributes=objects.ZoneAttributeList(),
)
mock.patch.object(rpcapi.CentralAPI, 'get_instance',
return_value=self.central_api).start()
self.controller = zones.ZonesController()
@mock.patch('pecan.response')
@mock.patch('pecan.request')
def test_post_all_zone_error(self, mock_request, mock_response):
mock_response.headers = {}
mock_request.environ = {'context': mock.Mock()}
mock_request.body_dict = {
'name': 'example.com.',
'type': 'PRIMARY',
'email': 'example@example.com',
}
zone = objects.Zone(
name='example.com.',
type='PRIMARY',
email='example@example.com',
status='ERROR',
masters=objects.ZoneMasterList(),
attributes=objects.ZoneAttributeList(),
)
self.central_api.create_zone.return_value = zone
self.controller.post_all()
self.assertEqual(201, mock_response.status_int)
@mock.patch('pecan.response')
@mock.patch('pecan.request')
def test_patch_one_zone_error(self, mock_request, mock_response):
mock_response.headers = {}
mock_request.environ = {'context': mock.Mock()}
mock_request.body_dict = {
'name': 'example.com.',
'type': 'PRIMARY',
'email': 'example@example.com',
}
self.central_api.get_zone.return_value = self.zone
self.central_api.update_zone.return_value = self.zone
self.controller.patch_one(self.zone.id)
self.assertEqual(200, mock_response.status_int)