diff --git a/tempest/api/compute/servers/test_server_actions.py b/tempest/api/compute/servers/test_server_actions.py
index 961737a..4a1f8e3 100644
--- a/tempest/api/compute/servers/test_server_actions.py
+++ b/tempest/api/compute/servers/test_server_actions.py
@@ -245,6 +245,31 @@
         self.client.wait_for_server_status(self.server_id, 'ACTIVE')
 
     @attr(type='gate')
+    def test_shelve_unshelve_server(self):
+        resp, server = self.client.shelve_server(self.server_id)
+        self.assertEqual(202, resp.status)
+
+        offload_time = self.config.compute.shelved_offload_time
+        if offload_time >= 0:
+            self.client.wait_for_server_status(self.server_id,
+                                               'SHELVED_OFFLOADED',
+                                               extra_timeout=offload_time)
+        else:
+            self.client.wait_for_server_status(self.server_id,
+                                               'SHELVED')
+
+        resp, server = self.client.get_server(self.server_id)
+        image_name = server['name'] + '-shelved'
+        params = {'name': image_name}
+        resp, images = self.images_client.list_images(params)
+        self.assertEqual(1, len(images))
+        self.assertEqual(image_name, images[0]['name'])
+
+        resp, server = self.client.unshelve_server(self.server_id)
+        self.assertEqual(202, resp.status)
+        self.client.wait_for_server_status(self.server_id, 'ACTIVE')
+
+    @attr(type='gate')
     def test_stop_start_server(self):
         resp, server = self.servers_client.stop(self.server_id)
         self.assertEqual(202, resp.status)
diff --git a/tempest/api/compute/servers/test_servers_negative.py b/tempest/api/compute/servers/test_servers_negative.py
index 7eb127d..b12afd1 100644
--- a/tempest/api/compute/servers/test_servers_negative.py
+++ b/tempest/api/compute/servers/test_servers_negative.py
@@ -396,6 +396,54 @@
                           self.client.restore_soft_deleted_server,
                           self.server_id)
 
+    @attr(type=['negative', 'gate'])
+    def test_shelve_non_existent_server(self):
+        # shelve a non existent server
+        nonexistent_server = str(uuid.uuid4())
+        self.assertRaises(exceptions.NotFound, self.client.shelve_server,
+                          nonexistent_server)
+
+    @attr(type=['negative', 'gate'])
+    def test_shelve_shelved_server(self):
+        # shelve a shelved server.
+        resp, server = self.client.shelve_server(self.server_id)
+        self.assertEqual(202, resp.status)
+        self.addCleanup(self.client.unshelve_server, self.server_id)
+
+        offload_time = self.config.compute.shelved_offload_time
+        if offload_time >= 0:
+            self.client.wait_for_server_status(self.server_id,
+                                               'SHELVED_OFFLOADED',
+                                               extra_timeout=offload_time)
+        else:
+            self.client.wait_for_server_status(self.server_id,
+                                               'SHELVED')
+
+        resp, server = self.client.get_server(self.server_id)
+        image_name = server['name'] + '-shelved'
+        params = {'name': image_name}
+        resp, images = self.images_client.list_images(params)
+        self.assertEqual(1, len(images))
+        self.assertEqual(image_name, images[0]['name'])
+
+        self.assertRaises(exceptions.Conflict,
+                          self.client.shelve_server,
+                          self.server_id)
+
+    @attr(type=['negative', 'gate'])
+    def test_unshelve_non_existent_server(self):
+        # unshelve a non existent server
+        nonexistent_server = str(uuid.uuid4())
+        self.assertRaises(exceptions.NotFound, self.client.unshelve_server,
+                          nonexistent_server)
+
+    @attr(type=['negative', 'gate'])
+    def test_unshelve_server_invalid_state(self):
+        # unshelve an active server.
+        self.assertRaises(exceptions.Conflict,
+                          self.client.unshelve_server,
+                          self.server_id)
+
 
 class ServersNegativeTestXML(ServersNegativeTestJSON):
     _interface = 'xml'
