Blackify openstack.dns

Black used with the '-l 79 -S' flags.

A future change will ignore this commit in git-blame history by adding a
'git-blame-ignore-revs' file.

Change-Id: I5c62d40f41b4632e9f8c59c505a46f8b11866f67
Signed-off-by: Stephen Finucane <stephenfin@redhat.com>
This commit is contained in:
Stephen Finucane 2023-05-05 11:03:20 +01:00
parent 19ec9ba383
commit 10018dbf5b
19 changed files with 437 additions and 292 deletions

View File

@ -16,7 +16,6 @@ from openstack import resource
class Resource(resource.Resource):
@classmethod
def find(cls, session, name_or_id, ignore_missing=True, **params):
"""Find a resource by its name or id.
@ -46,16 +45,17 @@ class Resource(resource.Resource):
# Try to short-circuit by looking directly for a matching ID.
try:
match = cls.existing(
id=name_or_id,
connection=session._get_connection(),
**params)
id=name_or_id, connection=session._get_connection(), **params
)
return match.fetch(session)
except exceptions.SDKException:
# DNS may return 400 when we try to do GET with name
pass
if ('name' in cls._query_mapping._mapping.keys()
and 'name' not in params):
if (
'name' in cls._query_mapping._mapping.keys()
and 'name' not in params
):
params['name'] = name_or_id
data = cls.list(session, **params)
@ -67,7 +67,8 @@ class Resource(resource.Resource):
if ignore_missing:
return None
raise exceptions.ResourceNotFound(
"No %s found for %s" % (cls.__name__, name_or_id))
"No %s found for %s" % (cls.__name__, name_or_id)
)
@classmethod
def _get_next_link(cls, uri, response, data, marker, limit, total_yielded):

View File

@ -87,8 +87,12 @@ class Proxy(proxy.Proxy):
:returns: Zone been deleted
:rtype: :class:`~openstack.dns.v2.zone.Zone`
"""
return self._delete(_zone.Zone, zone, ignore_missing=ignore_missing,
delete_shares=delete_shares)
return self._delete(
_zone.Zone,
zone,
ignore_missing=ignore_missing,
delete_shares=delete_shares,
)
def update_zone(self, zone, **attrs):
"""Update zone attributes
@ -114,8 +118,9 @@ class Proxy(proxy.Proxy):
:returns: :class:`~openstack.dns.v2.zone.Zone`
"""
return self._find(_zone.Zone, name_or_id,
ignore_missing=ignore_missing)
return self._find(
_zone.Zone, name_or_id, ignore_missing=ignore_missing
)
def abandon_zone(self, zone, **attrs):
"""Abandon Zone
@ -225,10 +230,10 @@ class Proxy(proxy.Proxy):
"""
if zone:
zone = self._get_resource(_zone.Zone, zone)
recordset = self._get(
_rs.Recordset, recordset, zone_id=zone.id)
return self._delete(_rs.Recordset, recordset,
ignore_missing=ignore_missing)
recordset = self._get(_rs.Recordset, recordset, zone_id=zone.id)
return self._delete(
_rs.Recordset, recordset, ignore_missing=ignore_missing
)
def find_recordset(self, zone, name_or_id, ignore_missing=True, **query):
"""Find a single recordset
@ -278,8 +283,9 @@ class Proxy(proxy.Proxy):
:returns: The results of zone creation.
:rtype: :class:`~openstack.dns.v2.zone_import.ZoneImport`
"""
return self._create(_zone_import.ZoneImport, prepend_key=False,
**attrs)
return self._create(
_zone_import.ZoneImport, prepend_key=False, **attrs
)
def get_zone_import(self, zone_import):
"""Get a zone import record
@ -304,8 +310,9 @@ class Proxy(proxy.Proxy):
:returns: None
"""
return self._delete(_zone_import.ZoneImport, zone_import,
ignore_missing=ignore_missing)
return self._delete(
_zone_import.ZoneImport, zone_import, ignore_missing=ignore_missing
)
# ======== Zone Exports ========
def zone_exports(self, **query):
@ -335,11 +342,13 @@ class Proxy(proxy.Proxy):
:rtype: :class:`~openstack.dns.v2.zone_export.ZoneExport`
"""
zone = self._get_resource(_zone.Zone, zone)
return self._create(_zone_export.ZoneExport,
base_path='/zones/%(zone_id)s/tasks/export',
prepend_key=False,
zone_id=zone.id,
**attrs)
return self._create(
_zone_export.ZoneExport,
base_path='/zones/%(zone_id)s/tasks/export',
prepend_key=False,
zone_id=zone.id,
**attrs,
)
def get_zone_export(self, zone_export):
"""Get a zone export record
@ -359,8 +368,11 @@ class Proxy(proxy.Proxy):
:returns: ZoneExport instance.
:rtype: :class:`~openstack.dns.v2.zone_export.ZoneExport`
"""
return self._get(_zone_export.ZoneExport, zone_export,
base_path='/zones/tasks/export/%(id)s/export')
return self._get(
_zone_export.ZoneExport,
zone_export,
base_path='/zones/tasks/export/%(id)s/export',
)
def delete_zone_export(self, zone_export, ignore_missing=True):
"""Delete a zone export
@ -375,8 +387,9 @@ class Proxy(proxy.Proxy):
:returns: None
"""
return self._delete(_zone_export.ZoneExport, zone_export,
ignore_missing=ignore_missing)
return self._delete(
_zone_export.ZoneExport, zone_export, ignore_missing=ignore_missing
)
# ======== FloatingIPs ========
def floating_ips(self, **query):
@ -473,7 +486,8 @@ class Proxy(proxy.Proxy):
base_path='/zones/%(zone_id)s/tasks/transfer_requests',
prepend_key=False,
zone_id=zone.id,
**attrs)
**attrs,
)
def update_zone_transfer_request(self, request, **attrs):
"""Update ZoneTransfer Request attributes
@ -485,8 +499,9 @@ class Proxy(proxy.Proxy):
:rtype: :class:`~openstack.dns.v2.zone_transfer.ZoneTransferRequest`
"""
return self._update(_zone_transfer.ZoneTransferRequest,
request, **attrs)
return self._update(
_zone_transfer.ZoneTransferRequest, request, **attrs
)
def delete_zone_transfer_request(self, request, ignore_missing=True):
"""Delete a ZoneTransfer Request
@ -502,8 +517,11 @@ class Proxy(proxy.Proxy):
:returns: None
"""
return self._delete(_zone_transfer.ZoneTransferRequest, request,
ignore_missing=ignore_missing)
return self._delete(
_zone_transfer.ZoneTransferRequest,
request,
ignore_missing=ignore_missing,
)
def zone_transfer_accepts(self, **query):
"""Retrieve a generator of zone transfer accepts
@ -571,8 +589,9 @@ class Proxy(proxy.Proxy):
:rtype: :class:`~openstack.dns.v2.zone_share.ZoneShare`
"""
zone_obj = self._get_resource(_zone.Zone, zone)
return self._get(_zone_share.ZoneShare, zone_share,
zone_id=zone_obj.id)
return self._get(
_zone_share.ZoneShare, zone_share, zone_id=zone_obj.id
)
def find_zone_share(self, zone, zone_share_id, ignore_missing=True):
"""Find a single zone share
@ -589,8 +608,12 @@ class Proxy(proxy.Proxy):
:returns: :class:`~openstack.dns.v2.zone_share.ZoneShare`
"""
zone_obj = self._get_resource(_zone.Zone, zone)
return self._find(_zone_share.ZoneShare, zone_share_id,
ignore_missing=ignore_missing, zone_id=zone_obj.id)
return self._find(
_zone_share.ZoneShare,
zone_share_id,
ignore_missing=ignore_missing,
zone_id=zone_obj.id,
)
def create_zone_share(self, zone, **attrs):
"""Create a new zone share from attributes
@ -605,8 +628,9 @@ class Proxy(proxy.Proxy):
:rtype: :class:`~openstack.dns.v2.zone_share.ZoneShare`
"""
zone_obj = self._get_resource(_zone.Zone, zone)
return self._create(_zone_share.ZoneShare, zone_id=zone_obj.id,
**attrs)
return self._create(
_zone_share.ZoneShare, zone_id=zone_obj.id, **attrs
)
def delete_zone_share(self, zone, zone_share, ignore_missing=True):
"""Delete a zone share
@ -625,20 +649,25 @@ class Proxy(proxy.Proxy):
:returns: ``None``
"""
zone_obj = self._get_resource(_zone.Zone, zone)
self._delete(_zone_share.ZoneShare, zone_share,
ignore_missing=ignore_missing, zone_id=zone_obj.id)
self._delete(
_zone_share.ZoneShare,
zone_share,
ignore_missing=ignore_missing,
zone_id=zone_obj.id,
)
def _get_cleanup_dependencies(self):
# DNS may depend on floating ip
return {
'dns': {
'before': ['network']
}
}
return {'dns': {'before': ['network']}}
def _service_cleanup(self, dry_run=True, client_status_queue=False,
identified_resources=None,
filters=None, resource_evaluation_fn=None):
def _service_cleanup(
self,
dry_run=True,
client_status_queue=False,
identified_resources=None,
filters=None,
resource_evaluation_fn=None,
):
# Delete all zones
for obj in self.zones():
self._service_cleanup_del_res(
@ -648,7 +677,8 @@ class Proxy(proxy.Proxy):
client_status_queue=client_status_queue,
identified_resources=identified_resources,
filters=filters,
resource_evaluation_fn=resource_evaluation_fn)
resource_evaluation_fn=resource_evaluation_fn,
)
# Unset all floatingIPs
# NOTE: FloatingIPs are not cleaned when filters are set
for obj in self.floating_ips():
@ -659,4 +689,5 @@ class Proxy(proxy.Proxy):
client_status_queue=client_status_queue,
identified_resources=identified_resources,
filters=filters,
resource_evaluation_fn=resource_evaluation_fn)
resource_evaluation_fn=resource_evaluation_fn,
)

View File

@ -16,6 +16,7 @@ from openstack import resource
class FloatingIP(_base.Resource):
"""DNS Floating IP Resource"""
resource_key = ''
resources_key = 'floatingips'
base_path = '/reverse/floatingips'

View File

@ -16,6 +16,7 @@ from openstack import resource
class Recordset(_base.Resource):
"""DNS Recordset Resource"""
resources_key = 'recordsets'
base_path = '/zones/%(zone_id)s/recordsets'
@ -27,8 +28,15 @@ class Recordset(_base.Resource):
allow_list = True
_query_mapping = resource.QueryParameters(
'name', 'type', 'ttl', 'data', 'status', 'description',
'limit', 'marker')
'name',
'type',
'ttl',
'data',
'status',
'description',
'limit',
'marker',
)
#: Properties
#: current action in progress on the resource

View File

@ -18,6 +18,7 @@ from openstack import utils
class Zone(_base.Resource):
"""DNS ZONE Resource"""
resources_key = 'zones'
base_path = '/zones'
@ -31,8 +32,14 @@ class Zone(_base.Resource):
commit_method = "PATCH"
_query_mapping = resource.QueryParameters(
'name', 'type', 'email', 'status', 'description', 'ttl',
'limit', 'marker'
'name',
'type',
'email',
'status',
'description',
'ttl',
'limit',
'marker',
)
#: Properties
@ -88,13 +95,9 @@ class Zone(_base.Resource):
delete_shares = resource.Header('x-designate-delete-shares', type=bool)
def _action(self, session, action, body):
"""Preform actions given the message body.
"""
"""Preform actions given the message body."""
url = utils.urljoin(self.base_path, self.id, 'tasks', action)
response = session.post(
url,
json=body)
response = session.post(url, json=body)
exceptions.raise_from_response(response)
return response

View File

@ -17,6 +17,7 @@ from openstack import resource
class ZoneExport(_base.Resource):
"""DNS Zone Exports Resource"""
resource_key = ''
resources_key = 'exports'
base_path = '/zones/tasks/export'
@ -27,9 +28,7 @@ class ZoneExport(_base.Resource):
allow_delete = True
allow_list = True
_query_mapping = resource.QueryParameters(
'zone_id', 'message', 'status'
)
_query_mapping = resource.QueryParameters('zone_id', 'message', 'status')
#: Properties
#: Timestamp when the zone was created
@ -74,14 +73,13 @@ class ZoneExport(_base.Resource):
microversion = self._get_microversion(session, action='create')
# Create ZoneExport requires empty body
# skip _prepare_request completely, since we need just empty body
request = resource._Request(
self.base_path,
None,
None
request = resource._Request(self.base_path, None, None)
response = session.post(
request.url,
json=request.body,
headers=request.headers,
microversion=microversion,
)
response = session.post(request.url,
json=request.body, headers=request.headers,
microversion=microversion)
self.microversion = microversion
self._translate_response(response)

View File

@ -17,6 +17,7 @@ from openstack import resource
class ZoneImport(_base.Resource):
"""DNS Zone Import Resource"""
resource_key = ''
resources_key = 'imports'
base_path = '/zones/tasks/import'
@ -27,9 +28,7 @@ class ZoneImport(_base.Resource):
allow_delete = True
allow_list = True
_query_mapping = resource.QueryParameters(
'zone_id', 'message', 'status'
)
_query_mapping = resource.QueryParameters('zone_id', 'message', 'status')
#: Properties
#: Timestamp when the zone was created
@ -75,13 +74,14 @@ class ZoneImport(_base.Resource):
# Create ZoneImport requires empty body and 'text/dns' as content-type
# skip _prepare_request completely, since we need just empty body
request = resource._Request(
self.base_path,
None,
{'content-type': 'text/dns'}
self.base_path, None, {'content-type': 'text/dns'}
)
response = session.post(
request.url,
json=request.body,
headers=request.headers,
microversion=microversion,
)
response = session.post(request.url,
json=request.body, headers=request.headers,
microversion=microversion)
self.microversion = microversion
self._translate_response(response)

View File

@ -17,9 +17,7 @@ from openstack import resource
class ZoneTransferBase(_base.Resource):
"""DNS Zone Transfer Request/Accept Base Resource"""
_query_mapping = resource.QueryParameters(
'status'
)
_query_mapping = resource.QueryParameters('status')
#: Properties
#: Timestamp when the resource was created
@ -41,6 +39,7 @@ class ZoneTransferBase(_base.Resource):
class ZoneTransferRequest(ZoneTransferBase):
"""DNS Zone Transfer Request Resource"""
base_path = '/zones/tasks/transfer_requests'
resources_key = 'transfer_requests'
@ -62,6 +61,7 @@ class ZoneTransferRequest(ZoneTransferBase):
class ZoneTransferAccept(ZoneTransferBase):
"""DNS Zone Transfer Accept Resource"""
base_path = '/zones/tasks/transfer_accepts'
resources_key = 'transfer_accepts'

View File

@ -17,7 +17,6 @@ from openstack.tests.functional import base
class TestZone(base.BaseFunctionalTest):
def setUp(self):
super(TestZone, self).setUp()
self.require_service('dns')
@ -35,7 +34,7 @@ class TestZone(base.BaseFunctionalTest):
email='joe@example.org',
type='PRIMARY',
ttl=7200,
description='example zone'
description='example zone',
)
self.addCleanup(self.conn.dns.delete_zone, self.zone)
@ -55,18 +54,22 @@ class TestZone(base.BaseFunctionalTest):
current_ttl + 1,
updated_zone_ttl,
'Failed, updated TTL value is:{} instead of expected:{}'.format(
updated_zone_ttl, current_ttl + 1))
updated_zone_ttl, current_ttl + 1
),
)
def test_create_rs(self):
zone = self.conn.dns.get_zone(self.zone)
self.assertIsNotNone(self.conn.dns.create_recordset(
zone=zone,
name='www.{zone}'.format(zone=zone.name),
type='A',
description='Example zone rec',
ttl=3600,
records=['192.168.1.1']
))
self.assertIsNotNone(
self.conn.dns.create_recordset(
zone=zone,
name='www.{zone}'.format(zone=zone.name),
type='A',
description='Example zone rec',
ttl=3600,
records=['192.168.1.1'],
)
)
def test_delete_zone_with_shares(self):
zone_name = 'example-{0}.org.'.format(random.randint(1, 10000))
@ -75,17 +78,19 @@ class TestZone(base.BaseFunctionalTest):
email='joe@example.org',
type='PRIMARY',
ttl=7200,
description='example zone'
description='example zone',
)
self.addCleanup(self.conn.dns.delete_zone, zone)
demo_project_id = self.operator_cloud.get_project('demo')['id']
zone_share = self.conn.dns.create_zone_share(
zone, target_project_id=demo_project_id)
zone, target_project_id=demo_project_id
)
self.addCleanup(self.conn.dns.delete_zone_share, zone, zone_share)
# Test that we cannot delete a zone with shares
self.assertRaises(exceptions.BadRequestException,
self.conn.dns.delete_zone, zone)
self.assertRaises(
exceptions.BadRequestException, self.conn.dns.delete_zone, zone
)
self.conn.dns.delete_zone(zone, delete_shares=True)

View File

@ -16,7 +16,6 @@ from openstack.tests.functional import base
class TestZoneShare(base.BaseFunctionalTest):
def setUp(self):
super(TestZoneShare, self).setUp()
self.require_service('dns')
@ -34,19 +33,22 @@ class TestZoneShare(base.BaseFunctionalTest):
email='joe@example.org',
type='PRIMARY',
ttl=7200,
description='example zone for sdk zone share tests'
description='example zone for sdk zone share tests',
)
self.addCleanup(
self.operator_cloud.dns.delete_zone, self.zone, delete_shares=True
)
self.addCleanup(self.operator_cloud.dns.delete_zone, self.zone,
delete_shares=True)
self.project_id = self.operator_cloud.session.get_project_id()
self.demo_project_id = self.user_cloud.session.get_project_id()
def test_create_delete_zone_share(self):
zone_share = self.operator_cloud.dns.create_zone_share(
self.zone, target_project_id=self.demo_project_id)
self.addCleanup(self.operator_cloud.dns.delete_zone_share, self.zone,
zone_share)
self.zone, target_project_id=self.demo_project_id
)
self.addCleanup(
self.operator_cloud.dns.delete_zone_share, self.zone, zone_share
)
self.assertEqual(self.zone.id, zone_share.zone_id)
self.assertEqual(self.project_id, zone_share.project_id)
@ -57,12 +59,17 @@ class TestZoneShare(base.BaseFunctionalTest):
def test_get_zone_share(self):
orig_zone_share = self.operator_cloud.dns.create_zone_share(
self.zone, target_project_id=self.demo_project_id)
self.addCleanup(self.operator_cloud.dns.delete_zone_share, self.zone,
orig_zone_share)
self.zone, target_project_id=self.demo_project_id
)
self.addCleanup(
self.operator_cloud.dns.delete_zone_share,
self.zone,
orig_zone_share,
)
zone_share = self.operator_cloud.dns.get_zone_share(self.zone,
orig_zone_share)
zone_share = self.operator_cloud.dns.get_zone_share(
self.zone, orig_zone_share
)
self.assertEqual(self.zone.id, zone_share.zone_id)
self.assertEqual(self.project_id, zone_share.project_id)
@ -73,12 +80,17 @@ class TestZoneShare(base.BaseFunctionalTest):
def test_find_zone_share(self):
orig_zone_share = self.operator_cloud.dns.create_zone_share(
self.zone, target_project_id=self.demo_project_id)
self.addCleanup(self.operator_cloud.dns.delete_zone_share, self.zone,
orig_zone_share)
self.zone, target_project_id=self.demo_project_id
)
self.addCleanup(
self.operator_cloud.dns.delete_zone_share,
self.zone,
orig_zone_share,
)
zone_share = self.operator_cloud.dns.find_zone_share(
self.zone, orig_zone_share.id)
self.zone, orig_zone_share.id
)
self.assertEqual(self.zone.id, zone_share.zone_id)
self.assertEqual(self.project_id, zone_share.project_id)
@ -88,32 +100,46 @@ class TestZoneShare(base.BaseFunctionalTest):
self.assertEqual(orig_zone_share.updated_at, zone_share.updated_at)
def test_find_zone_share_ignore_missing(self):
zone_share = self.operator_cloud.dns.find_zone_share(self.zone,
'bogus_id')
zone_share = self.operator_cloud.dns.find_zone_share(
self.zone, 'bogus_id'
)
self.assertIsNone(zone_share)
def test_find_zone_share_ignore_missing_false(self):
self.assertRaises(exceptions.ResourceNotFound,
self.operator_cloud.dns.find_zone_share,
self.zone, 'bogus_id', ignore_missing=False)
self.assertRaises(
exceptions.ResourceNotFound,
self.operator_cloud.dns.find_zone_share,
self.zone,
'bogus_id',
ignore_missing=False,
)
def test_list_zone_shares(self):
zone_share = self.operator_cloud.dns.create_zone_share(
self.zone, target_project_id=self.demo_project_id)
self.addCleanup(self.operator_cloud.dns.delete_zone_share, self.zone,
zone_share)
self.zone, target_project_id=self.demo_project_id
)
self.addCleanup(
self.operator_cloud.dns.delete_zone_share, self.zone, zone_share
)
target_ids = [o.target_project_id for o in
self.operator_cloud.dns.zone_shares(self.zone)]
target_ids = [
o.target_project_id
for o in self.operator_cloud.dns.zone_shares(self.zone)
]
self.assertIn(self.demo_project_id, target_ids)
def test_list_zone_shares_with_target_id(self):
zone_share = self.operator_cloud.dns.create_zone_share(
self.zone, target_project_id=self.demo_project_id)
self.addCleanup(self.operator_cloud.dns.delete_zone_share, self.zone,
zone_share)
self.zone, target_project_id=self.demo_project_id
)
self.addCleanup(
self.operator_cloud.dns.delete_zone_share, self.zone, zone_share
)
target_ids = [o.target_project_id for o in
self.operator_cloud.dns.zone_shares(
self.zone, target_project_id=self.demo_project_id)]
target_ids = [
o.target_project_id
for o in self.operator_cloud.dns.zone_shares(
self.zone, target_project_id=self.demo_project_id
)
]
self.assertIn(self.demo_project_id, target_ids)

View File

@ -23,7 +23,6 @@ EXAMPLE = {
class TestVersion(base.TestCase):
def test_basic(self):
sot = version.Version()
self.assertEqual('version', sot.resource_key)

View File

@ -19,18 +19,15 @@ EXAMPLE = {
'status': 'PENDING',
'ptrdname': 'smtp.example.com.',
'description': 'This is a floating ip for 127.0.0.1',
'links': {
'self': 'dummylink/reverse/floatingips/RegionOne:id'
},
'links': {'self': 'dummylink/reverse/floatingips/RegionOne:id'},
'ttl': 600,
'address': '172.24.4.10',
'action': 'CREATE',
'id': IDENTIFIER
'id': IDENTIFIER,
}
class TestFloatingIP(base.TestCase):
def test_basic(self):
sot = fip.FloatingIP()
self.assertEqual('', sot.resource_key)

View File

@ -29,15 +29,20 @@ class TestDnsProxy(test_proxy_base.TestProxyBase):
class TestDnsZone(TestDnsProxy):
def test_zone_create(self):
self.verify_create(self.proxy.create_zone, zone.Zone,
method_kwargs={'name': 'id'},
expected_kwargs={'name': 'id',
'prepend_key': False})
self.verify_create(
self.proxy.create_zone,
zone.Zone,
method_kwargs={'name': 'id'},
expected_kwargs={'name': 'id', 'prepend_key': False},
)
def test_zone_delete(self):
self.verify_delete(
self.proxy.delete_zone, zone.Zone, True,
expected_kwargs={'ignore_missing': True, 'delete_shares': False})
self.proxy.delete_zone,
zone.Zone,
True,
expected_kwargs={'ignore_missing': True, 'delete_shares': False},
)
def test_zone_find(self):
self.verify_find(self.proxy.find_zone, zone.Zone)
@ -56,44 +61,57 @@ class TestDnsZone(TestDnsProxy):
"openstack.dns.v2.zone.Zone.abandon",
self.proxy.abandon_zone,
method_args=[{'zone': 'id'}],
expected_args=[self.proxy])
expected_args=[self.proxy],
)
def test_zone_xfr(self):
self._verify(
"openstack.dns.v2.zone.Zone.xfr",
self.proxy.xfr_zone,
method_args=[{'zone': 'id'}],
expected_args=[self.proxy])
expected_args=[self.proxy],
)
class TestDnsRecordset(TestDnsProxy):
def test_recordset_create(self):
self.verify_create(self.proxy.create_recordset, recordset.Recordset,
method_kwargs={'zone': 'id'},
expected_kwargs={'zone_id': 'id',
'prepend_key': False})
self.verify_create(
self.proxy.create_recordset,
recordset.Recordset,
method_kwargs={'zone': 'id'},
expected_kwargs={'zone_id': 'id', 'prepend_key': False},
)
def test_recordset_delete(self):
self.verify_delete(self.proxy.delete_recordset,
recordset.Recordset, True)
self.verify_delete(
self.proxy.delete_recordset, recordset.Recordset, True
)
def test_recordset_update(self):
self.verify_update(self.proxy.update_recordset, recordset.Recordset)
def test_recordset_get(self):
self.verify_get(self.proxy.get_recordset, recordset.Recordset,
method_kwargs={'zone': 'zid'},
expected_kwargs={'zone_id': 'zid'}
)
self.verify_get(
self.proxy.get_recordset,
recordset.Recordset,
method_kwargs={'zone': 'zid'},
expected_kwargs={'zone_id': 'zid'},
)
def test_recordsets(self):
self.verify_list(self.proxy.recordsets, recordset.Recordset,
expected_kwargs={'base_path': '/recordsets'})
self.verify_list(
self.proxy.recordsets,
recordset.Recordset,
expected_kwargs={'base_path': '/recordsets'},
)
def test_recordsets_zone(self):
self.verify_list(self.proxy.recordsets, recordset.Recordset,
method_kwargs={'zone': 'zid'},
expected_kwargs={'zone_id': 'zid'})
self.verify_list(
self.proxy.recordsets,
recordset.Recordset,
method_kwargs={'zone': 'zid'},
expected_kwargs={'zone_id': 'zid'},
)
def test_recordset_find(self):
self._verify(
@ -102,7 +120,8 @@ class TestDnsRecordset(TestDnsProxy):
method_args=['zone', 'rs'],
method_kwargs={},
expected_args=[recordset.Recordset, 'rs'],
expected_kwargs={'ignore_missing': True, 'zone_id': 'zone'})
expected_kwargs={'ignore_missing': True, 'zone_id': 'zone'},
)
class TestDnsFloatIP(TestDnsProxy):
@ -113,8 +132,9 @@ class TestDnsFloatIP(TestDnsProxy):
self.verify_get(self.proxy.get_floating_ip, floating_ip.FloatingIP)
def test_floating_ip_update(self):
self.verify_update(self.proxy.update_floating_ip,
floating_ip.FloatingIP)
self.verify_update(
self.proxy.update_floating_ip, floating_ip.FloatingIP
)
def test_floating_ip_unset(self):
self._verify(
@ -123,13 +143,15 @@ class TestDnsFloatIP(TestDnsProxy):
method_args=['value'],
method_kwargs={},
expected_args=[floating_ip.FloatingIP, 'value'],
expected_kwargs={'ptrdname': None})
expected_kwargs={'ptrdname': None},
)
class TestDnsZoneImport(TestDnsProxy):
def test_zone_import_delete(self):
self.verify_delete(self.proxy.delete_zone_import,
zone_import.ZoneImport, True)
self.verify_delete(
self.proxy.delete_zone_import, zone_import.ZoneImport, True
)
def test_zone_import_get(self):
self.verify_get(self.proxy.get_zone_import, zone_import.ZoneImport)
@ -138,117 +160,154 @@ class TestDnsZoneImport(TestDnsProxy):
self.verify_list(self.proxy.zone_imports, zone_import.ZoneImport)
def test_zone_import_create(self):
self.verify_create(self.proxy.create_zone_import,
zone_import.ZoneImport,
method_kwargs={'name': 'id'},
expected_kwargs={'name': 'id',
'prepend_key': False})
self.verify_create(
self.proxy.create_zone_import,
zone_import.ZoneImport,
method_kwargs={'name': 'id'},
expected_kwargs={'name': 'id', 'prepend_key': False},
)
class TestDnsZoneExport(TestDnsProxy):
def test_zone_export_delete(self):
self.verify_delete(self.proxy.delete_zone_export,
zone_export.ZoneExport, True)
self.verify_delete(
self.proxy.delete_zone_export, zone_export.ZoneExport, True
)
def test_zone_export_get(self):
self.verify_get(self.proxy.get_zone_export, zone_export.ZoneExport)
def test_zone_export_get_text(self):
self.verify_get(self.proxy.get_zone_export_text,
zone_export.ZoneExport,
method_args=[{'id': 'zone_export_id_value'}],
expected_kwargs={
'base_path': '/zones/tasks/export/%(id)s/export'
})
self.verify_get(
self.proxy.get_zone_export_text,
zone_export.ZoneExport,
method_args=[{'id': 'zone_export_id_value'}],
expected_kwargs={'base_path': '/zones/tasks/export/%(id)s/export'},
)
def test_zone_exports(self):
self.verify_list(self.proxy.zone_exports, zone_export.ZoneExport)
def test_zone_export_create(self):
self.verify_create(self.proxy.create_zone_export,
zone_export.ZoneExport,
method_args=[{'id': 'zone_id_value'}],
method_kwargs={'name': 'id'},
expected_args=[],
expected_kwargs={'name': 'id',
'zone_id': 'zone_id_value',
'prepend_key': False})
self.verify_create(
self.proxy.create_zone_export,
zone_export.ZoneExport,
method_args=[{'id': 'zone_id_value'}],
method_kwargs={'name': 'id'},
expected_args=[],
expected_kwargs={
'name': 'id',
'zone_id': 'zone_id_value',
'prepend_key': False,
},
)
class TestDnsZoneTransferRequest(TestDnsProxy):
def test_zone_transfer_request_delete(self):
self.verify_delete(self.proxy.delete_zone_transfer_request,
zone_transfer.ZoneTransferRequest, True)
self.verify_delete(
self.proxy.delete_zone_transfer_request,
zone_transfer.ZoneTransferRequest,
True,
)
def test_zone_transfer_request_get(self):
self.verify_get(self.proxy.get_zone_transfer_request,
zone_transfer.ZoneTransferRequest)
self.verify_get(
self.proxy.get_zone_transfer_request,
zone_transfer.ZoneTransferRequest,
)
def test_zone_transfer_requests(self):
self.verify_list(self.proxy.zone_transfer_requests,
zone_transfer.ZoneTransferRequest)
self.verify_list(
self.proxy.zone_transfer_requests,
zone_transfer.ZoneTransferRequest,
)
def test_zone_transfer_request_create(self):
self.verify_create(self.proxy.create_zone_transfer_request,
zone_transfer.ZoneTransferRequest,
method_args=[{'id': 'zone_id_value'}],
method_kwargs={'name': 'id'},
expected_args=[],
expected_kwargs={'name': 'id',
'zone_id': 'zone_id_value',
'prepend_key': False})
self.verify_create(
self.proxy.create_zone_transfer_request,
zone_transfer.ZoneTransferRequest,
method_args=[{'id': 'zone_id_value'}],
method_kwargs={'name': 'id'},
expected_args=[],
expected_kwargs={
'name': 'id',
'zone_id': 'zone_id_value',
'prepend_key': False,
},
)
def test_zone_transfer_request_update(self):
self.verify_update(self.proxy.update_zone_transfer_request,
zone_transfer.ZoneTransferRequest)
self.verify_update(
self.proxy.update_zone_transfer_request,
zone_transfer.ZoneTransferRequest,
)
class TestDnsZoneTransferAccept(TestDnsProxy):
def test_zone_transfer_accept_get(self):
self.verify_get(self.proxy.get_zone_transfer_accept,
zone_transfer.ZoneTransferAccept)
self.verify_get(
self.proxy.get_zone_transfer_accept,
zone_transfer.ZoneTransferAccept,
)
def test_zone_transfer_accepts(self):
self.verify_list(self.proxy.zone_transfer_accepts,
zone_transfer.ZoneTransferAccept)
self.verify_list(
self.proxy.zone_transfer_accepts, zone_transfer.ZoneTransferAccept
)
def test_zone_transfer_accept_create(self):
self.verify_create(self.proxy.create_zone_transfer_accept,
zone_transfer.ZoneTransferAccept)
self.verify_create(
self.proxy.create_zone_transfer_accept,
zone_transfer.ZoneTransferAccept,
)
class TestDnsZoneShare(TestDnsProxy):
def test_zone_share_create(self):
self.verify_create(self.proxy.create_zone_share, zone_share.ZoneShare,
method_kwargs={'zone': 'bogus_id'},
expected_kwargs={'zone_id': 'bogus_id'})
self.verify_create(
self.proxy.create_zone_share,
zone_share.ZoneShare,
method_kwargs={'zone': 'bogus_id'},
expected_kwargs={'zone_id': 'bogus_id'},
)
def test_zone_share_delete(self):
self.verify_delete(
self.proxy.delete_zone_share, zone_share.ZoneShare,
self.proxy.delete_zone_share,
zone_share.ZoneShare,
ignore_missing=True,
method_args={'zone': 'bogus_id', 'zone_share': 'bogus_id'},
expected_args=['zone_share'],
expected_kwargs={'zone_id': 'zone', 'ignore_missing': True})
expected_kwargs={'zone_id': 'zone', 'ignore_missing': True},
)
def test_zone_share_find(self):
self.verify_find(
self.proxy.find_zone_share, zone_share.ZoneShare,
self.proxy.find_zone_share,
zone_share.ZoneShare,
method_args=['zone'],
expected_args=['zone'],
expected_kwargs={'zone_id': 'resource_name',
'ignore_missing': True})
expected_kwargs={
'zone_id': 'resource_name',
'ignore_missing': True,
},
)
def test_zone_share_get(self):
self.verify_get(
self.proxy.get_zone_share, zone_share.ZoneShare,
self.proxy.get_zone_share,
zone_share.ZoneShare,
method_args=['zone', 'zone_share'],
expected_args=['zone_share'],
expected_kwargs={'zone_id': 'zone'})
expected_kwargs={'zone_id': 'zone'},
)
def test_zone_shares(self):
self.verify_list(
self.proxy.zone_shares, zone_share.ZoneShare,
self.proxy.zone_shares,
zone_share.ZoneShare,
method_args=['zone'],
expected_args=[],
expected_kwargs={'zone_id': 'zone'})
expected_kwargs={'zone_id': 'zone'},
)

View File

@ -18,9 +18,7 @@ IDENTIFIER = 'NAME'
EXAMPLE = {
'description': 'This is an example record set.',
'updated_at': None,
'records': [
'10.1.0.2'
],
'records': ['10.1.0.2'],
'ttl': 3600,
'id': IDENTIFIER,
'name': 'example.org.',
@ -31,12 +29,11 @@ EXAMPLE = {
'version': 1,
'type': 'A',
'status': 'ACTIVE',
'action': 'NONE'
'action': 'NONE',
}
class TestRecordset(base.TestCase):
def test_basic(self):
sot = recordset.Recordset()
self.assertIsNone(sot.resource_key)
@ -48,15 +45,19 @@ class TestRecordset(base.TestCase):
self.assertTrue(sot.allow_commit)
self.assertTrue(sot.allow_delete)
self.assertDictEqual({'data': 'data',
'description': 'description',
'limit': 'limit',
'marker': 'marker',
'name': 'name',
'status': 'status',
'ttl': 'ttl',
'type': 'type'},
sot._query_mapping._mapping)
self.assertDictEqual(
{
'data': 'data',
'description': 'description',
'limit': 'limit',
'marker': 'marker',
'name': 'name',
'status': 'status',
'ttl': 'ttl',
'type': 'type',
},
sot._query_mapping._mapping,
)
def test_make_it(self):
sot = recordset.Recordset(**EXAMPLE)

View File

@ -19,9 +19,7 @@ from openstack.tests.unit import base
IDENTIFIER = 'NAME'
EXAMPLE = {
'attributes': {
'tier': 'gold', 'ha': 'true'
},
'attributes': {'tier': 'gold', 'ha': 'true'},
'id': IDENTIFIER,
'name': 'test.org',
'email': 'joe@example.org',
@ -29,12 +27,11 @@ EXAMPLE = {
'ttl': 7200,
'description': 'This is an example zone.',
'status': 'ACTIVE',
'shared': False
'shared': False,
}
class TestZone(base.TestCase):
def setUp(self):
super(TestZone, self).setUp()
self.resp = mock.Mock()
@ -58,15 +55,19 @@ class TestZone(base.TestCase):
self.assertEqual('PATCH', sot.commit_method)
self.assertDictEqual({'description': 'description',
'email': 'email',
'limit': 'limit',
'marker': 'marker',
'name': 'name',
'status': 'status',
'ttl': 'ttl',
'type': 'type'},
sot._query_mapping._mapping)
self.assertDictEqual(
{
'description': 'description',
'email': 'email',
'limit': 'limit',
'marker': 'marker',
'name': 'name',
'status': 'status',
'ttl': 'ttl',
'type': 'type',
},
sot._query_mapping._mapping,
)
def test_make_it(self):
sot = zone.Zone(**EXAMPLE)
@ -83,6 +84,5 @@ class TestZone(base.TestCase):
sot = zone.Zone(**EXAMPLE)
self.assertIsNone(sot.abandon(self.sess))
self.sess.post.assert_called_with(
'zones/NAME/tasks/abandon',
json=None
'zones/NAME/tasks/abandon', json=None
)

View File

@ -23,7 +23,7 @@ EXAMPLE = {
'zone_id': '6625198b-d67d-47dc-8d29-f90bd60f3ac4',
'links': {
'self': 'http://127.0.0.1:9001/v2/zones/tasks/exports/074e805e-f',
'href': 'http://127.0.0.1:9001/v2/zones/6625198b-d67d-'
'href': 'http://127.0.0.1:9001/v2/zones/6625198b-d67d-',
},
'created_at': '2015-05-08T15:43:42.000000',
'updated_at': '2015-05-08T15:43:43.000000',
@ -31,13 +31,12 @@ EXAMPLE = {
'location': 'designate://v2/zones/tasks/exports/8ec17fe1/export',
'message': 'example.com. exported',
'project_id': 'noauth-project',
'id': IDENTIFIER
'id': IDENTIFIER,
}
@mock.patch.object(zone_export.ZoneExport, '_translate_response', mock.Mock())
class TestZoneExport(base.TestCase):
def test_basic(self):
sot = zone_export.ZoneExport()
self.assertEqual('', sot.resource_key)
@ -49,12 +48,16 @@ class TestZoneExport(base.TestCase):
self.assertFalse(sot.allow_commit)
self.assertTrue(sot.allow_delete)
self.assertDictEqual({'limit': 'limit',
'marker': 'marker',
'message': 'message',
'status': 'status',
'zone_id': 'zone_id'},
sot._query_mapping._mapping)
self.assertDictEqual(
{
'limit': 'limit',
'marker': 'marker',
'message': 'message',
'status': 'status',
'zone_id': 'zone_id',
},
sot._query_mapping._mapping,
)
def test_make_it(self):
sot = zone_export.ZoneExport(**EXAMPLE)
@ -76,6 +79,8 @@ class TestZoneExport(base.TestCase):
sot.create(self.session)
self.session.post.assert_called_once_with(
mock.ANY, json=None,
mock.ANY,
json=None,
headers=None,
microversion=self.session.default_microversion)
microversion=self.session.default_microversion,
)