Controller and API changes for backups.

-API controller for backups
-adding swift url to the config
-Fixing failing tests
-Renaming 'instance' param.
 Checking for NotFound models so that the returned error is friendly
-adding feature to list all backups from a specific instance
-Adding checks for creating/deleting/restoring backups when it is not completed
-Adding unit tests for backup controller
-adding check to see if file is in swift
-Adding skeleton code for delete backup in the task manager
-Fixed backups API to pass in the backup_id during create_backup.
-adding int tests for backup controller
-Adding backup list and delete int tests
-Adding list backups for instance test
-Adding quota for create backup

BP: https://blueprints.launchpad.net/reddwarf/+spec/consistent-snapshots

Change-Id: I35c2fefcce4b3009e76ba7232c52dabf502a3ac0

27 files changed, 1219 insertions, 83 deletions

diff --git a/etc/reddwarf/reddwarf-taskmanager.conf.sample b/etc/reddwarf/reddwarf-taskmanager.conf.sample
index 3d434f1c..1f736657 100644
--- a/etc/reddwarf/reddwarf-taskmanager.conf.sample
+++ b/etc/reddwarf/reddwarf-taskmanager.conf.sample
@@ -30,6 +30,7 @@ db_api_implementation = reddwarf.db.sqlalchemy.api
 reddwarf_auth_url = http://0.0.0.0:5000/v2.0
 nova_compute_url = http://localhost:8774/v2
 nova_volume_url = http://localhost:8776/v1
+swift_url = http://localhost:8080/v1/AUTH_
 
 # Config options for enabling volume service
 reddwarf_volume_support = True
diff --git a/etc/reddwarf/reddwarf.conf.sample b/etc/reddwarf/reddwarf.conf.sample
index 729d4389..bf4061fc 100644
--- a/etc/reddwarf/reddwarf.conf.sample
+++ b/etc/reddwarf/reddwarf.conf.sample
@@ -40,6 +40,7 @@ api_extensions_path = reddwarf/extensions
 reddwarf_auth_url = http://0.0.0.0:5000/v2.0
 nova_compute_url = http://localhost:8774/v2
 nova_volume_url = http://localhost:8776/v1
+swift_url = http://localhost:8080/v1/AUTH_
 
 # Config option for showing the IP address that nova doles out
 add_addresses = True
@@ -52,6 +53,7 @@ mount_point = /var/lib/mysql
 max_accepted_volume_size = 10
 max_instances_per_user = 5
 max_volumes_per_user = 100
+max_backups_per_user = 5
 volume_time_out=30
 
 # Config options for rate limits
diff --git a/etc/reddwarf/reddwarf.conf.test b/etc/reddwarf/reddwarf.conf.test
index cbd5ca84..3365a47c 100644
--- a/etc/reddwarf/reddwarf.conf.test
+++ b/etc/reddwarf/reddwarf.conf.test
@@ -67,6 +67,7 @@ mount_point = /var/lib/mysql
 max_accepted_volume_size = 25
 max_instances_per_user = 55
 max_volumes_per_user = 100
+max_backups_per_user = 5
 volume_time_out=30
 
 # Config options for rate limits
diff --git a/reddwarf/backup/models.py b/reddwarf/backup/models.py
index 33281d44..0331d1d9 100644
--- a/reddwarf/backup/models.py
+++ b/reddwarf/backup/models.py
@@ -18,6 +18,11 @@ from reddwarf.common import cfg
 from reddwarf.common import exception
 from reddwarf.db.models import DatabaseModelBase
 from reddwarf.openstack.common import log as logging
+from swiftclient.client import ClientException
+from reddwarf.taskmanager import api
+from reddwarf.common.remote import create_swift_client
+from reddwarf.common import utils
+from reddwarf.quota.quota import run_with_quotas
 
 CONF = cfg.CONF
 LOG = logging.getLogger(__name__)
@@ -37,27 +42,45 @@ class BackupState(object):
 class Backup(object):
 
     @classmethod
-    def create(cls, context, instance_id, name, description=None):
+    def create(cls, context, instance, name, description=None):
         """
         create db record for Backup
         :param cls:
         :param context: tenant_id included
-        :param instance_id:
+        :param instance:
         :param name:
-        :param note:
+        :param description:
         :return:
         """
-        try:
-            db_info = DBBackup.create(name=name,
-                                      description=description,
-                                      tenant_id=context.tenant,
-                                      state=BackupState.NEW,
-                                      instance_id=instance_id,
-                                      deleted=False)
+
+        def _create_resources():
+            # parse the ID from the Ref
+            instance_id = utils.get_id_from_href(instance)
+
+            # verify that the instance exist and can perform actions
+            from reddwarf.instance.models import Instance
+            instance_model = Instance.load(context, instance_id)
+            instance_model.validate_can_perform_action()
+
+            cls.verify_swift_auth_token(context)
+
+            try:
+                db_info = DBBackup.create(name=name,
+                                          description=description,
+                                          tenant_id=context.tenant,
+                                          state=BackupState.NEW,
+                                          instance_id=instance_id,
+                                          deleted=False)
+            except exception.InvalidModelError as ex:
+                LOG.exception("Unable to create Backup record:")
+                raise exception.BackupCreationError(str(ex))
+
+            api.API(context).create_backup(db_info.id, instance_id)
             return db_info
-        except exception.InvalidModelError as ex:
-            LOG.exception("Unable to create Backup record:")
-            raise exception.BackupCreationError(str(ex))
+
+        return run_with_quotas(context.tenant,
+                               {'backups': 1},
+                               _create_resources)
 
     @classmethod
     def running(cls, instance_id, exclude=None):
@@ -115,17 +138,50 @@ class Backup(object):
         return db_info
 
     @classmethod
-    def delete(cls, backup_id):
+    def delete(cls, context, backup_id):
         """
         update Backup table on deleted flag for given Backup
         :param cls:
+        :param context: context containing the tenant id and token
         :param backup_id: Backup uuid
         :return:
         """
-        #TODO: api (service.py) might take care of actual deletion
-        # on remote swift
-        db_info = cls.get_by_id(backup_id)
-        db_info.delete()
+
+        def _delete_resources():
+            backup = cls.get_by_id(backup_id)
+            if backup.is_running:
+                msg = ("Backup %s cannot be delete because it is running." %
+                       backup_id)
+                raise exception.UnprocessableEntity(msg)
+            cls.verify_swift_auth_token(context)
+            api.API(context).delete_backup(backup_id)
+
+        return run_with_quotas(context.tenant,
+                               {'backups': -1},
+                               _delete_resources)
+
+    @classmethod
+    def verify_swift_auth_token(cls, context):
+        try:
+            client = create_swift_client(context)
+            client.get_account()
+        except ClientException:
+            raise exception.SwiftAuthError(tenant_id=context.tenant)
+
+    @classmethod
+    def check_object_exist(cls, context, location):
+        try:
+            parts = location.split('/')
+            obj = parts[-1]
+            container = parts[-2]
+            client = create_swift_client(context)
+            client.head_object(container, obj)
+            return True
+        except ClientException as e:
+            if e.http_status == 404:
+                return False
+            else:
+                raise exception.SwiftAuthError(tenant_id=context.tenant)
 
 
 def persisted_models():
