Fixed/added unit tests...more to come

This commit is contained in:
John Wood 2013-04-14 09:26:14 -05:00
parent 1dd058eb4c
commit 4cd6aa2515
10 changed files with 139 additions and 10 deletions

View File

@ -21,6 +21,8 @@ import falcon
from barbican.api.resources import VersionResource
from barbican.api.resources import TenantsResource, TenantResource
from barbican.api.resources import CSRsResource, CSRResource
from barbican.api.resources import CertificatesResource, CertificateResource
# TBD: from barbican.api.resources import SecretsResource, SecretResource
# Resources
@ -31,7 +33,7 @@ TENANT = TenantResource()
# TBD: SECRET = SecretResource()
CSRS = CSRsResource()
CSR = CSRResource()
CERTS = CertifcatesResource()
CERTS = CertificatesResource()
CERT = CertificateResource()
# Routing

View File

@ -4,8 +4,8 @@ import logging
from barbican.version import __version__
from barbican.api import ApiResource, load_body, abort
from barbican.model.models import Tenant, States
from barbican.model.repositories import TenantRepo
from barbican.model.models import Tenant, States, CSR, Certificate
from barbican.model.repositories import TenantRepo, CSRRepo, CertificateRepo
from barbican.queue.resources import QueueResource, StartCSRMessage
from barbican.common import config

View File

@ -415,3 +415,51 @@ class TenantRepo(BaseRepo):
def _do_validate(self, values):
"""Sub-class hook: validate values."""
pass
class CSRRepo(BaseRepo):
"""Repository for the CSR entity."""
def _do_entity_name(self):
"""Sub-class hook: return entity name, such as for debugging."""
return "CSR"
def _do_create_instance(self):
return models.CSR()
def _do_build_query_by_name(self, name, session):
"""Sub-class hook: find entity by name."""
raise TypeError(_("No support for retrieving by 'name' a CSR record."))
def _do_build_get_query(self, entity_id, session):
"""Sub-class hook: build a retrieve query."""
return session.query(models.CSR)\
.filter_by(id=entity_id)
def _do_validate(self, values):
"""Sub-class hook: validate values."""
pass
class CertificateRepo(BaseRepo):
"""Repository for the Certificate entity."""
def _do_entity_name(self):
"""Sub-class hook: return entity name, such as for debugging."""
return "Certificate"
def _do_create_instance(self):
return models.Certificate()
def _do_build_query_by_name(self, name, session):
"""Sub-class hook: find entity by name."""
raise TypeError(_("No support for retrieving by 'name' a Certificate record."))
def _do_build_get_query(self, entity_id, session):
"""Sub-class hook: build a retrieve query."""
return session.query(models.Certificate)\
.filter_by(id=entity_id)
def _do_validate(self, values):
"""Sub-class hook: validate values."""
pass

View File

