openstack/designate
Code Issues Proposed changes

Merge "Add first API test for listing servers"

Browse Source
This commit is contained in:
Jenkins 2012-11-21 13:43:45 +00:00 committed by Gerrit Code Review
commit e98b5b72b6
7 changed files with 238 additions and 73 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

43
moniker/tests/__init__.py
View File

@ -34,7 +34,8 @@ class TestCase(unittest2.TestCase):
database_connection='sqlite://',
rpc_backend='moniker.openstack.common.rpc.impl_fake',
notification_driver=[],
backend_driver='fake'
backend_driver='fake',
auth_strategy='noauth'
)
storage.setup_schema()
@ -62,3 +63,43 @@ class TestCase(unittest2.TestCase):
def get_admin_context(self):
return MonikerContext.get_admin_context()
# Fixture methods
def create_server(self, **kwargs):
context = kwargs.pop('context', self.get_admin_context())
values = dict(
name='ns1.example.org',
ipv4='192.0.2.1',
ipv6='2001:db8::1',
)
values.update(kwargs)
return self.central_service.create_server(context, values=values)
def create_domain(self, **kwargs):
context = kwargs.pop('context', self.get_admin_context())
values = dict(
name='example.com',
email='info@example.com',
)
values.update(kwargs)
return self.central_service.create_domain(context, values=values)
def create_record(self, domain_id, **kwargs):
context = kwargs.pop('context', self.get_admin_context())
values = dict(
name='www.example.com',
type='A',
data='127.0.0.1'
)
values.update(kwargs)
return self.central_service.create_record(context, domain_id,
values=values)

29
moniker/tests/moniker-api-paste.ini Normal file
View File

@ -0,0 +1,29 @@
[composite:osapi_dns]
use = egg:Paste#urlmap
/v1: osapi_dns_api_v1
[composite:osapi_dns_api_v1]
use = call:moniker.api.auth:pipeline_factory
noauth = noauth osapi_dns_app_v1
keystone = authtoken keystonecontext osapi_dns_app_v1
[app:osapi_dns_app_v1]
paste.app_factory = moniker.api.v1:factory
[filter:noauth]
paste.filter_factory = moniker.api.auth:NoAuthMiddleware.factory
[filter:keystonecontext]
paste.filter_factory = moniker.api.auth:KeystoneContextMiddleware.factory
[filter:authtoken]
paste.filter_factory = keystoneclient.middleware.auth_token:filter_factory
service_protocol = http
service_host = 127.0.0.1
service_port = 5000
auth_host = 127.0.0.1
auth_port = 35357
auth_protocol = http
admin_tenant_name = %SERVICE_TENANT_NAME%
admin_user = %SERVICE_USER%
admin_password = %SERVICE_PASSWORD%

20
moniker/tests/test_api/__init__.py Normal file
View File

@ -0,0 +1,20 @@
# 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.
from moniker.tests import TestCase
class ApiTestCase(TestCase):
__test__ = False

29
moniker/tests/test_api/test_service.py Normal file
View File

@ -0,0 +1,29 @@
# 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.
from moniker.tests.test_api import ApiTestCase
class ApiServiceTest(ApiTestCase):
__test__ = True
def setUp(self):
super(ApiServiceTest, self).setUp()
self.service = self.get_api_service()
def test_start_and_stop(self):
# Ensures the start/stop actions don't raise
self.service.start()
self.service.stop()

77
moniker/tests/test_api/test_v1.py Normal file
View File

@ -0,0 +1,77 @@
# 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.
from moniker.openstack.common import log as logging
from moniker.openstack.common import jsonutils as json
from flask import Flask
from moniker.api.v1 import blueprint
from moniker.api.auth import NoAuthMiddleware
from moniker.tests.test_api import ApiTestCase
LOG = logging.getLogger(__name__)
class ApiV1Test(ApiTestCase):
__test__ = True
def setUp(self):
super(ApiV1Test, self).setUp()
# Create a Flask application and register the V1 blueprint
self.app = Flask(__name__)
self.app.register_blueprint(blueprint)
# Inject the NoAuth middleware
self.app.wsgi_app = NoAuthMiddleware(self.app.wsgi_app)
# Obtain a test client
self.client = self.app.test_client()
# Create and start an instance of the central service
self.central_service = self.get_central_service()
self.central_service.start()
def tearDown(self):
self.central_service.stop()
def test_list_servers(self):
response = self.client.get('servers')
response_body = json.loads(response.data)
self.assertEquals(200, response.status_code)
self.assertIn('servers', response_body)
self.assertEqual(0, len(response_body['servers']))
# Create a server
self.create_server()
response = self.client.get('servers')
response_body = json.loads(response.data)
self.assertEquals(200, response.status_code)
self.assertIn('servers', response_body)
self.assertEqual(1, len(response_body['servers']))
# Create a second server
self.create_server(name='ns2.example.org', ipv4='192.0.2.2',
ipv6='2001:db8::2')
response = self.client.get('servers')
response_body = json.loads(response.data)
self.assertEquals(200, response.status_code)
self.assertIn('servers', response_body)
self.assertEqual(2, len(response_body['servers']))