diff --git a/reddwarf/backup/service.py b/reddwarf/backup/service.py
new file mode 100644
index 00000000..5e503aeb
--- /dev/null
+++ b/reddwarf/backup/service.py
@@ -0,0 +1,77 @@
+# vim: tabstop=4 shiftwidth=4 softtabstop=4
+
+# Copyright 2013 OpenStack LLC
+# Copyright 2013 Hewlett-Packard Development Company, L.P.
+#
+#    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 reddwarf.common import wsgi
+from reddwarf.backup import views
+from reddwarf.backup.models import Backup
+from reddwarf.common import exception
+from reddwarf.common import cfg
+from reddwarf.openstack.common import log as logging
+from reddwarf.openstack.common.gettextutils import _
+
+CONF = cfg.CONF
+LOG = logging.getLogger(__name__)
+
+
+class BackupController(wsgi.Controller):
+    """
+    Controller for accessing backups in the OpenStack API.
+    """
+
+    def index(self, req, tenant_id):
+        """
+        Return all backups information for a tenant ID.
+        """
+        LOG.debug("Listing Backups for tenant '%s'" % tenant_id)
+        context = req.environ[wsgi.CONTEXT_KEY]
+        backups = Backup.list(context)
+        return wsgi.Result(views.BackupViews(backups).data(), 200)
+
+    def show(self, req, tenant_id, id):
+        """Return a single backup."""
+        LOG.info(_("Showing a backup for tenant '%s'") % tenant_id)
+        LOG.info(_("id : '%s'\n\n") % id)
+        backup = Backup.get_by_id(id)
+        return wsgi.Result(views.BackupView(backup).data(), 200)
+
+    def create(self, req, body, tenant_id):
+        LOG.debug("Creating a Backup for tenant '%s'" % tenant_id)
+        self._validate_create_body(body)
+        context = req.environ[wsgi.CONTEXT_KEY]
+        data = body['backup']
+        instance = data['instance']
+        name = data['name']
+        desc = data.get('description')
+        backup = Backup.create(context, instance, name, desc)
+        return wsgi.Result(views.BackupView(backup).data(), 202)
+
+    def delete(self, req, tenant_id, id):
+        LOG.debug("Delete Backup for tenant: %s, ID: %s" % (tenant_id, id))
+        context = req.environ[wsgi.CONTEXT_KEY]
+        Backup.delete(context, id)
+        return wsgi.Result(None, 202)
+
+    def _validate_create_body(self, body):
+        try:
+            body['backup']
+            body['backup']['name']
+            body['backup']['instance']
+        except KeyError as e:
+            LOG.error(_("Create Backup Required field(s) "
+                        "- %s") % e)
+            raise exception.ReddwarfError(
+                "Required element/key - %s was not specified" % e)
diff --git a/reddwarf/backup/views.py b/reddwarf/backup/views.py
new file mode 100644
index 00000000..3685e4b9
--- /dev/null
+++ b/reddwarf/backup/views.py
@@ -0,0 +1,48 @@
+# vim: tabstop=4 shiftwidth=4 softtabstop=4
+
+# Copyright 2013 OpenStack LLC.
+# All Rights Reserved.
+#
+#    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.
+
+
+class BackupView(object):
+
+    def __init__(self, backup):
+        self.backup = backup
+
+    def data(self):
+        return {"backup": {
+            "id": self.backup.id,
+            "name": self.backup.name,
+            "description": self.backup.description,
+            "locationRef": self.backup.location,
+            "instance_id": self.backup.instance_id,
+            "created": self.backup.created,
+            "updated": self.backup.updated,
+            "status": self.backup.state
+        }
+        }
+
+
+class BackupViews(object):
+
+    def __init__(self, backups):
+        self.backups = backups
+
+    def data(self):
+        backups = []
+
+        for b in self.backups:
+            backups.append(BackupView(b).data()["backup"])
+        return {"backups": backups}
diff --git a/reddwarf/common/api.py b/reddwarf/common/api.py
index 3c14581a..149a523d 100644
--- a/reddwarf/common/api.py
+++ b/reddwarf/common/api.py
@@ -14,14 +14,11 @@
 
 import routes
 
-from reddwarf.common import exception
 from reddwarf.common import wsgi
-from reddwarf.extensions.mgmt.host.instance import service as hostservice
 from reddwarf.flavor.service import FlavorController
 from reddwarf.instance.service import InstanceController
 from reddwarf.limits.service import LimitsController
-from reddwarf.openstack.common import log as logging
-from reddwarf.openstack.common import rpc
+from reddwarf.backup.service import BackupController
 from reddwarf.versions import VersionsController
 
 
@@ -34,6 +31,7 @@ class API(wsgi.Router):
         self._flavor_router(mapper)
         self._versions_router(mapper)
         self._limits_router(mapper)
+        self._backups_router(mapper)
 
     def _versions_router(self, mapper):
         versions_resource = VersionsController().create_resource()
@@ -43,7 +41,7 @@ class API(wsgi.Router):
         instance_resource = InstanceController().create_resource()
         path = "/{tenant_id}/instances"
         mapper.resource("instance", path, controller=instance_resource,
-                        member={'action': 'POST'})
+                        member={'action': 'POST', 'backups': 'GET'})
 
     def _flavor_router(self, mapper):
         flavor_resource = FlavorController().create_resource()
@@ -55,6 +53,12 @@ class API(wsgi.Router):
         path = "/{tenant_id}/limits"
         mapper.resource("limits", path, controller=limits_resource)
 
+    def _backups_router(self, mapper):
+        backups_resource = BackupController().create_resource()
+        path = "/{tenant_id}/backups"
+        mapper.resource("backups", path, controller=backups_resource,
+                        member={'action': 'POST'})
+
 
 def app_factory(global_conf, **local_conf):
     return API()
diff --git a/reddwarf/common/cfg.py b/reddwarf/common/cfg.py
index 181ee428..ec709b19 100644
--- a/reddwarf/common/cfg.py
+++ b/reddwarf/common/cfg.py
@@ -42,6 +42,7 @@ common_opts = [
                help='Remote implementation for using fake integration code'),
     cfg.StrOpt('nova_compute_url', default='http://localhost:8774/v2'),
     cfg.StrOpt('nova_volume_url', default='http://localhost:8776/v2'),
+    cfg.StrOpt('swift_url', default='http://localhost:8080/v1/AUTH_'),
     cfg.StrOpt('reddwarf_auth_url', default='http://0.0.0.0:5000/v2.0'),
     cfg.StrOpt('backup_swift_container', default='DBaaS-backup'),
     cfg.StrOpt('host', default='0.0.0.0'),
@@ -83,6 +84,8 @@ common_opts = [
                help='default maximum volume size for an instance'),
     cfg.IntOpt('max_volumes_per_user', default=20,
                help='default maximum for total volume used by a tenant'),
+    cfg.IntOpt('max_backups_per_user', default=5,
+               help='default maximum number of backups created by a tenant'),
     cfg.StrOpt('quota_driver',
                default='reddwarf.quota.quota.DbQuotaDriver',
                help='default driver to use for quota checks'),
diff --git a/reddwarf/common/exception.py b/reddwarf/common/exception.py
index c9c06ce8..2a98efc0 100644
--- a/reddwarf/common/exception.py
+++ b/reddwarf/common/exception.py
@@ -249,3 +249,19 @@ class SecurityGroupRuleCreationError(ReddwarfError):
 class SecurityGroupRuleDeletionError(ReddwarfError):
 
     message = _("Failed to delete Security Group Rule.")
+
+
+class BackupNotCompleteError(ReddwarfError):
+
+    message = _("Unable to create instance because backup %(backup_id)s is "
+                "not completed")
+
+
+class BackupFileNotFound(NotFound):
+    message = _("Backup file in %(location)s was not found in the object "
+                "storage.")
+
+
+class SwiftAuthError(ReddwarfError):
+
+    message = _("Swift account not accessible for tenant %(tenant_id)s.")
diff --git a/reddwarf/common/remote.py b/reddwarf/common/remote.py
index d32459a7..60ccd60c 100644
--- a/reddwarf/common/remote.py
+++ b/reddwarf/common/remote.py
@@ -17,13 +17,14 @@
 
 from reddwarf.common import cfg
 from novaclient.v1_1.client import Client
+from swiftclient.client import Connection
 
 
 CONF = cfg.CONF
 COMPUTE_URL = CONF.nova_compute_url
 PROXY_AUTH_URL = CONF.reddwarf_auth_url
 VOLUME_URL = CONF.nova_volume_url
-PROXY_AUTH_URL = CONF.reddwarf_auth_url
+OBJECT_STORE_URL = CONF.swift_url
 
 
 def create_dns_client(context):
@@ -56,12 +57,18 @@ def create_nova_volume_client(context):
     return client
 
 
-if CONF.remote_implementation == "fake":
-    # Override the functions above with fakes.
+def create_swift_client(context):
+    client = Connection(preauthurl=OBJECT_STORE_URL + context.tenant,
+                        preauthtoken=context.auth_token,
+                        tenant_name=context.tenant)
+    return client
 
+# Override the functions above with fakes.
+if CONF.remote_implementation == "fake":
     from reddwarf.tests.fakes.nova import fake_create_nova_client
     from reddwarf.tests.fakes.nova import fake_create_nova_volume_client
     from reddwarf.tests.fakes.guestagent import fake_create_guest_client
+    from reddwarf.tests.fakes.swift import FakeSwiftClient
 
     def create_guest_client(context, id):
         return fake_create_guest_client(context, id)
@@ -71,3 +78,6 @@ if CONF.remote_implementation == "fake":
 
     def create_nova_volume_client(context):
         return fake_create_nova_volume_client(context)
+
+    def create_swift_client(context):
+        return FakeSwiftClient.Connection(context)
diff --git a/reddwarf/common/wsgi.py b/reddwarf/common/wsgi.py
index bd9194d8..95de2f96 100644
--- a/reddwarf/common/wsgi.py
+++ b/reddwarf/common/wsgi.py
@@ -102,7 +102,7 @@ CUSTOM_SERIALIZER_METADATA = {
     # mgmt/account
     'account': {'id': '', 'num_instances': ''},
     # mgmt/quotas
-    'quotas': {'instances': '', 'volumes': ''},
+    'quotas': {'instances': '', 'volumes': '', 'backups': ''},
     #mgmt/instance
     'guest_status': {'state_description': ''},
     #mgmt/instance/diagnostics
@@ -367,6 +367,7 @@ class Controller(object):
         ],
         webob.exc.HTTPUnauthorized: [
             exception.Forbidden,
+            exception.SwiftAuthError,
         ],
         webob.exc.HTTPBadRequest: [
             exception.InvalidModelError,
@@ -383,8 +384,11 @@ class Controller(object):
             exception.UserNotFound,
             exception.DatabaseNotFound,
             exception.QuotaResourceUnknown,
+            exception.BackupFileNotFound
+        ],
+        webob.exc.HTTPConflict: [
+            exception.BackupNotCompleteError,
         ],