@ -19,6 +19,7 @@ Queue objects for Cloudkeep's Barbican
from oslo.config import cfg
from barbican.common import config
from barbican.openstack.common.gettextutils import _
queue_opts = [
cfg.StrOpt('queue_api', default='barbican.queue.simple',

View File

@ -17,12 +17,13 @@
Queue Resources related objects and functions.
"""
from oslo.config import cfg
from barbican.openstack.common import importutils
CONF = cfg.CONF
def get_queue_api():
return importutils.import_module(CONF.queue_api)
return importutils.import_module(CONF.queue.queue_api)
class StartCSRMessage(object):

View File

@ -1,6 +1,6 @@
from datetime import datetime
from barbican.api.resources import *
from barbican.model.models import Tenant
from barbican.model.models import *
from barbican.common import config
from mock import MagicMock
@ -59,7 +59,8 @@ class WhenCreatingTenantsUsingTenantsResource(unittest.TestCase):
self.resource.on_post(self.req, self.resp)
self.tenant_repo.find_by_name.assert_called_once_with(name=self.username, suppress_exception=True)
# TBD: Make this work: self.tenant_repo.create_from.assert_called_once_with(unittest.mock.ANY)
args, kwargs = self.tenant_repo.create_from.call_args
assert isinstance(args[0], Tenant)
def test_should_throw_exception_for_tenants_that_exist(self):
self.tenant_repo.find_by_name.return_value = Tenant()
@ -110,5 +111,74 @@ class WhenGettingOrDeletingTenantUsingTenantResource(unittest.TestCase):
self.resource.on_delete(self.req, self.resp, self.tenant.id)
class WhenCreatingCSRsUsingTenantsResource(unittest.TestCase):
def setUp(self):
self.requestor = 'requestor1234'
self.tenant_id = 'tenant1234'
self.csr_repo = MagicMock()
self.csr_repo.create_from.return_value = None
self.queue_resource = MagicMock()
self.queue_resource.send.return_value = None
self.stream = MagicMock()
self.stream.read.return_value = u'{ "requestor" : "%s" }' % self.requestor
self.req = MagicMock()
self.req.stream = self.stream
self.resp = MagicMock()
self.resource = CSRsResource(self.csr_repo, self.queue_resource)
def test_should_add_new_csr(self):
self.resource.on_post(self.req, self.resp, self.tenant_id)
args, kwargs = self.csr_repo.create_from.call_args
assert isinstance(args[0], CSR)
class WhenGettingOrDeletingCSRUsingCSRResource(unittest.TestCase):
def setUp(self):
self.tenant_id = 'tenant1234'
self.requestor = 'requestor1234'
self.csr = CSR()
self.csr.id = "id1"
self.csr.requestor = self.requestor
self.csr_repo = MagicMock()
self.csr_repo.get.return_value = self.csr
self.csr_repo.delete_entity.return_value = None
self.req = MagicMock()
self.resp = MagicMock()
self.resource = CSRResource(self.csr_repo)
def test_should_get_csr(self):
self.resource.on_get(self.req, self.resp, self.tenant_id, self.csr.id)
self.csr_repo.get.assert_called_once_with(entity_id=self.csr.id)
def test_should_delete_csr(self):
self.resource.on_delete(self.req, self.resp, self.tenant_id, self.csr.id)
self.csr_repo.get.assert_called_once_with(entity_id=self.csr.id)
self.csr_repo.delete_entity.assert_called_once_with(self.csr)
def test_should_throw_exception_for_get_when_csr_not_found(self):
self.csr_repo.get.side_effect = exception.NotFound("Test not found exception")
with self.assertRaises(exception.NotFound):
self.resource.on_get(self.req, self.resp, self.tenant_id, self.csr.id)
def test_should_throw_exception_for_delete_when_csr_not_found(self):
self.csr_repo.get.side_effect = exception.NotFound("Test not found exception")
with self.assertRaises(exception.NotFound):
self.resource.on_delete(self.req, self.resp, self.tenant_id, self.csr.id)
if __name__ == '__main__':
unittest.main()

View File

@ -19,6 +19,7 @@ Worker objects for Cloudkeep's Barbican
from oslo.config import cfg
from barbican.common import config
from barbican.openstack.common.gettextutils import _
worker_opts = [
cfg.StrOpt('worker_api', default='barbican.worker.simple',

View File

@ -17,12 +17,13 @@
Queue Resources related objects and functions.
"""
from oslo.config import cfg
from barbican.openstack.common import importutils
CONF = cfg.CONF
def get_worker_api():
return importutils.import_module(CONF.worker_api)
return importutils.import_module(CONF.worker.worker_api)
class WorkerResource(object):
"""Handles Queue related requests"""

View File

@ -18,8 +18,13 @@ Simple Worker API implementation.
"""
from barbican.queue.resources import StartCSRMessage
class StartCSRProcessor(object):
print "Processing CSR with ID = ", csr_id
"""Process the start of CSR processing."""
def process(self, message):
print "Processing CSR with ID = ", message.csr_id
PROCESSES = {StartCSRMessage : StartCSRProcessor()}

View File

@ -1,7 +1,7 @@
[DEFAULT]
# The list of modules to copy from openstack-common
modules=gettextutils,jsonutils,log,local,notifier,timeutils,uuidutils
modules=gettextutils,jsonutils,log,local,notifier,timeutils,uuidutils,importutils
# The base module to hold the copy of openstack.common
base=barbican