openstack/designate
Code Issues Proposed changes

Cleanup cross function / unit imports

Browse Source 
Change-Id: I741bd5342625dd0617c9d319323a343cbb2f8cc5
This commit is contained in:
Erik Olof Gunnar Andersson 2024-01-07 10:00:04 -08:00
parent 2e2b652419
commit ddef40c9b1
7 changed files with 61 additions and 82 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

15
designate/tests/base_fixtures.py
View File

@ -31,6 +31,7 @@ import tempfile
import fixtures
from oslo_log import log as logging
from oslo_serialization import jsonutils
from oslo_utils import importutils
import tooz.coordination
@ -41,6 +42,7 @@ from designate.network_api import fake as fake_network_api
from designate import policy
from designate import rpc
import designate.service
from designate.tests import resources
import designate.utils
@ -49,6 +51,19 @@ CONF = designate.conf.CONF
LOG = logging.getLogger(__name__)
def get_notification_fixture(service, name):
sample_fixture_path = os.path.join(resources.path, 'sample_notifications')
filename = os.path.join(sample_fixture_path, service, f'{name}.json')
if not os.path.exists(filename):
raise Exception('Invalid notification fixture requested')
with open(filename) as fp:
fixture = fp.read()
return jsonutils.loads(fixture)
class CoordinatorFixture(fixtures.Fixture):
def __init__(self, *args, **kwargs):
self._args = args

38
designate/tests/functional/notification_handler/__init__.py
View File

@ -1,38 +0,0 @@
# Copyright 2012 Managed I.T.
#
# Author: Kiall Mac Innes <kiall@managedit.ie>
#
# 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.
import os
from oslo_serialization import jsonutils
from designate.tests import resources
FIXTURES_PATH = os.path.join(resources.path, 'sample_notifications')
class NotificationHandlerMixin:
def get_notification_fixture(self, service, name):
filename = os.path.join(FIXTURES_PATH, service, '%s.json' % name)
if not os.path.exists(filename):
raise Exception('Invalid notification fixture requested')
with open(filename) as fp:
fixture = fp.read()
return jsonutils.loads(fixture)

28
designate/tests/functional/notification_handler/test_neutron.py
View File