-        webob.exc.HTTPConflict: [],
         webob.exc.HTTPRequestEntityTooLarge: [
             exception.OverLimit,
             exception.QuotaExceeded,
diff --git a/reddwarf/instance/models.py b/reddwarf/instance/models.py
index f672a673..362cca1e 100644
--- a/reddwarf/instance/models.py
+++ b/reddwarf/instance/models.py
@@ -17,31 +17,25 @@
 
 """Model classes that form the core of instances functionality."""
 
-import eventlet
-import netaddr
-
 from datetime import datetime
 from novaclient import exceptions as nova_exceptions
 from reddwarf.common import cfg
 from reddwarf.common import exception
-from reddwarf.common import utils
 from reddwarf.common.remote import create_dns_client
 from reddwarf.common.remote import create_guest_client
 from reddwarf.common.remote import create_nova_client
 from reddwarf.common.remote import create_nova_volume_client
 from reddwarf.extensions.security_group.models import SecurityGroup
 from reddwarf.db import models as dbmodels
+from reddwarf.backup.models import Backup
+from reddwarf.quota.quota import run_with_quotas
 from reddwarf.instance.tasks import InstanceTask
 from reddwarf.instance.tasks import InstanceTasks
-from reddwarf.guestagent import models as agent_models
 from reddwarf.taskmanager import api as task_api
 from reddwarf.openstack.common import log as logging
 from reddwarf.openstack.common.gettextutils import _
 
 
-from eventlet import greenthread
-
-
 CONF = cfg.CONF
 LOG = logging.getLogger(__name__)
 
@@ -68,6 +62,7 @@ class InstanceStatus(object):
     FAILED = "FAILED"
     REBOOT = "REBOOT"
     RESIZE = "RESIZE"
+    BACKUP = "BACKUP"
     SHUTDOWN = "SHUTDOWN"
     ERROR = "ERROR"
 
@@ -81,22 +76,6 @@ def validate_volume_size(size):
         raise exception.VolumeQuotaExceeded(msg)
 
 
-def run_with_quotas(tenant_id, deltas, f):
-    """ Quota wrapper """
-
-    from reddwarf.quota.quota import QUOTAS as quota_engine
-    reservations = quota_engine.reserve(tenant_id, **deltas)
-    result = None
-    try:
-        result = f()
-    except:
-        quota_engine.rollback(reservations)
-        raise
-    else:
-        quota_engine.commit(reservations)
-    return result
-
-
 def load_simple_instance_server_status(context, db_info):
     """Loads a server or raises an exception."""
     if 'BUILDING' == db_info.task_status.action:
@@ -202,6 +181,10 @@ class SimpleInstance(object):
                                           "RESIZE"]:
             return self.db_info.server_status
 
+        ### Check if there is a backup running for this instance
+        if Backup.running(self.id):
+            return InstanceStatus.BACKUP
+
         ### Report as Shutdown while deleting, unless there's an error.
         if 'DELETING' == ACTION:
             if self.db_info.server_status in ["ACTIVE", "SHUTDOWN", "DELETED"]:
@@ -430,7 +413,8 @@ class Instance(BuiltInstance):
 
     @classmethod
     def create(cls, context, name, flavor_id, image_id,
-               databases, users, service_type, volume_size):
+               databases, users, service_type, volume_size, backup_id):
+
         def _create_resources():
             client = create_nova_client(context)
             security_groups = None
@@ -439,6 +423,16 @@ class Instance(BuiltInstance):
             except nova_exceptions.NotFound:
                 raise exception.FlavorNotFound(uuid=flavor_id)
 
+            if backup_id is not None:
+                backup_info = Backup.get_by_id(backup_id)
+                if backup_info.is_running:
+                    raise exception.BackupNotCompleteError(backup_id=backup_id)
+
+                location = backup_info.location
+                LOG.info(_("Checking if backup exist in '%s'") % location)
+                if not Backup.check_object_exist(context, location):
+                    raise exception.BackupFileNotFound(location=location)
+
             db_info = DBInstance.create(name=name, flavor_id=flavor_id,
                                         tenant_id=context.tenant,
                                         volume_size=volume_size,
@@ -466,7 +460,7 @@ class Instance(BuiltInstance):
                                                   flavor.ram, image_id,
                                                   databases, users,
                                                   service_type, volume_size,
-                                                  security_groups)
+                                                  security_groups, backup_id)
 
             return SimpleInstance(context, db_info, service_status)
 
@@ -476,7 +470,7 @@ class Instance(BuiltInstance):
                                _create_resources)
 
     def resize_flavor(self, new_flavor_id):
-        self._validate_can_perform_action()
+        self.validate_can_perform_action()
         LOG.debug("resizing instance %s flavor to %s"
                   % (self.id, new_flavor_id))
         # Validate that the flavor can be found and that it isn't the same size
@@ -501,7 +495,7 @@ class Instance(BuiltInstance):
 
     def resize_volume(self, new_size):
         def _resize_resources():
-            self._validate_can_perform_action()
+            self.validate_can_perform_action()
             LOG.info("Resizing volume of instance %s..." % self.id)
             if not self.volume_size:
                 raise exception.BadRequest("Instance %s has no volume."
@@ -522,13 +516,13 @@ class Instance(BuiltInstance):
                                _resize_resources)
 
     def reboot(self):
-        self._validate_can_perform_action()
+        self.validate_can_perform_action()
         LOG.info("Rebooting instance %s..." % self.id)
         self.update_db(task_status=InstanceTasks.REBOOTING)
         task_api.API(self.context).reboot(self.id)
 
     def restart(self):
-        self._validate_can_perform_action()
+        self.validate_can_perform_action()
         LOG.info("Restarting MySQL on instance %s..." % self.id)
         # Set our local status since Nova might not change it quick enough.
         #TODO(tim.simpson): Possible bad stuff can happen if this service
@@ -540,7 +534,7 @@ class Instance(BuiltInstance):
         task_api.API(self.context).restart(self.id)
 
     def migrate(self):
-        self._validate_can_perform_action()
+        self.validate_can_perform_action()
         LOG.info("Migrating instance %s..." % self.id)
         self.update_db(task_status=InstanceTasks.MIGRATING)
         task_api.API(self.context).migrate(self.id)
@@ -549,7 +543,7 @@ class Instance(BuiltInstance):
         LOG.info("Settting task status to NONE on instance %s..." % self.id)
         self.update_db(task_status=InstanceTasks.NONE)
 
-    def _validate_can_perform_action(self):
+    def validate_can_perform_action(self):
         """
         Raises exception if an instance action cannot currently be performed.
         """
@@ -560,6 +554,8 @@ class Instance(BuiltInstance):
             status = self.db_info.task_status
         elif not self.service_status.status.action_is_allowed:
             status = self.status
+        elif Backup.running(self.id):
+            status = InstanceStatus.BACKUP
         else:
             return
         msg = ("Instance is not currently available for an action to be "
diff --git a/reddwarf/instance/service.py b/reddwarf/instance/service.py
index f0e948db..48c0247f 100644
--- a/reddwarf/instance/service.py
+++ b/reddwarf/instance/service.py
@@ -15,7 +15,6 @@
 #    License for the specific language governing permissions and limitations
 #    under the License.
 
-import routes
 import webob.exc
 
 from reddwarf.common import cfg
@@ -26,6 +25,8 @@ from reddwarf.common import wsgi
 from reddwarf.extensions.mysql.common import populate_databases
 from reddwarf.extensions.mysql.common import populate_users
 from reddwarf.instance import models, views
+from reddwarf.backup.models import Backup as backup_model
+from reddwarf.backup import views as backup_views
 from reddwarf.openstack.common import log as logging
 from reddwarf.openstack.common.gettextutils import _
 
@@ -141,6 +142,15 @@ class InstanceController(wsgi.Controller):
                                                    marker)
         return wsgi.Result(paged.data(), 200)
 
+    def backups(self, req, tenant_id, id):
+        """Return all backups for the specified instance."""
+        LOG.info(_("req : '%s'\n\n") % req)
+        LOG.info(_("Indexing backups for instance '%s'") %
+                 id)
+
+        backups = backup_model.list_for_instance(id)
+        return wsgi.Result(backup_views.BackupViews(backups).data(), 200)
+
     def show(self, req, tenant_id, id):
         """Return a single instance."""
         LOG.info(_("req : '%s'\n\n") % req)
@@ -194,9 +204,17 @@ class InstanceController(wsgi.Controller):
         else:
             volume_size = None
 