diff --git a/tempest/common/waiters.py b/tempest/common/waiters.py
index 15569cd..bea2cdc 100644
--- a/tempest/common/waiters.py
+++ b/tempest/common/waiters.py
@@ -24,7 +24,8 @@
 
 
 # NOTE(afazekas): This function needs to know a token and a subject.
-def wait_for_server_status(client, server_id, status, ready_wait=True):
+def wait_for_server_status(client, server_id, status, ready_wait=True,
+                           extra_timeout=0):
     """Waits for a server to reach a given status."""
 
     def _get_task_state(body):
@@ -37,6 +38,7 @@
     old_status = server_status = body['status']
     old_task_state = task_state = _get_task_state(body)
     start_time = int(time.time())
+    timeout = client.build_timeout + extra_timeout
     while True:
         # NOTE(afazekas): Now the BUILD status only reached
         # between the UNKOWN->ACTIVE transition.
@@ -70,12 +72,12 @@
         if server_status == 'ERROR':
             raise exceptions.BuildErrorException(server_id=server_id)
 
-        timed_out = int(time.time()) - start_time >= client.build_timeout
+        timed_out = int(time.time()) - start_time >= timeout
 
         if timed_out:
             message = ('Server %s failed to reach %s status within the '
                        'required time (%s s).' %
-                       (server_id, status, client.build_timeout))
+                       (server_id, status, timeout))
             message += ' Current status: %s.' % server_status
             raise exceptions.TimeoutException(message)
         old_status = server_status