@ -18,16 +18,14 @@
from oslo_log import log as logging
from designate.notification_handler import neutron
from designate.tests import base_fixtures
import designate.tests.functional
from designate.tests.functional import notification_handler
LOG = logging.getLogger(__name__)
class NeutronFloatingHandlerTest(
designate.tests.functional.TestCase,
notification_handler.NotificationHandlerMixin):
class NeutronFloatingHandlerTest(designate.tests.functional.TestCase):
def setUp(self):
super().setUp()
@ -42,8 +40,9 @@ class NeutronFloatingHandlerTest(
def test_floatingip_associate(self):
event_type = 'floatingip.update.end'
fixture = self.get_notification_fixture(
'neutron', event_type + '_associate')
fixture = base_fixtures.get_notification_fixture(
'neutron', event_type + '_associate'
)
self.assertIn(event_type, self.plugin.get_event_types())
@ -66,15 +65,16 @@ class NeutronFloatingHandlerTest(
def test_floatingip_disassociate(self):
start_event_type = 'floatingip.update.end'
start_fixture = self.get_notification_fixture(
start_fixture = base_fixtures.get_notification_fixture(
'neutron', start_event_type + '_associate')
self.plugin.process_notification(self.admin_context.to_dict(),
start_event_type,
start_fixture['payload'])
event_type = 'floatingip.update.end'
fixture = self.get_notification_fixture(
'neutron', event_type + '_disassociate')
fixture = base_fixtures.get_notification_fixture(
'neutron', event_type + '_disassociate'
)
self.assertIn(event_type, self.plugin.get_event_types())
@ -97,15 +97,17 @@ class NeutronFloatingHandlerTest(
def test_floatingip_delete(self):
start_event_type = 'floatingip.update.end'
start_fixture = self.get_notification_fixture(
'neutron', start_event_type + '_associate')
start_fixture = base_fixtures.get_notification_fixture(
'neutron', start_event_type + '_associate'
)
self.plugin.process_notification(self.admin_context.to_dict(),
start_event_type,
start_fixture['payload'])
event_type = 'floatingip.delete.start'
fixture = self.get_notification_fixture(
'neutron', event_type)
fixture = base_fixtures.get_notification_fixture(
'neutron', event_type
)
self.assertIn(event_type, self.plugin.get_event_types())

46
designate/tests/functional/notification_handler/test_nova.py
View File

@ -22,16 +22,14 @@ from oslo_log import log as logging
from designate import exceptions
from designate.notification_handler import nova
from designate import objects
from designate.tests import base_fixtures
import designate.tests.functional
from designate.tests.functional import notification_handler
LOG = logging.getLogger(__name__)
class NovaFixedHandlerTest(
designate.tests.functional.TestCase,
notification_handler.NotificationHandlerMixin):
class NovaFixedHandlerTest(designate.tests.functional.TestCase):
def setUp(self):
super().setUp()
@ -50,7 +48,7 @@ class NovaFixedHandlerTest(
def test_instance_create_end(self):
event_type = 'compute.instance.create.end'
fixture = self.get_notification_fixture('nova', event_type)
fixture = base_fixtures.get_notification_fixture('nova', event_type)
self.assertIn(event_type, self.plugin.get_event_types())
@ -80,7 +78,7 @@ class NovaFixedHandlerTest(
group='handler:nova_fixed')
event_type = 'compute.instance.create.end'
fixture = self.get_notification_fixture('nova', event_type)
fixture = base_fixtures.get_notification_fixture('nova', event_type)
# Set the instance display_name to a string containing UTF8.
fixture['payload']['display_name'] = 'Test↟Instance'
@ -117,7 +115,9 @@ class NovaFixedHandlerTest(
def test_instance_delete_start(self):
# Prepare for the test
start_event_type = 'compute.instance.create.end'
start_fixture = self.get_notification_fixture('nova', start_event_type)
start_fixture = base_fixtures.get_notification_fixture(
'nova', start_event_type
)
self.plugin.process_notification(self.admin_context.to_dict(),
start_event_type,
@ -125,7 +125,7 @@ class NovaFixedHandlerTest(
# Now - Onto the real test
event_type = 'compute.instance.delete.start'
fixture = self.get_notification_fixture('nova', event_type)
fixture = base_fixtures.get_notification_fixture('nova', event_type)
self.assertIn(event_type, self.plugin.get_event_types())
@ -157,14 +157,16 @@ class NovaFixedHandlerTest(
def test_instance_delete_start_record_status_changed(self):
start_event_type = 'compute.instance.create.end'
start_fixture = self.get_notification_fixture('nova', start_event_type)
start_fixture = base_fixtures.get_notification_fixture(
'nova', start_event_type
)
self.plugin.process_notification(self.admin_context.to_dict(),
start_event_type,
start_fixture['payload'])
event_type = 'compute.instance.delete.start'
fixture = self.get_notification_fixture('nova', event_type)
fixture = base_fixtures.get_notification_fixture('nova', event_type)
self.assertIn(event_type, self.plugin.get_event_types())
@ -205,7 +207,7 @@ class NovaFixedHandlerTest(
# Prepare for the test
for start_event_type in ['compute.instance.create.end',
'compute.instance.create.end-2']:
start_fixture = self.get_notification_fixture(
start_fixture = base_fixtures.get_notification_fixture(
'nova', start_event_type
)
self.plugin.process_notification(
@ -216,7 +218,7 @@ class NovaFixedHandlerTest(
# Now - Onto the real test
event_type = 'compute.instance.delete.start'
fixture = self.get_notification_fixture('nova', event_type)
fixture = base_fixtures.get_notification_fixture('nova', event_type)
self.assertIn(event_type, self.plugin.get_event_types())
@ -248,7 +250,7 @@ class NovaFixedHandlerTest(
def test_instance_delete_with_no_record(self):
event_type = 'compute.instance.delete.start'
fixture = self.get_notification_fixture('nova', event_type)
fixture = base_fixtures.get_notification_fixture('nova', event_type)
self.assertIn(event_type, self.plugin.get_event_types())
@ -277,14 +279,16 @@ class NovaFixedHandlerTest(
def test_instance_delete_with_no_recordset(self):
start_event_type = 'compute.instance.create.end'
start_fixture = self.get_notification_fixture('nova', start_event_type)
start_fixture = base_fixtures.get_notification_fixture(
'nova', start_event_type
)
self.plugin.process_notification(self.admin_context.to_dict(),
start_event_type,
start_fixture['payload'])
event_type = 'compute.instance.delete.start'
fixture = self.get_notification_fixture('nova', event_type)
fixture = base_fixtures.get_notification_fixture('nova', event_type)
# Make sure we don't fail here, even though there is nothing to
# do, since the recordset we are trying to delete does not actually
@ -296,14 +300,16 @@ class NovaFixedHandlerTest(
def test_instance_delete_with_no_records_in_recordset(self):
start_event_type = 'compute.instance.create.end'
start_fixture = self.get_notification_fixture('nova', start_event_type)
start_fixture = base_fixtures.get_notification_fixture(
'nova', start_event_type
)
self.plugin.process_notification(self.admin_context.to_dict(),
start_event_type,
start_fixture['payload'])
event_type = 'compute.instance.delete.start'
fixture = self.get_notification_fixture('nova', event_type)
fixture = base_fixtures.get_notification_fixture('nova', event_type)
# Make sure we don't fail here, even though there is nothing to
# do, since the recordset we are trying to delete contains no records.
@ -318,7 +324,7 @@ class NovaFixedHandlerTest(
self.config(formatv4=['%(label)s.example.com.'],
formatv6=['%(label)s.example.com.'],
group='handler:nova_fixed')
fixture = self.get_notification_fixture('nova', event_type)
fixture = base_fixtures.get_notification_fixture('nova', event_type)
with mock.patch.object(
self.central_service.storage, 'find_recordset') as finder:
with mock.patch.object(self.central_service, 'update_recordset'):
@ -338,7 +344,7 @@ class NovaFixedHandlerTest(
event_type = 'compute.instance.create.end'
self.config(formatv4=['%(label)s-v4.example.com.'],
group='handler:nova_fixed')
fixture = self.get_notification_fixture('nova', event_type)
fixture = base_fixtures.get_notification_fixture('nova', event_type)
with mock.patch.object(
self.central_service.storage, 'find_recordset') as finder:
with mock.patch.object(self.central_service, 'update_recordset'):
@ -360,7 +366,7 @@ class NovaFixedHandlerTest(
formatv6=['%(label)s-v6.example.com.'],
group='handler:nova_fixed'
)
fixture = self.get_notification_fixture('nova', event_type)
fixture = base_fixtures.get_notification_fixture('nova', event_type)
with mock.patch.object(
self.central_service.storage, 'find_recordset') as finder:
with mock.patch.object(self.central_service, 'update_recordset'):

5
designate/tests/unit/notification_handler/test_fake.py
View File

@ -15,15 +15,12 @@ import oslotest.base
import designate.conf
from designate.notification_handler import fake
from designate.tests.functional import notification_handler
CONF = designate.conf.CONF
class TestFakeHandler(oslotest.base.BaseTestCase,
notification_handler.NotificationHandlerMixin):
class TestFakeHandler(oslotest.base.BaseTestCase):
@mock.patch('designate.rpc.get_client')
def setUp(self, mock_get_instance):
super().setUp()

10
designate/tests/unit/sink/test_notifications.py
View File

@ -22,14 +22,12 @@ from designate import policy
from designate import rpc
from designate.sink import service
from designate.tests import base_fixtures
from designate.tests.functional import notification_handler
CONF = designate.conf.CONF
class TestSinkNotification(oslotest.base.BaseTestCase,
notification_handler.NotificationHandlerMixin):
class TestSinkNotification(oslotest.base.BaseTestCase):
@mock.patch.object(policy, 'init')
@mock.patch.object(rpc, 'get_client')
@ -64,7 +62,7 @@ class TestSinkNotification(oslotest.base.BaseTestCase,
def test_notification(self):
event_type = 'compute.instance.create.end'
fixture = self.get_notification_fixture('nova', event_type)
fixture = base_fixtures.get_notification_fixture('nova', event_type)
self.service.info(self.context, None, event_type,
fixture['payload'], None)
@ -76,7 +74,7 @@ class TestSinkNotification(oslotest.base.BaseTestCase,
def test_notification_with_unknown_event(self):
event_type = 'compute.instance.create.start'
fixture = self.get_notification_fixture('nova', event_type)
fixture = base_fixtures.get_notification_fixture('nova', event_type)
self.service.info(self.context, None, event_type,
fixture['payload'], None)
@ -96,7 +94,7 @@ class TestSinkNotification(oslotest.base.BaseTestCase,
self.service = service.Service()
event_type = 'compute.instance.create.end'
fixture = self.get_notification_fixture('nova', event_type)
fixture = base_fixtures.get_notification_fixture('nova', event_type)
self.service.info(self.context, None, event_type,
fixture['payload'], None)

1
designate/tests/unit/sink/test_service.py
View File

@ -20,7 +20,6 @@ import designate.conf
import designate.rpc
from designate.sink import service
from designate.tests import base_fixtures
import designate.tests.functional
CONF = designate.conf.CONF