+        if 'restorePoint' in body['instance']:
+            backupRef = body['instance']['restorePoint']['backupRef']
+            backup_id = utils.get_id_from_href(backupRef)
+
+        else:
+            backup_id = None
+
         instance = models.Instance.create(context, name, flavor_id,
                                           image_id, databases, users,
-                                          service_type, volume_size)
+                                          service_type, volume_size,
+                                          backup_id)
 
         view = views.InstanceDetailView(instance, req=req)
         return wsgi.Result(view.data(), 200)
diff --git a/reddwarf/quota/models.py b/reddwarf/quota/models.py
index c947f07d..204e252a 100644
--- a/reddwarf/quota/models.py
+++ b/reddwarf/quota/models.py
@@ -76,6 +76,7 @@ class Resource(object):
 
     INSTANCES = 'instances'
     VOLUMES = 'volumes'
+    BACKUPS = 'backups'
 
     def __init__(self, name, flag=None):
         """
diff --git a/reddwarf/quota/quota.py b/reddwarf/quota/quota.py
index 046f4fd6..1c666759 100644
--- a/reddwarf/quota/quota.py
+++ b/reddwarf/quota/quota.py
@@ -310,6 +310,22 @@ QUOTAS = QuotaEngine()
 
 ''' Define all kind of resources here '''
 resources = [Resource(Resource.INSTANCES, 'max_instances_per_user'),
-             Resource(Resource.VOLUMES, 'max_volumes_per_user')]
+             Resource(Resource.VOLUMES, 'max_volumes_per_user'),
+             Resource(Resource.BACKUPS, 'max_backups_per_user')]
 
 QUOTAS.register_resources(resources)
+
+
+def run_with_quotas(tenant_id, deltas, f):
+    """ Quota wrapper """
+
+    reservations = QUOTAS.reserve(tenant_id, **deltas)
+    result = None
+    try:
+        result = f()
+    except:
+        QUOTAS.rollback(reservations)
+        raise
+    else:
+        QUOTAS.commit(reservations)
+    return result
diff --git a/reddwarf/taskmanager/api.py b/reddwarf/taskmanager/api.py
index 8de5d732..81333028 100644
--- a/reddwarf/taskmanager/api.py
+++ b/reddwarf/taskmanager/api.py
@@ -88,12 +88,23 @@ class API(ManagerAPI):
         LOG.debug("Making async call to delete instance: %s" % instance_id)
         self._cast("delete_instance", instance_id=instance_id)
 
+    def create_backup(self, backup_id, instance_id):
+        LOG.debug("Making async call to create a backup for instance: %s" %
+                  instance_id)
+        self._cast("create_backup",
+                   backup_id=backup_id,
+                   instance_id=instance_id)
+
+    def delete_backup(self, backup_id):
+        LOG.debug("Making async call to delete backup: %s" % backup_id)
+        self._cast("delete_backup", backup_id=backup_id)
+
     def create_instance(self, instance_id, name, flavor_id, flavor_ram,
-                        image_id, databases, users, service_type, volume_size,
-                        security_groups):
+                        image_id, databases, users, service_type,
+                        volume_size, security_groups, backup_id=None):
         LOG.debug("Making async call to create instance %s " % instance_id)
         self._cast("create_instance", instance_id=instance_id, name=name,
                    flavor_id=flavor_id, flavor_ram=flavor_ram,
                    image_id=image_id, databases=databases, users=users,
                    service_type=service_type, volume_size=volume_size,
-                   security_groups=security_groups)
+                   security_groups=security_groups, backup_id=backup_id)
diff --git a/reddwarf/taskmanager/manager.py b/reddwarf/taskmanager/manager.py
index cbdb0f01..c2d9806d 100644
--- a/reddwarf/taskmanager/manager.py
+++ b/reddwarf/taskmanager/manager.py
@@ -15,22 +15,13 @@
 #    License for the specific language governing permissions and limitations
 #    under the License.
 
-import traceback
-
-from eventlet import greenthread
-
 from reddwarf.common import exception
 from reddwarf.openstack.common import log as logging
 from reddwarf.openstack.common import periodic_task
-from reddwarf.openstack.common.rpc.common import UnsupportedRpcVersion
-from reddwarf.openstack.common.gettextutils import _
 from reddwarf.taskmanager import models
-from reddwarf.taskmanager.models import BuiltInstanceTasks
 from reddwarf.taskmanager.models import FreshInstanceTasks
 
-
 LOG = logging.getLogger(__name__)
-
 RPC_API_VERSION = "1.0"
 
 
@@ -68,9 +59,16 @@ class Manager(periodic_task.PeriodicTasks):
                                                             instance_id)
             instance_tasks.delete_async()
 
+    def delete_backup(self, context, backup_id):
+        models.BackupTasks.delete_backup(backup_id)
+
+    def create_backup(self, context, backup_id, instance_id):
+        instance_tasks = models.BuiltInstanceTasks.load(context, instance_id)
+        instance_tasks.create_backup(backup_id)
+
     def create_instance(self, context, instance_id, name, flavor_id,
                         flavor_ram, image_id, databases, users, service_type,
-                        volume_size, security_groups):
+                        volume_size, security_groups, backup_id):
         instance_tasks = FreshInstanceTasks.load(context, instance_id)
         instance_tasks.create_instance(flavor_id, flavor_ram, image_id,
                                        databases, users, service_type,
diff --git a/reddwarf/taskmanager/models.py b/reddwarf/taskmanager/models.py
index 8328fbe0..3b374bbd 100644
--- a/reddwarf/taskmanager/models.py
+++ b/reddwarf/taskmanager/models.py
@@ -360,6 +360,18 @@ class BuiltInstanceTasks(BuiltInstance):
         action = MigrateAction(self)
         action.execute()
 
+    def create_backup(self, backup_id):
+        # TODO
+        # create a temp volume
+        # nova list
+        # nova show
+        # check in progress - make sure no other snapshot creation in progress
+        # volume create
+        # volume attach
+        # call GA.create_backup()
+        self.guest.create_backup(backup_id)
+        LOG.debug("Called create_backup  %s " % self.id)
+
     def reboot(self):
         try:
             LOG.debug("Instance %s calling stop_db..." % self.id)
diff --git a/reddwarf/tests/api/backups.py b/reddwarf/tests/api/backups.py
new file mode 100644
index 00000000..f2b8f332
--- /dev/null
+++ b/reddwarf/tests/api/backups.py
@@ -0,0 +1,161 @@
+# Copyright 2011 OpenStack LLC.
+# All Rights Reserved.
+#
+#    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 proboscis.asserts import assert_equal
+from proboscis.asserts import assert_not_equal
+from proboscis.asserts import assert_raises
+from proboscis import test
+from proboscis.decorators import time_out
+from reddwarf.tests.util import poll_until
+from reddwarfclient import exceptions
+from reddwarf.tests.api.instances import WaitForGuestInstallationToFinish
+from reddwarf.tests.api.instances import instance_info, assert_unprocessable
+
+
+GROUP = "dbaas.api.backups"
+BACKUP_NAME = 'backup_test'
+BACKUP_DESC = 'test description'
+
+
+backup_info = None
+
+
+@test(depends_on_classes=[WaitForGuestInstallationToFinish],
+      groups=[GROUP])
+class CreateBackups(object):
+
+    @test
+    def test_backup_create_instance_not_found(self):
+        """test create backup with unknown instance"""
+        assert_raises(exceptions.NotFound, instance_info.dbaas.backups.create,
+                      BACKUP_NAME, 'nonexistent_instance', BACKUP_DESC)
+
+    @test
+    def test_backup_create_instance(self):
+        """test create backup for a given instance"""
+        result = instance_info.dbaas.backups.create(BACKUP_NAME,
+                                                    instance_info.id,
+                                                    BACKUP_DESC)
+        assert_equal(BACKUP_NAME, result.name)
+        assert_equal(BACKUP_DESC, result.description)
+        assert_equal(instance_info.id, result.instance_id)
+        assert_equal('NEW', result.status)
+        instance = instance_info.dbaas.instances.list()[0]
+        assert_equal('BACKUP', instance.status)
+        global backup_info
+        backup_info = result
+
+
+@test(runs_after=[CreateBackups],
+      groups=[GROUP])
+class AfterBackupCreation(object):
+
+    @test
+    def test_instance_action_right_after_backup_create(self):
+        """test any instance action while backup is running"""
+        assert_unprocessable(instance_info.dbaas.instances.resize_volume,
+                             instance_info.id, 1)
+
+    @test
+    def test_backup_create_another_backup_running(self):
+        """test create backup when another backup is running"""
+        assert_unprocessable(instance_info.dbaas.backups.create,
+                             'backup_test2', instance_info.id,
+                             'test description2')
+
+    @test
+    def test_backup_delete_still_running(self):
+        """test delete backup when it is running"""
+        result = instance_info.dbaas.backups.list()
+        backup = result[0]
+        assert_unprocessable(instance_info.dbaas.backups.delete, backup.id)
+
+    @test
+    def test_backup_create_quota_exceeded(self):
+        """test quota exceeded when creating a backup"""
+        instance_info.dbaas_admin.quota.update(instance_info.user.tenant_id,
+                                               {'backups': 1})
+        assert_raises(exceptions.OverLimit,
+                      instance_info.dbaas.backups.create,
+                      'Too_many_backups', instance_info.id, BACKUP_DESC)
+
+
+@test(runs_after=[AfterBackupCreation],
+      groups=[GROUP])
+class WaitForBackupCreateToFinish(object):
+    """
+        Wait until the backup create is finished.
+    """
+
+    @test
+    @time_out(60 * 30)
+    def test_backup_created(self):
+        # This version just checks the REST API status.
+        def result_is_active():
+            backup = instance_info.dbaas.backups.get(backup_info.id)
+            if backup.status == "COMPLETED":
+                return True
+            else:
+                assert_not_equal("FAILED", backup.status)
+                return False
+
+        poll_until(result_is_active)
+
+
+@test(depends_on=[WaitForBackupCreateToFinish],
+      groups=[GROUP])
+class ListBackups(object):
+
+    @test
+    def test_backup_list(self):
+        """test list backups"""
+        result = instance_info.dbaas.backups.list()
+        assert_equal(1, len(result))
+        backup = result[0]
+        assert_equal(BACKUP_NAME, backup.name)
+        assert_equal(BACKUP_DESC, backup.description)
+        assert_equal(instance_info.id, backup.instance_id)
+        assert_equal('COMPLETED', backup.status)
+
+    @test
+    def test_backup_list_for_instance(self):
+        """test list backups"""
+        result = instance_info.dbaas.instances.backups(instance_info.id)
+        assert_equal(1, len(result))
+        backup = result[0]
+        assert_equal(BACKUP_NAME, backup.name)
+        assert_equal(BACKUP_DESC, backup.description)
+        assert_equal(instance_info.id, backup.instance_id)
+        assert_equal('COMPLETED', backup.status)
+
+    @test
+    def test_backup_get(self):
+        """test get backup"""
+        backup = instance_info.dbaas.backups.get(backup_info.id)
+        assert_equal(backup_info.id, backup.id)
+        assert_equal(backup_info.name, backup.name)
+        assert_equal(backup_info.description, backup.description)
+        assert_equal(instance_info.id, backup.instance_id)
+        assert_equal('COMPLETED', backup.status)
+
+
+@test(runs_after=[ListBackups],
+      groups=[GROUP])
+class DeleteBackups(object):
+
+    @test
+    def test_backup_delete_not_found(self):
+        """test delete unknown backup"""
+        assert_raises(exceptions.NotFound, instance_info.dbaas.backups.delete,
+                      'nonexistent_backup')
diff --git a/reddwarf/tests/fakes/guestagent.py b/reddwarf/tests/fakes/guestagent.py
index 1d2d6c7d..5d37f555 100644
--- a/reddwarf/tests/fakes/guestagent.py
+++ b/reddwarf/tests/fakes/guestagent.py
@@ -267,6 +267,15 @@ class FakeGuest(object):
                 } for db in current_grants]
         return dbs
 
+    def create_backup(self, backup_id):
+        from reddwarf.backup.models import Backup, BackupState
+        backup = Backup.get_by_id(backup_id)
+
+        def finish_create_backup():
+            backup.state = BackupState.COMPLETED
+            backup.save()
+        self.event_spawn(1.0, finish_create_backup)
+
 
 def get_or_create(id):
     if id not in DB:
diff --git a/reddwarf/tests/fakes/swift.py b/reddwarf/tests/fakes/swift.py
new file mode 100644
index 00000000..5bc9ca9f
--- /dev/null
+++ b/reddwarf/tests/fakes/swift.py
@@ -0,0 +1,398 @@
+import uuid
+import logging
+from mockito import when, any
+import swiftclient.client as swift_client
+import swiftclient
+
+
+# Copyright (C) 2012 Hewlett-Packard Development Company, L.P.
+# All Rights Reserved.
+#
+#    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 httplib
+import json
+import os
+import socket
+
+from swiftclient import client as swift
+
+LOG = logging.getLogger(__name__)
+
+
+class FakeSwiftClient(object):
+    """Logs calls instead of executing."""
+    def __init__(self, *args, **kwargs):
+        pass
+
+    @classmethod
+    def Connection(self, *args, **kargs):
+        LOG.debug("fake FakeSwiftClient Connection")
+        return FakeSwiftConnection()
+
+
+class FakeSwiftConnection(object):
+    """Logging calls instead of executing"""
+    def __init__(self, *args, **kwargs):
+        pass
+
+    def get_auth(self):
+        return (
+            u"http://127.0.0.1:8080/v1/AUTH_c7b038976df24d96bf1980f5da17bd89",
+            u'MIINrwYJKoZIhvcNAQcCoIINoDCCDZwCAQExCTAHBgUrDgMCGjCCDIgGCSqGSIb3'
+            u'DQEHAaCCDHkEggx1eyJhY2Nlc3MiOiB7InRva2VuIjogeyJpc3N1ZWRfYXQiOiAi'
+            u'MjAxMy0wMy0xOFQxODoxMzoyMC41OTMyNzYiLCAiZXhwaXJlcyI6ICIyMDEzLTAz'
+            u'LTE5VDE4OjEzOjIwWiIsICJpZCI6ICJwbGFjZWhvbGRlciIsICJ0ZW5hbnQiOiB7'
+            u'ImVuYWJsZWQiOiB0cnVlLCAiZGVzY3JpcHRpb24iOiBudWxsLCAibmFtZSI6ICJy'
+            u'ZWRkd2FyZiIsICJpZCI6ICJjN2IwMzg5NzZkZjI0ZDk2YmYxOTgwZjVkYTE3YmQ4'
+            u'OSJ9fSwgInNlcnZpY2VDYXRhbG9nIjogW3siZW5kcG9pbnRzIjogW3siYWRtaW5')
+
+    def get_account(self):
+        return ({'content-length': '2', 'accept-ranges': 'bytes',
+                 'x-timestamp': '1363049003.92304',
+                 'x-trans-id': 'tx9e5da02c49ed496395008309c8032a53',
+                 'date': 'Tue, 10 Mar 2013 00:43:23 GMT',
+                 'x-account-bytes-used': '0',
+                 'x-account-container-count': '0',
+                 'content-type': 'application/json; charset=utf-8',
+                 'x-account-object-count': '0'}, [])
+
+    def head_container(self, container):
+        LOG.debug("fake head_container(%s)" % container)
+        if container == 'missing_container':
+            raise swift.ClientException('fake exception',
+                                        http_status=httplib.NOT_FOUND)
+        elif container == 'unauthorized_container':
+            raise swift.ClientException('fake exception',
+                                        http_status=httplib.UNAUTHORIZED)
+        elif container == 'socket_error_on_head':
+            raise socket.error(111, 'ECONNREFUSED')
+        pass
+
+    def put_container(self, container):
+        LOG.debug("fake put_container(%s)" % container)
+        pass
+
+    def get_container(self, container, **kwargs):
+        LOG.debug("fake get_container(%s)" % container)
+        fake_header = None
+        fake_body = [{'name': 'backup_001'},
+                     {'name': 'backup_002'},
+                     {'name': 'backup_003'}]
+        return fake_header, fake_body
+
+    def head_object(self, container, name):
+        LOG.debug("fake put_container(%s, %s)" % (container, name))
+        return {'etag': 'fake-md5-sum'}
+
+    def get_object(self, container, name):
+        LOG.debug("fake get_object(%s, %s)" % (container, name))
+        if container == 'socket_error_on_get':
+            raise socket.error(111, 'ECONNREFUSED')
+        if 'metadata' in name:
+            fake_object_header = None
+            metadata = {}
+            if container == 'unsupported_version':
+                metadata['version'] = '9.9.9'
+            else:
+                metadata['version'] = '1.0.0'
+            metadata['backup_id'] = 123
+            metadata['volume_id'] = 123
+            metadata['backup_name'] = 'fake backup'
+            metadata['backup_description'] = 'fake backup description'
+            metadata['created_at'] = '2013-02-19 11:20:54,805'
+            metadata['objects'] = [{
+                'backup_001': {'compression': 'zlib', 'length': 10},
+                'backup_002': {'compression': 'zlib', 'length': 10},
+                'backup_003': {'compression': 'zlib', 'length': 10}
+            }]
+            metadata_json = json.dumps(metadata, sort_keys=True, indent=2)
+            fake_object_body = metadata_json
+            return (fake_object_header, fake_object_body)
+
+        fake_header = None
+        fake_object_body = os.urandom(1024 * 1024)
+        return (fake_header, fake_object_body)
+
+    def put_object(self, container, name, reader):
+        LOG.debug("fake put_object(%s, %s)" % (container, name))
+        if container == 'socket_error_on_put':
+            raise socket.error(111, 'ECONNREFUSED')
+        return 'fake-md5-sum'
+
+    def delete_object(self, container, name):
+        LOG.debug("fake delete_object(%s, %s)" % (container, name))
+        if container == 'socket_error_on_delete':
+            raise socket.error(111, 'ECONNREFUSED')
+        pass
+
+
+class SwiftClientStub(object):
+    """
+    Component for controlling behavior of Swift Client Stub.  Instantiated
+    before tests are invoked in "fake" mode.  Invoke methods to control
+    behavior so that systems under test can interact with this as it is a
+    real swift client with a real backend
+
+    example:
+
+    if FAKE:
+        swift_stub = SwiftClientStub()
+        swift_stub.with_account('xyz')
+
+    # returns swift account info and auth token
+    component_using_swift.get_swift_account()
+
+    if FAKE:
+        swift_stub.with_container('test-container-name')
+
+    # returns swift container information - mostly faked
+    component_using.swift.create_container('test-container-name')
+    component_using_swift.get_container_info('test-container-name')
+
+    if FAKE:
+        swift_stub.with_object('test-container-name', 'test-object-name',
+            'test-object-contents')
+
+    # returns swift object info and contents
+    component_using_swift.create_object('test-container-name',
+        'test-object-name', 'test-contents')
+    component_using_swift.get_object('test-container-name', 'test-object-name')
+
+    if FAKE:
+        swift_stub.without_object('test-container-name', 'test-object-name')
+
+    # allows object to be removed ONCE
+    component_using_swift.remove_object('test-container-name',
+        'test-object-name')
+    # throws ClientException - 404
+    component_using_swift.get_object('test-container-name', 'test-object-name')
+    component_using_swift.remove_object('test-container-name',
+        'test-object-name')
+
+    if FAKE:
+        swift_stub.without_object('test-container-name', 'test-object-name')
+
+    # allows container to be removed ONCE
+    component_using_swift.remove_container('test-container-name')
+    # throws ClientException - 404
+    component_using_swift.get_container('test-container-name')
+    component_using_swift.remove_container('test-container-name')
+    """
+
+    def __init__(self):
+        self._connection = swift_client.Connection()
+        # simulate getting an unknown container
+        when(swift_client.Connection).get_container(any()).thenRaise(
+            swiftclient.ClientException('Resource Not Found', http_status=404))
+
+        self._containers = {}
+        self._containers_list = []
+        self._objects = {}
+
+    def _remove_object(self, name, some_list):
+        idx = [i for i, obj in enumerate(some_list) if obj['name'] == name]
+        if len(idx) == 1:
+            del some_list[idx[0]]
+
+    def _ensure_object_exists(self, container, name):
+        self._connection.get_object(container, name)
+
+    def with_account(self, account_id):
+        """
+        setups up account headers
+
+        example:
+
+        if FAKE:
+            swift_stub = SwiftClientStub()
+            swift_stub.with_account('xyz')
+
+        # returns swift account info and auth token
+        component_using_swift.get_swift_account()
+
+        :param account_id: account id
+        """
+
+        def account_resp():
+            return ({'content-length': '2', 'accept-ranges': 'bytes',
+                     'x-timestamp': '1363049003.92304',
+                     'x-trans-id': 'tx9e5da02c49ed496395008309c8032a53',
+                     'date': 'Tue, 10 Mar 2013 00:43:23 GMT',
+                     'x-account-bytes-used': '0',
+                     'x-account-container-count': '0',
+                     'content-type': 'application/json; charset=utf-8',
+                     'x-account-object-count': '0'}, self._containers_list)
+
+        when(swift_client.Connection).get_auth().thenReturn((
+            u"http://127.0.0.1:8080/v1/AUTH_c7b038976df24d96bf1980f5da17bd89",
+            u'MIINrwYJKoZIhvcNAQcCoIINoDCCDZwCAQExCTAHBgUrDgMCGjCCDIgGCSqGSIb3'
+            u'DQEHAaCCDHkEggx1eyJhY2Nlc3MiOiB7InRva2VuIjogeyJpc3N1ZWRfYXQiOiAi'
+            u'MjAxMy0wMy0xOFQxODoxMzoyMC41OTMyNzYiLCAiZXhwaXJlcyI6ICIyMDEzLTAz'
+            u'LTE5VDE4OjEzOjIwWiIsICJpZCI6ICJwbGFjZWhvbGRlciIsICJ0ZW5hbnQiOiB7'
+            u'ImVuYWJsZWQiOiB0cnVlLCAiZGVzY3JpcHRpb24iOiBudWxsLCAibmFtZSI6ICJy'
+            u'ZWRkd2FyZiIsICJpZCI6ICJjN2IwMzg5NzZkZjI0ZDk2YmYxOTgwZjVkYTE3YmQ4'
+            u'OSJ9fSwgInNlcnZpY2VDYXRhbG9nIjogW3siZW5kcG9pbnRzIjogW3siYWRtaW5')
+        )
+        when(swift_client.Connection).get_account().thenReturn(account_resp())
+        return self
+
+    def _create_container(self, container_name):
+        container = {'count': 0, 'bytes': 0, 'name': container_name}
+        self._containers[container_name] = container
+        self._containers_list.append(container)
+        self._objects[container_name] = []
+
+    def _ensure_container_exists(self, container):
+        self._connection.get_container(container)
+
+    def _delete_container(self, container):
+        self._remove_object(container, self._containers_list)
+        del self._containers[container]
+        del self._objects[container]
+
+    def with_container(self, container_name):
+        """
+        sets expectations for creating a container and subsequently getting its
+        information
+
+        example:
+
+        if FAKE:
+            swift_stub.with_container('test-container-name')
+
+        # returns swift container information - mostly faked
+        component_using.swift.create_container('test-container-name')
+        component_using_swift.get_container_info('test-container-name')
+
+        :param container_name: container name that is expected to be created
+        """
+
+        def container_resp(container):
+            return ({'content-length': '2', 'x-container-object-count': '0',
+                     'accept-ranges': 'bytes', 'x-container-bytes-used': '0',
+                     'x-timestamp': '1363370869.72356',
+                     'x-trans-id': 'tx7731801ac6ec4e5f8f7da61cde46bed7',
+                     'date': 'Fri, 10 Mar 2013 18:07:58 GMT',
+                     'content-type': 'application/json; charset=utf-8'},
+                    self._objects[container])
+
+        # if this is called multiple times then nothing happens
+        when(swift_client.Connection).put_container(container_name).thenReturn(
+            None)
+        self._create_container(container_name)
+        # return container headers
+        when(swift_client.Connection).get_container(container_name).thenReturn(
+            container_resp(container_name))
+
+        return self
+
+    def without_container(self, container):
+        """
+        sets expectations for removing a container and subsequently throwing an
+        exception for further interactions
+
+        example:
+
+        if FAKE:
+            swift_stub.without_container('test-container-name')
+
+        # returns swift container information - mostly faked
+        component_using.swift.remove_container('test-container-name')
+        # throws exception "Resource Not Found - 404"
+        component_using_swift.get_container_info('test-container-name')
+
+        :param container: container name that is expected to be removed
+        """
+        # first ensure container
+        self._ensure_container_exists(container)
+        # allow one call to get container and then throw exceptions (may need
+        # to be revised
+        when(swift_client.Connection).delete_container(container).thenRaise(
+            swiftclient.ClientException("Resource Not Found", http_status=404))
+        when(swift_client.Connection).get_container(container).thenRaise(
+            swiftclient.ClientException("Resource Not Found", http_status=404))
+        self._delete_container(container)
+        return self
+
+    def with_object(self, container, name, contents):
+        """
+        sets expectations for creating an object and subsequently getting its
+        contents
+
+        example:
+
+        if FAKE:
+        swift_stub.with_object('test-container-name', 'test-object-name',
+            'test-object-contents')
+
+        # returns swift object info and contents
+        component_using_swift.create_object('test-container-name',
+            'test-object-name', 'test-contents')
+        component_using_swift.get_object('test-container-name',
+            'test-object-name')
+
+        :param container: container name that is the object belongs
+        :param name: the name of the object expected to be created
+        :param contents: the contents of the object
+        """
+
+        self._connection.get_container(container)
+        when(swift_client.Connection).put_object(container, name,
+                                                 contents).thenReturn(
+                                                     uuid.uuid1())
+        when(swift_client.Connection).get_object(container, name).thenReturn(
+            ({'content-length': len(contents), 'accept-ranges': 'bytes',
+              'last-modified': 'Mon, 10 Mar 2013 01:06:34 GMT',
+              'etag': 'eb15a6874ce265e2c3eb1b4891567bab',
+              'x-timestamp': '1363568794.67584',
+              'x-trans-id': 'txef3aaf26c897420c8e77c9750ce6a501',
+              'date': 'Mon, 10 Mar 2013 05:35:14 GMT',
+              'content-type': 'application/octet-stream'}, contents)
+        )
+        self._remove_object(name, self._objects[container])
+        self._objects[container].append(
+            {'bytes': 13, 'last_modified': '2013-03-15T22:10:49.361950',
+             'hash': 'ccc55aefbf92aa66f42b638802c5e7f6', 'name': name,
+             'content_type': 'application/octet-stream'})
+        return self
+
+    def without_object(self, container, name):
+        """
+        sets expectations for deleting an object
+
+        example:
+
+        if FAKE:
+        swift_stub.without_object('test-container-name', 'test-object-name')
+
+        # allows container to be removed ONCE
+        component_using_swift.remove_container('test-container-name')
+        # throws ClientException - 404
+        component_using_swift.get_container('test-container-name')
+        component_using_swift.remove_container('test-container-name')
+
+        :param container: container name that is the object belongs
+        :param name: the name of the object expected to be removed
+        """
+        self._ensure_container_exists(container)
+        self._ensure_object_exists(container, name)
+        # throw exception if someone calls get object
+        when(swift_client.Connection).get_object(container, name).thenRaise(
+            swiftclient.ClientException('Resource Not found', http_status=404))
+        when(swift_client.Connection).delete_object(
+            container, name).thenReturn(None).thenRaise(
+                swiftclient.ClientException('Resource Not Found',
+                                            http_status=404))
+        self._remove_object(name, self._objects[container])
+        return self
diff --git a/reddwarf/tests/unittests/backup/test_backup_models.py b/reddwarf/tests/unittests/backup/test_backup_models.py
index 40eff479..2eb7d6da 100644
--- a/reddwarf/tests/unittests/backup/test_backup_models.py
+++ b/reddwarf/tests/unittests/backup/test_backup_models.py
@@ -15,8 +15,11 @@
 import testtools
 from reddwarf.backup import models
 from reddwarf.tests.unittests.util import util
-from reddwarf.common import utils
+from reddwarf.common import utils, exception
 from reddwarf.common.context import ReddwarfContext
+from reddwarf.instance.models import BuiltInstance, InstanceTasks, Instance
+from mockito import mock, when, unstub, any
+from reddwarf.taskmanager import api
 
 
 def _prep_conf(current_time):
@@ -28,6 +31,7 @@ def _prep_conf(current_time):
 BACKUP_NAME = 'WORKS'
 BACKUP_NAME_2 = 'IT-WORKS'
 BACKUP_STATE = "NEW"
+BACKUP_DESC = 'Backup test'
 
 
 class BackupCreateTest(testtools.TestCase):
@@ -39,17 +43,89 @@ class BackupCreateTest(testtools.TestCase):
 
     def tearDown(self):
         super(BackupCreateTest, self).tearDown()
+        unstub()
         if self.created:
             models.DBBackup.find_by(
                 tenant_id=self.context.tenant).delete()
 
     def test_create(self):
-        models.Backup.create(
-            self.context, self.instance_id, BACKUP_NAME)
+        instance = mock(Instance)
+        when(BuiltInstance).load(any(), any()).thenReturn(instance)
+        when(instance).validate_can_perform_action().thenReturn(None)
+        when(models.Backup).verify_swift_auth_token(any()).thenReturn(
+            None)
+        when(api.API).create_backup(any()).thenReturn(None)
+
+        bu = models.Backup.create(self.context, self.instance_id,
+                                  BACKUP_NAME, BACKUP_DESC)
         self.created = True
-        db_record = models.DBBackup.find_by(
-            tenant_id=self.context.tenant)
+
+        self.assertEqual(BACKUP_NAME, bu.name)
+        self.assertEqual(BACKUP_DESC, bu.description)
+        self.assertEqual(self.instance_id, bu.instance_id)
+        self.assertEqual(models.BackupState.NEW, bu.state)
+
+        db_record = models.DBBackup.find_by(id=bu.id)
+        self.assertEqual(bu.id, db_record['id'])
+        self.assertEqual(BACKUP_NAME, db_record['name'])
+        self.assertEqual(BACKUP_DESC, db_record['description'])
         self.assertEqual(self.instance_id, db_record['instance_id'])
+        self.assertEqual(models.BackupState.NEW, db_record['state'])
+
+    def test_create_instance_not_found(self):
+        self.assertRaises(exception.NotFound, models.Backup.create,
+                          self.context, self.instance_id,
+                          BACKUP_NAME, BACKUP_DESC)
+
+    def test_create_instance_not_active(self):
+        instance = mock(Instance)
+        when(BuiltInstance).load(any(), any()).thenReturn(instance)
+        when(instance).validate_can_perform_action().thenRaise(
+            exception.UnprocessableEntity)
+        self.assertRaises(exception.UnprocessableEntity, models.Backup.create,
+                          self.context, self.instance_id,
+                          BACKUP_NAME, BACKUP_DESC)
+
+    def test_create_backup_swift_token_invalid(self):
+        instance = mock(Instance)
+        when(BuiltInstance).load(any(), any()).thenReturn(instance)
+        when(instance).validate_can_perform_action().thenReturn(None)
+        when(models.Backup).verify_swift_auth_token(any()).thenRaise(
+            exception.SwiftAuthError)
+        self.assertRaises(exception.SwiftAuthError, models.Backup.create,
+                          self.context, self.instance_id,
+                          BACKUP_NAME, BACKUP_DESC)
+
+
+class BackupDeleteTest(testtools.TestCase):
+    def setUp(self):
+        super(BackupDeleteTest, self).setUp()
+        util.init_db()
+        self.context, self.instance_id = _prep_conf(utils.utcnow())
+
+    def tearDown(self):
+        super(BackupDeleteTest, self).tearDown()
+        unstub()
+
+    def test_delete_backup_not_found(self):
+        self.assertRaises(exception.NotFound, models.Backup.delete,
+                          self.context, 'backup-id')
+
+    def test_delete_backup_is_running(self):
+        backup = mock()
+        backup.is_running = True
+        when(models.Backup).get_by_id(any()).thenReturn(backup)
+        self.assertRaises(exception.UnprocessableEntity,
+                          models.Backup.delete, self.context, 'backup_id')
+
+    def test_delete_backup_swift_token_invalid(self):
+        backup = mock()
+        backup.is_running = False
+        when(models.Backup).get_by_id(any()).thenReturn(backup)
+        when(models.Backup).verify_swift_auth_token(any()).thenRaise(
+            exception.SwiftAuthError)
+        self.assertRaises(exception.SwiftAuthError, models.Backup.delete,
+                          self.context, 'backup_id')
 
 
 class BackupORMTest(testtools.TestCase):
@@ -66,6 +142,7 @@ class BackupORMTest(testtools.TestCase):
 
     def tearDown(self):
         super(BackupORMTest, self).tearDown()
+        unstub()
         if not self.deleted:
             models.DBBackup.find_by(tenant_id=self.context.tenant).delete()
 
@@ -112,7 +189,8 @@ class BackupORMTest(testtools.TestCase):
         self.assertFalse(self.backup.is_done)
 
     def test_backup_delete(self):
-        models.Backup.delete(self.backup.id)
+        backup = models.DBBackup.find_by(id=self.backup.id)
+        backup.delete()
         query = models.Backup.list_for_instance(self.instance_id)
         self.assertEqual(query.count(), 0)
 
diff --git a/reddwarf/tests/unittests/common/__init__.py b/reddwarf/tests/unittests/common/__init__.py
new file mode 100644
index 00000000..e69de29b
--- /dev/null
+++ b/reddwarf/tests/unittests/common/__init__.py
diff --git a/reddwarf/tests/unittests/common/test_remote.py b/reddwarf/tests/unittests/common/test_remote.py
new file mode 100644
index 00000000..ddb7bc11
--- /dev/null
+++ b/reddwarf/tests/unittests/common/test_remote.py
@@ -0,0 +1,214 @@
+from mockito import mock, when, unstub
+import testtools
+from testtools.matchers import *
+
+import swiftclient.client
+
+from reddwarf.tests.fakes.swift import SwiftClientStub
+from reddwarf.common.context import ReddwarfContext
+from reddwarf.common import remote
+
+
+class TestRemote(testtools.TestCase):
+    def setUp(self):
+        super(TestRemote, self).setUp()
+
+    def tearDown(self):
+        super(TestRemote, self).tearDown()
+        unstub()
+
+    def test_creation(self):
+        when(swiftclient.client.Connection).get_auth().thenReturn(None)
+        conn = swiftclient.client.Connection()
+        self.assertIsNone(conn.get_auth())
+
+    def test_create_swift_client(self):
+        mock_resp = mock(dict)
+        when(swiftclient.client.Connection).get_container('bob').thenReturn(
+            ["text", mock_resp])
+        client = remote.create_swift_client(ReddwarfContext(tenant='123'))
+        headers, container = client.get_container('bob')
+        self.assertIs(headers, "text")
+        self.assertIs(container, mock_resp)
+
+    def test_empty_account(self):
+        """
+        this is an account with no containers and no objects
+        """
+        # setup expectation
+        swift_stub = SwiftClientStub()
+        swift_stub.with_account('123223')
+        # interact
+        conn = swiftclient.client.Connection()
+        account_info = conn.get_account()
+        self.assertThat(account_info, Not(Is(None)))
+        self.assertThat(len(account_info), Is(2))
+        self.assertThat(account_info, IsInstance(tuple))
+        self.assertThat(account_info[0], IsInstance(dict))
+        self.assertThat(account_info[0],
+                        KeysEqual('content-length', 'accept-ranges',
+                                  'x-timestamp', 'x-trans-id', 'date',
+                                  'x-account-bytes-used',
+                                  'x-account-container-count', 'content-type',
+                                  'x-account-object-count'))
+        self.assertThat(account_info[1], IsInstance(list))
+        self.assertThat(len(account_info[1]), Is(0))
+
+    def test_one_container(self):
+        """
+        tests to ensure behavior is normal with one container
+        """
+        # setup expectation
+        swift_stub = SwiftClientStub()
+        swift_stub.with_account('123223')
+        cont_name = 'a-container-name'
+        swift_stub.with_container(cont_name)
+        # interact
+        conn = swiftclient.client.Connection()
+        conn.get_auth()
+        conn.put_container(cont_name)
+        # get headers plus container metadata
+        self.assertThat(len(conn.get_account()), Is(2))
+        # verify container details
+        account_containers = conn.get_account()[1]
+        self.assertThat(len(account_containers), Is(1))
+        self.assertThat(account_containers[0],
+                        KeysEqual('count', 'bytes', 'name'))
+        self.assertThat(account_containers[0]['name'], Is(cont_name))
+        # get container details
+        cont_info = conn.get_container(cont_name)
+        self.assertIsNotNone(cont_info)
+        self.assertThat(cont_info[0], KeysEqual('content-length',
+                                                "x-container-object-count",
+                                                'accept-ranges',
+                                                'x-container-bytes-used',
+                                                'x-timestamp', 'x-trans-id',
+                                                'date', 'content-type'))
+        self.assertThat(len(cont_info[1]), Equals(0))
+        # remove container
+        swift_stub.without_container(cont_name)
+        with testtools.ExpectedException(swiftclient.ClientException):
+            conn.get_container(cont_name)
+            # ensure there are no more containers in account
+        self.assertThat(len(conn.get_account()[1]), Is(0))
+
+    def test_one_object(self):
+        swift_stub = SwiftClientStub()
+        swift_stub.with_account('123223')
+        swift_stub.with_container('bob')
+        swift_stub.with_object('bob', 'test', 'test_contents')
+        # create connection
+        conn = swiftclient.client.Connection()
+        # test container lightly
+        cont_info = conn.get_container('bob')
+        self.assertIsNotNone(cont_info)
+        self.assertThat(cont_info[0],
+                        KeysEqual('content-length', 'x-container-object-count',
+                                  'accept-ranges', 'x-container-bytes-used',
+                                  'x-timestamp', 'x-trans-id', 'date',
+                                  'content-type'))
+        cont_objects = cont_info[1]
+        self.assertThat(len(cont_objects), Equals(1))
+        obj_1 = cont_objects[0]
+        self.assertThat(obj_1, Equals(
+            {'bytes': 13, 'last_modified': '2013-03-15T22:10:49.361950',
+             'hash': 'ccc55aefbf92aa66f42b638802c5e7f6', 'name': 'test',
+             'content_type': 'application/octet-stream'}))
+        # test object api - not much to do here
+        self.assertThat(conn.get_object('bob', 'test')[1], Is('test_contents'))
+
+        # test remove object
+        swift_stub.without_object('bob', 'test')
+        # interact
+        conn.delete_object('bob', 'test')
+        with testtools.ExpectedException(swiftclient.ClientException):
+            conn.delete_object('bob', 'test')
+        self.assertThat(len(conn.get_container('bob')[1]), Is(0))
+
+    def test_two_objects(self):
+        swift_stub = SwiftClientStub()
+        swift_stub.with_account('123223')
+        swift_stub.with_container('bob')
+        swift_stub.with_container('bob2')
+        swift_stub.with_object('bob', 'test', 'test_contents')
+        swift_stub.with_object('bob', 'test2', 'test_contents2')
+
+        conn = swiftclient.client.Connection()
+
+        self.assertIs(len(conn.get_account()), 2)
+        cont_info = conn.get_container('bob')
+        self.assertIsNotNone(cont_info)
+        self.assertThat(cont_info[0],
+                        KeysEqual('content-length', 'x-container-object-count',
+                                  'accept-ranges', 'x-container-bytes-used',
+                                  'x-timestamp', 'x-trans-id', 'date',
+                                  'content-type'))
+        self.assertThat(len(cont_info[1]), Equals(2))
+        self.assertThat(cont_info[1][0], Equals(
+            {'bytes': 13, 'last_modified': '2013-03-15T22:10:49.361950',
+             'hash': 'ccc55aefbf92aa66f42b638802c5e7f6', 'name': 'test',
+             'content_type': 'application/octet-stream'}))
+        self.assertThat(conn.get_object('bob', 'test')[1], Is('test_contents'))
+        self.assertThat(conn.get_object('bob', 'test2')[1],
+                        Is('test_contents2'))
+
+        swift_stub.without_object('bob', 'test')
+        conn.delete_object('bob', 'test')
+        with testtools.ExpectedException(swiftclient.ClientException):
+            conn.delete_object('bob', 'test')
+        self.assertThat(len(conn.get_container('bob')[1]), Is(1))
+
+        swift_stub.without_container('bob')
+        with testtools.ExpectedException(swiftclient.ClientException):
+            conn.get_container('bob')
+
+        self.assertThat(len(conn.get_account()), Is(2))
+
+    def test_nonexisting_container(self):
+        """
+        when a container does not exist and is accessed then a 404 is returned
+        """
+        from reddwarf.tests.fakes.swift import SwiftClientStub
+
+        swift_stub = SwiftClientStub()
+        swift_stub.with_account('123223')
+        swift_stub.with_container('existing')
+
+        conn = swiftclient.client.Connection()
+
+        with testtools.ExpectedException(swiftclient.ClientException):
+            conn.get_container('nonexisting')
+
+    def test_replace_object(self):
+        """
+        Test to ensure that if an object is updated the container object
+        count is the same and the contents of the object are updated
+        """
+        swift_stub = SwiftClientStub()
+        swift_stub.with_account('1223df2')
+        swift_stub.with_container('new-container')
+        swift_stub.with_object('new-container', 'new-object',
+                               'new-object-contents')
+
+        conn = swiftclient.client.Connection()
+
+        conn.put_object('new-container', 'new-object', 'new-object-contents')
+        obj_resp = conn.get_object('new-container', 'new-object')
+        self.assertThat(obj_resp, Not(Is(None)))
+        self.assertThat(len(obj_resp), Is(2))
+        self.assertThat(obj_resp[1], Is('new-object-contents'))
+
+        # set expected behavior - trivial here since it is the intended
+        # behavior however keep in mind this is just to support testing of
+        # reddwarf components
+        swift_stub.with_object('new-container', 'new-object',
+                               'updated-object-contents')
+
+        conn.put_object('new-container', 'new-object',
+                        'updated-object-contents')
+        obj_resp = conn.get_object('new-container', 'new-object')
+        self.assertThat(obj_resp, Not(Is(None)))
+        self.assertThat(len(obj_resp), Is(2))
+        self.assertThat(obj_resp[1], Is('updated-object-contents'))
+        # ensure object count has not increased
+        self.assertThat(len(conn.get_container('new-container')[1]), Is(1))
diff --git a/reddwarf/tests/unittests/quota/test_quota.py b/reddwarf/tests/unittests/quota/test_quota.py
index b2b39aa5..8449f4cf 100644
--- a/reddwarf/tests/unittests/quota/test_quota.py
+++ b/reddwarf/tests/unittests/quota/test_quota.py
@@ -23,7 +23,7 @@ from reddwarf.db.models import DatabaseModelBase
 from reddwarf.extensions.mgmt.quota.service import QuotaController
 from reddwarf.common import exception
 from reddwarf.common import cfg
-from reddwarf.instance.models import run_with_quotas
+from reddwarf.quota.quota import run_with_quotas
 from reddwarf.quota.quota import QUOTAS
 """
 Unit tests for the classes and functions in DbQuotaDriver.py.
diff --git a/run_tests.py b/run_tests.py
index fd971e0c..73d77c0e 100644
--- a/run_tests.py
+++ b/run_tests.py
@@ -123,6 +123,7 @@ if __name__ == "__main__":
         test_config_file = parse_args_for_test_config()
         CONFIG.load_from_file(test_config_file)
 
+        from reddwarf.tests.api import backups
         from reddwarf.tests.api import header
         from reddwarf.tests.api import limits
         from reddwarf.tests.api import flavors
diff --git a/tools/pip-requires b/tools/pip-requires
index 847b9e70..94a76cd2 100644
--- a/tools/pip-requires
+++ b/tools/pip-requires
@@ -13,5 +13,6 @@ httplib2
 lxml
 python-novaclient
 python-keystoneclient
+python-swiftclient
 iso8601
 oslo.config>=1.1.0