diff --git a/tempest/config.py b/tempest/config.py
index 76461fb..ff8a170 100644
--- a/tempest/config.py
+++ b/tempest/config.py
@@ -196,7 +196,14 @@
     cfg.StrOpt('volume_device_name',
                default='vdb',
                help="Expected device name when a volume is attached to "
-                    "an instance")
+                    "an instance"),
+    cfg.IntOpt('shelved_offload_time',
+               default=0,
+               help='Time in seconds before a shelved instance is eligible '
+                    'for removing from a host.  -1 never offload, 0 offload '
+                    'when shelved. This time should be the same as the time '
+                    'of nova.conf, and some tests will run for as long as the '
+                    'time.')
 ]
 
 compute_features_group = cfg.OptGroup(name='compute-feature-enabled',
diff --git a/tempest/services/compute/json/servers_client.py b/tempest/services/compute/json/servers_client.py
index 55a4a1b..f2386b0 100644
--- a/tempest/services/compute/json/servers_client.py
+++ b/tempest/services/compute/json/servers_client.py
@@ -154,9 +154,10 @@
         body = json.loads(body)
         return resp, body
 
-    def wait_for_server_status(self, server_id, status):
+    def wait_for_server_status(self, server_id, status, extra_timeout=0):
         """Waits for a server to reach a given status."""
-        return waiters.wait_for_server_status(self, server_id, status)
+        return waiters.wait_for_server_status(self, server_id, status,
+                                              extra_timeout=extra_timeout)
 
     def wait_for_server_termination(self, server_id, ignore_error=False):
         """Waits for server to reach termination."""
@@ -333,25 +334,33 @@
         return self.action(server_id, 'unlock', None, **kwargs)
 
     def suspend_server(self, server_id, **kwargs):
-        """Suspends the provded server."""
+        """Suspends the provided server."""
         return self.action(server_id, 'suspend', None, **kwargs)
 
     def resume_server(self, server_id, **kwargs):
-        """Un-suspends the provded server."""
+        """Un-suspends the provided server."""
         return self.action(server_id, 'resume', None, **kwargs)
 
     def pause_server(self, server_id, **kwargs):
-        """Pauses the provded server."""
+        """Pauses the provided server."""
         return self.action(server_id, 'pause', None, **kwargs)
 
     def unpause_server(self, server_id, **kwargs):
-        """Un-pauses the provded server."""
+        """Un-pauses the provided server."""
         return self.action(server_id, 'unpause', None, **kwargs)
 
     def reset_state(self, server_id, state='error'):
         """Resets the state of a server to active/error."""
         return self.action(server_id, 'os-resetState', None, state=state)
 
+    def shelve_server(self, server_id, **kwargs):
+        """Shelves the provided server."""
+        return self.action(server_id, 'shelve', None, **kwargs)
+
+    def unshelve_server(self, server_id, **kwargs):
+        """Un-shelves the provided server."""
+        return self.action(server_id, 'unshelve', None, **kwargs)
+
     def get_console_output(self, server_id, length):
         return self.action(server_id, 'os-getConsoleOutput', 'output',
                            length=length)
diff --git a/tempest/services/compute/v3/json/servers_client.py b/tempest/services/compute/v3/json/servers_client.py
index a005edb..f010580 100644
--- a/tempest/services/compute/v3/json/servers_client.py
+++ b/tempest/services/compute/v3/json/servers_client.py
@@ -161,9 +161,10 @@
         body = json.loads(body)
         return resp, body
 
-    def wait_for_server_status(self, server_id, status):
+    def wait_for_server_status(self, server_id, status, extra_timeout=0):
         """Waits for a server to reach a given status."""
-        return waiters.wait_for_server_status(self, server_id, status)
+        return waiters.wait_for_server_status(self, server_id, status,
+                                              extra_timeout=extra_timeout)
 
     def wait_for_server_termination(self, server_id, ignore_error=False):
         """Waits for server to reach termination."""
diff --git a/tempest/services/compute/v3/xml/servers_client.py b/tempest/services/compute/v3/xml/servers_client.py
index 6f38b6a..af3a152 100644
--- a/tempest/services/compute/v3/xml/servers_client.py
+++ b/tempest/services/compute/v3/xml/servers_client.py
@@ -376,9 +376,10 @@
         server = self._parse_server(etree.fromstring(body))
         return resp, server
 
-    def wait_for_server_status(self, server_id, status):
+    def wait_for_server_status(self, server_id, status, extra_timeout=0):
         """Waits for a server to reach a given status."""
-        return waiters.wait_for_server_status(self, server_id, status)
+        return waiters.wait_for_server_status(self, server_id, status,
+                                              extra_timeout=extra_timeout)
 
     def wait_for_server_termination(self, server_id, ignore_error=False):
         """Waits for server to reach termination."""
diff --git a/tempest/services/compute/xml/servers_client.py b/tempest/services/compute/xml/servers_client.py
index e21bfc4..d97a659 100644
--- a/tempest/services/compute/xml/servers_client.py
+++ b/tempest/services/compute/xml/servers_client.py
@@ -217,6 +217,14 @@
         """Un-pauses the provided server."""
         return self.action(server_id, 'unpause', None, **kwargs)
 
+    def shelve_server(self, server_id, **kwargs):
+        """Shelves the provided server."""
+        return self.action(server_id, 'shelve', None, **kwargs)
+
+    def unshelve_server(self, server_id, **kwargs):
+        """Un-shelves the provided server."""
+        return self.action(server_id, 'unshelve', None, **kwargs)
+
     def reset_state(self, server_id, state='error'):
         """Resets the state of a server to active/error."""
         return self.action(server_id, 'os-resetState', None, state=state)
@@ -351,9 +359,10 @@
         server = self._parse_server(etree.fromstring(body))
         return resp, server
 
-    def wait_for_server_status(self, server_id, status):
+    def wait_for_server_status(self, server_id, status, extra_timeout=0):
         """Waits for a server to reach a given status."""
-        return waiters.wait_for_server_status(self, server_id, status)
+        return waiters.wait_for_server_status(self, server_id, status,
+                                              extra_timeout=extra_timeout)
 
     def wait_for_server_termination(self, server_id, ignore_error=False):
         """Waits for server to reach termination."""