112
moniker/tests/test_central/test_service.py
View File

@ -26,45 +26,7 @@ class CentralServiceTest(CentralTestCase):
def setUp(self):
super(CentralServiceTest, self).setUp()
self.service = self.get_central_service()
def create_server(self, **kwargs):
context = kwargs.pop('context', self.get_admin_context())
values = dict(
name='ns1.example.org',
ipv4='192.0.2.1',
ipv6='2001:db8::1',
)
values.update(kwargs)
return self.service.create_server(context, values=values)
def create_domain(self, **kwargs):
context = kwargs.pop('context', self.get_admin_context())
values = dict(
name='example.com',
email='info@example.com',
)
values.update(kwargs)
return self.service.create_domain(context, values=values)
def create_record(self, domain_id, **kwargs):
context = kwargs.pop('context', self.get_admin_context())
values = dict(
name='www.example.com',
type='A',
data='127.0.0.1'
)
values.update(kwargs)
return self.service.create_record(context, domain_id, values=values)
self.central_service = self.get_central_service()
# Server Tests
def test_create_server(self):
@ -77,7 +39,7 @@ class CentralServiceTest(CentralTestCase):
)
# Create a server
server = self.service.create_server(context, values=values)
server = self.central_service.create_server(context, values=values)
# Ensure all values have been set correctly
self.assertIsNotNone(server['id'])
@ -89,14 +51,14 @@ class CentralServiceTest(CentralTestCase):
context = self.get_admin_context()
# Ensure we have no servers to start with.
servers = self.service.get_servers(context)
servers = self.central_service.get_servers(context)
self.assertEqual(len(servers), 0)
# Create a single server (using default values)
self.create_server()
# Ensure we can retrieve the newly created server
servers = self.service.get_servers(context)
servers = self.central_service.get_servers(context)
self.assertEqual(len(servers), 1)
self.assertEqual(servers[0]['name'], 'ns1.example.org')
@ -105,7 +67,7 @@ class CentralServiceTest(CentralTestCase):
ipv6='2001:db8::2')
# Ensure we can retrieve both servers
servers = self.service.get_servers(context)
servers = self.central_service.get_servers(context)
self.assertEqual(len(servers), 2)
self.assertEqual(servers[0]['name'], 'ns1.example.org')
self.assertEqual(servers[1]['name'], 'ns2.example.org')
@ -118,7 +80,8 @@ class CentralServiceTest(CentralTestCase):
expected_server = self.create_server(name=server_name)
# Retrieve it, and ensure it's the same
server = self.service.get_server(context, expected_server['id'])
server = self.central_service.get_server(context,
expected_server['id'])
self.assertEqual(server['id'], expected_server['id'])
self.assertEqual(server['name'], expected_server['name'])
self.assertEqual(str(server['ipv4']), expected_server['ipv4'])
@ -132,11 +95,12 @@ class CentralServiceTest(CentralTestCase):
# Update the server
values = dict(ipv4='127.0.0.1')
self.service.update_server(context, expected_server['id'],
values=values)
self.central_service.update_server(context, expected_server['id'],
values=values)
# Fetch the server again
server = self.service.get_server(context, expected_server['id'])
server = self.central_service.get_server(context,
expected_server['id'])
# Ensure the server was updated correctly
self.assertEqual(str(server['ipv4']), '127.0.0.1')
@ -148,11 +112,11 @@ class CentralServiceTest(CentralTestCase):
server = self.create_server()
# Delete the server
self.service.delete_server(context, server['id'])
self.central_service.delete_server(context, server['id'])
# Fetch the server again, ensuring an exception is raised
with self.assertRaises(exceptions.ServerNotFound):
self.service.get_server(context, server['id'])
self.central_service.get_server(context, server['id'])
# Domain Tests
def test_create_domain(self):
@ -164,7 +128,7 @@ class CentralServiceTest(CentralTestCase):
)
# Create a domain
domain = self.service.create_domain(context, values=values)
domain = self.central_service.create_domain(context, values=values)
# Ensure all values have been set correctly
self.assertIsNotNone(domain['id'])
@ -175,14 +139,14 @@ class CentralServiceTest(CentralTestCase):
context = self.get_admin_context()
# Ensure we have no domains to start with.
domains = self.service.get_domains(context)
domains = self.central_service.get_domains(context)
self.assertEqual(len(domains), 0)
# Create a single domain (using default values)
self.create_domain()
# Ensure we can retrieve the newly created domain
domains = self.service.get_domains(context)
domains = self.central_service.get_domains(context)
self.assertEqual(len(domains), 1)
self.assertEqual(domains[0]['name'], 'example.com')
@ -190,7 +154,7 @@ class CentralServiceTest(CentralTestCase):
self.create_domain(name='example.net')
# Ensure we can retrieve both domain
domains = self.service.get_domains(context)
domains = self.central_service.get_domains(context)
self.assertEqual(len(domains), 2)
self.assertEqual(domains[0]['name'], 'example.com')
self.assertEqual(domains[1]['name'], 'example.net')
@ -203,7 +167,8 @@ class CentralServiceTest(CentralTestCase):
expected_domain = self.create_domain(name=domain_name)
# Retrieve it, and ensure it's the same
domain = self.service.get_domain(context, expected_domain['id'])
domain = self.central_service.get_domain(context,
expected_domain['id'])
self.assertEqual(domain['id'], expected_domain['id'])
self.assertEqual(domain['name'], expected_domain['name'])
self.assertEqual(domain['email'], expected_domain['email'])
@ -216,11 +181,12 @@ class CentralServiceTest(CentralTestCase):
# Update the domain
values = dict(email='new@example.com')
self.service.update_domain(context, expected_domain['id'],
values=values)
self.central_service.update_domain(context, expected_domain['id'],
values=values)
# Fetch the domain again
domain = self.service.get_domain(context, expected_domain['id'])
domain = self.central_service.get_domain(context,
expected_domain['id'])
# Ensure the domain was updated correctly
self.assertEqual(domain['email'], 'new@example.com')
@ -232,11 +198,11 @@ class CentralServiceTest(CentralTestCase):
domain = self.create_domain()
# Delete the domain
self.service.delete_domain(context, domain['id'])
self.central_service.delete_domain(context, domain['id'])
# Fetch the domain again, ensuring an exception is raised
with self.assertRaises(exceptions.DomainNotFound):
self.service.get_domain(context, domain['id'])
self.central_service.get_domain(context, domain['id'])
# Record Tests
def test_create_record(self):
@ -250,8 +216,8 @@ class CentralServiceTest(CentralTestCase):
)
# Create a record
record = self.service.create_record(context, domain['id'],
values=values)
record = self.central_service.create_record(context, domain['id'],
values=values)
# Ensure all values have been set correctly
self.assertIsNotNone(record['id'])
@ -265,14 +231,14 @@ class CentralServiceTest(CentralTestCase):
domain = self.create_domain()
# Ensure we have no records to start with.
records = self.service.get_records(context, domain['id'])
records = self.central_service.get_records(context, domain['id'])
self.assertEqual(len(records), 0)
# Create a single record (using default values)
self.create_record(domain['id'])
# Ensure we can retrieve the newly created record
records = self.service.get_records(context, domain['id'])
records = self.central_service.get_records(context, domain['id'])
self.assertEqual(len(records), 1)
self.assertEqual(records[0]['name'], 'www.example.com')
@ -280,7 +246,7 @@ class CentralServiceTest(CentralTestCase):
self.create_record(domain['id'], name='mail.example.com')
# Ensure we can retrieve both records
records = self.service.get_records(context, domain['id'])
records = self.central_service.get_records(context, domain['id'])
self.assertEqual(len(records), 2)
self.assertEqual(records[0]['name'], 'www.example.com')
self.assertEqual(records[1]['name'], 'mail.example.com')
@ -294,8 +260,8 @@ class CentralServiceTest(CentralTestCase):
expected_record = self.create_record(domain['id'], name=record_name)
# Retrieve it, and ensure it's the same
record = self.service.get_record(context, domain['id'],
expected_record['id'])
record = self.central_service.get_record(context, domain['id'],
expected_record['id'])
self.assertEqual(record['id'], expected_record['id'])
self.assertEqual(record['name'], expected_record['name'])
@ -308,12 +274,13 @@ class CentralServiceTest(CentralTestCase):
# Update the server
values = dict(data='127.0.0.2')
self.service.update_record(context, domain['id'],
expected_record['id'], values=values)
self.central_service.update_record(context, domain['id'],
expected_record['id'],
values=values)
# Fetch the record again
record = self.service.get_record(context, domain['id'],
expected_record['id'])
record = self.central_service.get_record(context, domain['id'],
expected_record['id'])
# Ensure the record was updated correctly
self.assertEqual(record['data'], '127.0.0.2')
@ -326,8 +293,9 @@ class CentralServiceTest(CentralTestCase):
record = self.create_record(domain['id'])
# Delete the record
self.service.delete_record(context, domain['id'], record['id'])
self.central_service.delete_record(context, domain['id'], record['id'])
# Fetch the record again, ensuring an exception is raised
with self.assertRaises(exceptions.RecordNotFound):
self.service.get_record(context, domain['id'], record['id'])
self.central_service.get_record(context, domain['id'],
record['id'])

1
tools/pip-requires
View File

@ -2,6 +2,7 @@ Flask>=0.8
eventlet
jsonschema>=0.6
ipaddr
Paste
PasteDeploy
stevedore
cliff