Deploy templates: API & notifications

Adds deploy_templates REST API endpoints for retrieving, creating,
updating and deleting deployment templates. Also adds notification
objects for deploy templates.

Bumps the minimum WSME requirement to 0.9.3, since the lower constraints
job was failing with a 500 error when sending data in an unexpected
format to the POST /deploy_templates API.

Change-Id: I0e8c97e600f9b1080c8bdec790e5710e7a92d016
Story: 1722275
Task: 28677

4 files changed, 993 insertions, 0 deletions

diff --git a/ironic/tests/unit/api/controllers/v1/test_deploy_template.py b/ironic/tests/unit/api/controllers/v1/test_deploy_template.py
new file mode 100644
index 000000000..2c2bbebb1
--- /dev/null
+++ b/ironic/tests/unit/api/controllers/v1/test_deploy_template.py
@@ -0,0 +1,942 @@
+#    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.
+"""
+Tests for the API /deploy_templates/ methods.
+"""
+
+import datetime
+
+import mock
+from oslo_config import cfg
+from oslo_utils import timeutils
+from oslo_utils import uuidutils
+import six
+from six.moves import http_client
+from six.moves.urllib import parse as urlparse
+
+from ironic.api.controllers import base as api_base
+from ironic.api.controllers import v1 as api_v1
+from ironic.api.controllers.v1 import deploy_template as api_deploy_template
+from ironic.api.controllers.v1 import notification_utils
+from ironic.common import exception
+from ironic import objects
+from ironic.objects import fields as obj_fields
+from ironic.tests import base
+from ironic.tests.unit.api import base as test_api_base
+from ironic.tests.unit.api import utils as test_api_utils
+from ironic.tests.unit.objects import utils as obj_utils
+
+
+def _obj_to_api_step(obj_step):
+    """Convert a deploy step in 'object' form to one in 'API' form."""
+    return {
+        'interface': obj_step['interface'],
+        'step': obj_step['step'],
+        'args': obj_step['args'],
+        'priority': obj_step['priority'],
+    }
+
+
+class TestDeployTemplateObject(base.TestCase):
+
+    def test_deploy_template_init(self):
+        template_dict = test_api_utils.deploy_template_post_data()
+        template = api_deploy_template.DeployTemplate(**template_dict)
+        self.assertEqual(template_dict['uuid'], template.uuid)
+        self.assertEqual(template_dict['name'], template.name)
+        self.assertEqual(template_dict['extra'], template.extra)
+        for t_dict_step, t_step in zip(template_dict['steps'], template.steps):
+            self.assertEqual(t_dict_step['interface'], t_step.interface)
+            self.assertEqual(t_dict_step['step'], t_step.step)
+            self.assertEqual(t_dict_step['args'], t_step.args)
+            self.assertEqual(t_dict_step['priority'], t_step.priority)
+
+    def test_deploy_template_sample(self):
+        sample = api_deploy_template.DeployTemplate.sample(expand=False)
+        self.assertEqual('534e73fa-1014-4e58-969a-814cc0cb9d43', sample.uuid)
+        self.assertEqual('CUSTOM_RAID1', sample.name)
+        self.assertEqual({'foo': 'bar'}, sample.extra)
+
+
+class BaseDeployTemplatesAPITest(test_api_base.BaseApiTest):
+    headers = {api_base.Version.string: str(api_v1.max_version())}
+    invalid_version_headers = {api_base.Version.string: '1.54'}
+
+
+class TestListDeployTemplates(BaseDeployTemplatesAPITest):
+
+    def test_empty(self):
+        data = self.get_json('/deploy_templates', headers=self.headers)
+        self.assertEqual([], data['deploy_templates'])
+
+    def test_one(self):
+        template = obj_utils.create_test_deploy_template(self.context)
+        data = self.get_json('/deploy_templates', headers=self.headers)
+        self.assertEqual(1, len(data['deploy_templates']))
+        self.assertEqual(template.uuid, data['deploy_templates'][0]['uuid'])
+        self.assertEqual(template.name, data['deploy_templates'][0]['name'])
+        self.assertNotIn('steps', data['deploy_templates'][0])
+        self.assertNotIn('extra', data['deploy_templates'][0])
+
+    def test_get_one(self):
+        template = obj_utils.create_test_deploy_template(self.context)
+        data = self.get_json('/deploy_templates/%s' % template.uuid,
+                             headers=self.headers)
+        self.assertEqual(template.uuid, data['uuid'])
+        self.assertEqual(template.name, data['name'])
+        self.assertEqual(template.extra, data['extra'])
+        for t_dict_step, t_step in zip(data['steps'], template.steps):
+            self.assertEqual(t_dict_step['interface'], t_step['interface'])
+            self.assertEqual(t_dict_step['step'], t_step['step'])
+            self.assertEqual(t_dict_step['args'], t_step['args'])
+            self.assertEqual(t_dict_step['priority'], t_step['priority'])
+
+    def test_get_one_with_json(self):
+        template = obj_utils.create_test_deploy_template(self.context)
+        data = self.get_json('/deploy_templates/%s.json' % template.uuid,
+                             headers=self.headers)
+        self.assertEqual(template.uuid, data['uuid'])
+
+    def test_get_one_with_suffix(self):
+        template = obj_utils.create_test_deploy_template(self.context,
+                                                         name='CUSTOM_DT1')
+        data = self.get_json('/deploy_templates/%s' % template.uuid,
+                             headers=self.headers)
+        self.assertEqual(template.uuid, data['uuid'])
+
+    def test_get_one_custom_fields(self):
+        template = obj_utils.create_test_deploy_template(self.context)
+        fields = 'name,steps'
+        data = self.get_json(
+            '/deploy_templates/%s?fields=%s' % (template.uuid, fields),
+            headers=self.headers)
+        # We always append "links"
+        self.assertItemsEqual(['name', 'steps', 'links'], data)
+
+    def test_get_collection_custom_fields(self):
+        fields = 'uuid,steps'
+        for i in range(3):
+            obj_utils.create_test_deploy_template(
+                self.context,
+                uuid=uuidutils.generate_uuid(),
+                name='CUSTOM_DT%s' % i)
+
+        data = self.get_json(
+            '/deploy_templates?fields=%s' % fields,
+            headers=self.headers)
+
+        self.assertEqual(3, len(data['deploy_templates']))
+        for template in data['deploy_templates']:
+            # We always append "links"
+            self.assertItemsEqual(['uuid', 'steps', 'links'], template)
+
+    def test_get_custom_fields_invalid_fields(self):
+        template = obj_utils.create_test_deploy_template(self.context)
+        fields = 'uuid,spongebob'
+        response = self.get_json(
+            '/deploy_templates/%s?fields=%s' % (template.uuid, fields),
+            headers=self.headers, expect_errors=True)
+        self.assertEqual(http_client.BAD_REQUEST, response.status_int)
+        self.assertEqual('application/json', response.content_type)
+        self.assertIn('spongebob', response.json['error_message'])
+
+    def test_get_all_invalid_api_version(self):
+        obj_utils.create_test_deploy_template(self.context)
+        response = self.get_json('/deploy_templates',
+                                 headers=self.invalid_version_headers,
+                                 expect_errors=True)
+        self.assertEqual(http_client.NOT_FOUND, response.status_int)
+
+    def test_get_one_invalid_api_version(self):
+        template = obj_utils.create_test_deploy_template(self.context)
+        response = self.get_json(
+            '/deploy_templates/%s' % (template.uuid),
+            headers=self.invalid_version_headers,
+            expect_errors=True)
+        self.assertEqual(http_client.NOT_FOUND, response.status_int)
+
+    def test_detail_query(self):
+        template = obj_utils.create_test_deploy_template(self.context)
+        data = self.get_json('/deploy_templates?detail=True',
+                             headers=self.headers)
+        self.assertEqual(template.uuid, data['deploy_templates'][0]['uuid'])
+        self.assertIn('name', data['deploy_templates'][0])
+        self.assertIn('steps', data['deploy_templates'][0])
+        self.assertIn('extra', data['deploy_templates'][0])
+
+    def test_detail_query_false(self):
+        obj_utils.create_test_deploy_template(self.context)
+        data1 = self.get_json(
+            '/deploy_templates',
+            headers={api_base.Version.string: str(api_v1.max_version())})
+        data2 = self.get_json(
+            '/deploy_templates?detail=False',
+            headers={api_base.Version.string: str(api_v1.max_version())})
+        self.assertEqual(data1['deploy_templates'], data2['deploy_templates'])
+
+    def test_detail_using_query_false_and_fields(self):
+        obj_utils.create_test_deploy_template(self.context)
+        data = self.get_json(
+            '/deploy_templates?detail=False&fields=steps',
+            headers={api_base.Version.string: str(api_v1.max_version())})
+        self.assertIn('steps', data['deploy_templates'][0])
+        self.assertNotIn('uuid', data['deploy_templates'][0])
+        self.assertNotIn('extra', data['deploy_templates'][0])
+
+    def test_detail_using_query_and_fields(self):
+        obj_utils.create_test_deploy_template(self.context)
+        response = self.get_json(
+            '/deploy_templates?detail=True&fields=name',
+            headers={api_base.Version.string: str(api_v1.max_version())},
+            expect_errors=True)
+        self.assertEqual(http_client.BAD_REQUEST, response.status_int)
+
+    def test_many(self):
+        templates = []
+        for id_ in range(5):
+            template = obj_utils.create_test_deploy_template(
+                self.context, uuid=uuidutils.generate_uuid(),
+                name='CUSTOM_DT%s' % id_)
+            templates.append(template.uuid)
+        data = self.get_json('/deploy_templates', headers=self.headers)
+        self.assertEqual(len(templates), len(data['deploy_templates']))
+
+        uuids = [n['uuid'] for n in data['deploy_templates']]
+        six.assertCountEqual(self, templates, uuids)
+
+    def test_links(self):
+        uuid = uuidutils.generate_uuid()
+        obj_utils.create_test_deploy_template(self.context, uuid=uuid)
+        data = self.get_json('/deploy_templates/%s' % uuid,
+                             headers=self.headers)
+        self.assertIn('links', data)
+        self.assertEqual(2, len(data['links']))
+        self.assertIn(uuid, data['links'][0]['href'])
+        for l in data['links']:
+            bookmark = l['rel'] == 'bookmark'
+            self.assertTrue(self.validate_link(l['href'], bookmark=bookmark,
+                            headers=self.headers))
+
+    def test_collection_links(self):
+        templates = []
+        for id_ in range(5):
+            template = obj_utils.create_test_deploy_template(
+                self.context, uuid=uuidutils.generate_uuid(),
+                name='CUSTOM_DT%s' % id_)
+            templates.append(template.uuid)
+        data = self.get_json('/deploy_templates/?limit=3',
+                             headers=self.headers)
+        self.assertEqual(3, len(data['deploy_templates']))
+
+        next_marker = data['deploy_templates'][-1]['uuid']
+        self.assertIn(next_marker, data['next'])
+
+    def test_collection_links_default_limit(self):
+        cfg.CONF.set_override('max_limit', 3, 'api')
+        templates = []
+        for id_ in range(5):
+            template = obj_utils.create_test_deploy_template(
+                self.context, uuid=uuidutils.generate_uuid(),
+                name='CUSTOM_DT%s' % id_)
+            templates.append(template.uuid)
+        data = self.get_json('/deploy_templates', headers=self.headers)
+        self.assertEqual(3, len(data['deploy_templates']))
+
+        next_marker = data['deploy_templates'][-1]['uuid']
+        self.assertIn(next_marker, data['next'])
+
+    def test_get_collection_pagination_no_uuid(self):
+        fields = 'name'
+        limit = 2
+        templates = []
+        for id_ in range(3):
+            template = obj_utils.create_test_deploy_template(
+                self.context,
+                uuid=uuidutils.generate_uuid(),
+                name='CUSTOM_DT%s' % id_)
+            templates.append(template)
+
+        data = self.get_json(
+            '/deploy_templates?fields=%s&limit=%s' % (fields, limit),
+            headers=self.headers)
+
+        self.assertEqual(limit, len(data['deploy_templates']))
+        self.assertIn('marker=%s' % templates[limit - 1].uuid, data['next'])
+
+    def test_sort_key(self):
+        templates = []
+        for id_ in range(3):
+            template = obj_utils.create_test_deploy_template(
+                self.context,
+                uuid=uuidutils.generate_uuid(),
+                name='CUSTOM_DT%s' % id_)
+            templates.append(template.uuid)
+        data = self.get_json('/deploy_templates?sort_key=uuid',
+                             headers=self.headers)
+        uuids = [n['uuid'] for n in data['deploy_templates']]
+        self.assertEqual(sorted(templates), uuids)
+
+    def test_sort_key_invalid(self):
+        invalid_keys_list = ['extra', 'foo', 'steps']
+        for invalid_key in invalid_keys_list:
+            path = '/deploy_templates?sort_key=%s' % invalid_key
+            response = self.get_json(path, expect_errors=True,
+                                     headers=self.headers)
+            self.assertEqual(http_client.BAD_REQUEST, response.status_int)
+            self.assertEqual('application/json', response.content_type)
+            self.assertIn(invalid_key, response.json['error_message'])
+
+    def _test_sort_key_allowed(self, detail=False):
+        template_uuids = []
+        for id_ in range(3, 0, -1):
+            template = obj_utils.create_test_deploy_template(
+                self.context,
+                uuid=uuidutils.generate_uuid(),
+                name='CUSTOM_DT%s' % id_)
+            template_uuids.append(template.uuid)
+        template_uuids.reverse()
+        url = '/deploy_templates?sort_key=name&detail=%s' % str(detail)
+        data = self.get_json(url, headers=self.headers)
+        data_uuids = [p['uuid'] for p in data['deploy_templates']]
+        self.assertEqual(template_uuids, data_uuids)
+
+    def test_sort_key_allowed(self):
+        self._test_sort_key_allowed()
+
+    def test_detail_sort_key_allowed(self):
+        self._test_sort_key_allowed(detail=True)
+
+    def test_sensitive_data_masked(self):
+        template = obj_utils.get_test_deploy_template(self.context)
+        template.steps[0]['args']['password'] = 'correcthorsebatterystaple'
+        template.create()
+        data = self.get_json('/deploy_templates/%s' % template.uuid,
+                             headers=self.headers)
+
+        self.assertEqual("******", data['steps'][0]['args']['password'])
+
+
+@mock.patch.object(objects.DeployTemplate, 'save', autospec=True)
+class TestPatch(BaseDeployTemplatesAPITest):
+
+    def setUp(self):
+        super(TestPatch, self).setUp()
+        self.template = obj_utils.create_test_deploy_template(
+            self.context, name='CUSTOM_DT1')
+
+    def _test_update_ok(self, mock_save, patch):
+        response = self.patch_json('/deploy_templates/%s' % self.template.uuid,
+                                   patch, headers=self.headers)
+        self.assertEqual('application/json', response.content_type)
+        self.assertEqual(http_client.OK, response.status_code)
+        mock_save.assert_called_once_with(mock.ANY)
+        return response
+
+    def _test_update_bad_request(self, mock_save, patch, error_msg):
+        response = self.patch_json('/deploy_templates/%s' % self.template.uuid,
+                                   patch, expect_errors=True,
+                                   headers=self.headers)
+        self.assertEqual('application/json', response.content_type)
+        self.assertEqual(http_client.BAD_REQUEST, response.status_code)
+        self.assertTrue(response.json['error_message'])
+        self.assertIn(error_msg, response.json['error_message'])
+        self.assertFalse(mock_save.called)
+        return response
+
+    @mock.patch.object(notification_utils, '_emit_api_notification',
+                       autospec=True)
+    def test_update_by_id(self, mock_notify, mock_save):
+        name = 'CUSTOM_DT2'
+        patch = [{'path': '/name', 'value': name, 'op': 'add'}]
+        response = self._test_update_ok(mock_save, patch)
+        self.assertEqual(name, response.json['name'])
+
+        mock_notify.assert_has_calls([mock.call(mock.ANY, mock.ANY, 'update',
+                                      obj_fields.NotificationLevel.INFO,
+                                      obj_fields.NotificationStatus.START),
+                                      mock.call(mock.ANY, mock.ANY, 'update',
+                                      obj_fields.NotificationLevel.INFO,
+                                      obj_fields.NotificationStatus.END)])
+
+    def test_update_by_name(self, mock_save):
+        steps = [{
+            'interface': 'bios',
+            'step': 'apply_configuration',
+            'args': {'foo': 'bar'},
+            'priority': 42
+        }]
+        patch = [{'path': '/steps', 'value': steps, 'op': 'replace'}]
+        response = self.patch_json('/deploy_templates/%s' % self.template.name,
+                                   patch, headers=self.headers)
+        self.assertEqual('application/json', response.content_type)
+        self.assertEqual(http_client.OK, response.status_code)
+        mock_save.assert_called_once_with(mock.ANY)
+        self.assertEqual(steps, response.json['steps'])
+
+    def test_update_by_name_with_json(self, mock_save):
+        interface = 'bios'
+        path = '/deploy_templates/%s.json' % self.template.name
+        response = self.patch_json(path,
+                                   [{'path': '/steps/0/interface',
+                                     'value': interface,
+                                     'op': 'replace'}],
+                                   headers=self.headers)
+        self.assertEqual('application/json', response.content_type)
+        self.assertEqual(http_client.OK, response.status_code)
+        self.assertEqual(interface, response.json['steps'][0]['interface'])
+
+    def test_update_name_standard_trait(self, mock_save):
+        name = 'HW_CPU_X86_VMX'
+        patch = [{'path': '/name', 'value': name, 'op': 'replace'}]
+        self._test_update_ok(mock_save, patch)
+
+    def test_update_invalid_name(self, mock_save):
+        self._test_update_bad_request(
+            mock_save,
+            [{'path': '/name', 'value': 'aa:bb_cc', 'op': 'replace'}],
+            'Deploy template name must be a valid trait')
+
+    def test_update_by_id_invalid_api_version(self, mock_save):
+        name = 'CUSTOM_DT2'
+        headers = self.invalid_version_headers
+        response = self.patch_json('/deploy_templates/%s' % self.template.uuid,
+                                   [{'path': '/name',
+                                     'value': name,
+                                     'op': 'add'}],
+                                   headers=headers,
+                                   expect_errors=True)
+        self.assertEqual(http_client.NOT_FOUND, response.status_int)
+        self.assertFalse(mock_save.called)
+
+    def test_update_not_found(self, mock_save):
+        name = 'CUSTOM_DT2'
+        uuid = uuidutils.generate_uuid()
+        response = self.patch_json('/deploy_templates/%s' % uuid,
+                                   [{'path': '/name',
+                                     'value': name,
+                                     'op': 'add'}],
+                                   expect_errors=True,
+                                   headers=self.headers)
+        self.assertEqual('application/json', response.content_type)
+        self.assertEqual(http_client.NOT_FOUND, response.status_int)
+        self.assertTrue(response.json['error_message'])
+        self.assertFalse(mock_save.called)
+
+    def test_replace_singular(self, mock_save):
+        name = 'CUSTOM_DT2'
+        patch = [{'path': '/name', 'value': name, 'op': 'replace'}]
+        response = self._test_update_ok(mock_save, patch)
+        self.assertEqual(name, response.json['name'])
+
+    @mock.patch.object(notification_utils, '_emit_api_notification',
+                       autospec=True)
+    def test_replace_name_already_exist(self, mock_notify, mock_save):
+        name = 'CUSTOM_DT2'
+        obj_utils.create_test_deploy_template(self.context,
+                                              uuid=uuidutils.generate_uuid(),
+                                              name=name)
+        mock_save.side_effect = exception.DeployTemplateAlreadyExists(
+            uuid=self.template.uuid)
+        response = self.patch_json('/deploy_templates/%s' % self.template.uuid,
+                                   [{'path': '/name',
+                                     'value': name,
+                                     'op': 'replace'}],
+                                   expect_errors=True,
+                                   headers=self.headers)
+        self.assertEqual('application/json', response.content_type)
+        self.assertEqual(http_client.CONFLICT, response.status_code)
+        self.assertTrue(response.json['error_message'])
+        mock_save.assert_called_once_with(mock.ANY)
+        mock_notify.assert_has_calls([mock.call(mock.ANY, mock.ANY, 'update',
+                                      obj_fields.NotificationLevel.INFO,
+                                      obj_fields.NotificationStatus.START),
+                                      mock.call(mock.ANY, mock.ANY, 'update',
+                                      obj_fields.NotificationLevel.ERROR,
+                                      obj_fields.NotificationStatus.ERROR)])
+
+    def test_replace_invalid_name_too_long(self, mock_save):
+        name = 'CUSTOM_' + 'X' * 249
+        patch = [{'path': '/name', 'op': 'replace', 'value': name}]
+        self._test_update_bad_request(
+            mock_save, patch, 'Deploy template name must be a valid trait')
+
+    def test_replace_invalid_name_not_a_trait(self, mock_save):
+        name = 'not-a-trait'
+        patch = [{'path': '/name', 'op': 'replace', 'value': name}]
+        self._test_update_bad_request(
+            mock_save, patch, 'Deploy template name must be a valid trait')
+
+    def test_replace_invalid_name_none(self, mock_save):
+        patch = [{'path': '/name', 'op': 'replace', 'value': None}]
+        self._test_update_bad_request(
+            mock_save, patch, "Deploy template name cannot be None")
+
+    def test_replace_duplicate_step(self, mock_save):
+        # interface & step combination must be unique.
+        steps = [
+            {
+                'interface': 'raid',
+                'step': 'create_configuration',
+                'args': {'foo': '%d' % i},
+                'priority': i,
+            }
+            for i in range(2)
+        ]
+        patch = [{'path': '/steps', 'op': 'replace', 'value': steps}]
+        self._test_update_bad_request(
+            mock_save, patch, "Duplicate deploy steps")
+
+    def test_replace_invalid_step_interface_fail(self, mock_save):
+        step = {
+            'interface': 'foo',
+            'step': 'apply_configuration',
+            'args': {'foo': 'bar'},
+            'priority': 42
+        }
+        patch = [{'path': '/steps/0', 'op': 'replace', 'value': step}]
+        self._test_update_bad_request(
+            mock_save, patch, "Invalid input for field/attribute interface.")
+
+    def test_replace_non_existent_step_fail(self, mock_save):
+        step = {
+            'interface': 'bios',
+            'step': 'apply_configuration',
+            'args': {'foo': 'bar'},
+            'priority': 42
+        }
+        patch = [{'path': '/steps/1', 'op': 'replace', 'value': step}]
+        self._test_update_bad_request(
+            mock_save, patch, "list assignment index out of range")
+
+    def test_replace_empty_step_list_fail(self, mock_save):
+        patch = [{'path': '/steps', 'op': 'replace', 'value': []}]
+        self._test_update_bad_request(
+            mock_save, patch, 'No deploy steps specified')
+
+    def _test_remove_not_allowed(self, mock_save, field, error_msg):
+        patch = [{'path': '/%s' % field, 'op': 'remove'}]
+        self._test_update_bad_request(mock_save, patch, error_msg)
+
+    def test_remove_uuid(self, mock_save):
+        self._test_remove_not_allowed(
+            mock_save, 'uuid',
+            "'/uuid' is an internal attribute and can not be updated")
+
+    def test_remove_name(self, mock_save):
+        self._test_remove_not_allowed(
+            mock_save, 'name',
+            "'/name' is a mandatory attribute and can not be removed")
+
+    def test_remove_steps(self, mock_save):
+        self._test_remove_not_allowed(
+            mock_save, 'steps',
+            "'/steps' is a mandatory attribute and can not be removed")
+
+    def test_remove_foo(self, mock_save):
+        self._test_remove_not_allowed(
+            mock_save, 'foo', "can't remove non-existent object 'foo'")
+
+    def test_replace_step_invalid_interface(self, mock_save):
+        patch = [{'path': '/steps/0/interface', 'op': 'replace',
+                  'value': 'foo'}]
+        self._test_update_bad_request(
+            mock_save, patch, "Invalid input for field/attribute interface.")
+
+    def test_replace_multi(self, mock_save):
+        steps = [
+            {
+                'interface': 'raid',
+                'step': 'create_configuration%d' % i,
+                'args': {},
+                'priority': 10,
+            }
+            for i in range(3)
+        ]
+        template = obj_utils.create_test_deploy_template(
+            self.context, uuid=uuidutils.generate_uuid(), name='CUSTOM_DT2',
+            steps=steps)
+
+        # mutate steps so we replace all of them
+        for step in steps:
+            step['priority'] = step['priority'] + 1
+
+        patch = []
+        for i, step in enumerate(steps):
+            patch.append({'path': '/steps/%s' % i,
+                          'value': steps[i],
+                          'op': 'replace'})
+        response = self.patch_json('/deploy_templates/%s' % template.uuid,
+                                   patch, headers=self.headers)
+        self.assertEqual('application/json', response.content_type)
+        self.assertEqual(http_client.OK, response.status_code)
+        self.assertEqual(steps, response.json['steps'])
+        mock_save.assert_called_once_with(mock.ANY)
+
+    def test_remove_multi(self, mock_save):
+        steps = [
+            {
+                'interface': 'raid',
+                'step': 'create_configuration%d' % i,
+                'args': {},
+                'priority': 10,
+            }
+            for i in range(3)
+        ]
+        template = obj_utils.create_test_deploy_template(
+            self.context, uuid=uuidutils.generate_uuid(), name='CUSTOM_DT2',
+            steps=steps)
+
+        # Removing one step from the collection
+        steps.pop(1)
+        response = self.patch_json('/deploy_templates/%s' % template.uuid,
+                                   [{'path': '/steps/1',
+                                     'op': 'remove'}],
+                                   headers=self.headers)
+        self.assertEqual('application/json', response.content_type)
+        self.assertEqual(http_client.OK, response.status_code)
+        self.assertEqual(steps, response.json['steps'])
+        mock_save.assert_called_once_with(mock.ANY)
+
+    def test_remove_non_existent_property_fail(self, mock_save):
+        patch = [{'path': '/non-existent', 'op': 'remove'}]
+        self._test_update_bad_request(
+            mock_save, patch,
+            "can't remove non-existent object 'non-existent'")
+
+    def test_remove_non_existent_step_fail(self, mock_save):
+        patch = [{'path': '/steps/1', 'op': 'remove'}]
+        self._test_update_bad_request(
+            mock_save, patch, "can't remove non-existent object '1'")
+
+    def test_remove_only_step_fail(self, mock_save):
+        patch = [{'path': '/steps/0', 'op': 'remove'}]
+        self._test_update_bad_request(
+            mock_save, patch, "No deploy steps specified")
+
+    def test_remove_non_existent_step_property_fail(self, mock_save):
+        patch = [{'path': '/steps/0/non-existent', 'op': 'remove'}]
+        self._test_update_bad_request(
+            mock_save, patch,
+            "can't remove non-existent object 'non-existent'")
+
+    def test_add_root_non_existent(self, mock_save):
+        patch = [{'path': '/foo', 'value': 'bar', 'op': 'add'}]
+        self._test_update_bad_request(
+            mock_save, patch, "Adding a new attribute (/foo)")
+
+    def test_add_too_high_index_step_fail(self, mock_save):
+        step = {
+            'interface': 'bios',
+            'step': 'apply_configuration',
+            'args': {'foo': 'bar'},
+            'priority': 42
+        }
+        patch = [{'path': '/steps/2', 'op': 'add', 'value': step}]
+        self._test_update_bad_request(
+            mock_save, patch, "can't insert outside of list")
+
+    def test_add_multi(self, mock_save):
+        steps = [
+            {
+                'interface': 'raid',
+                'step': 'create_configuration%d' % i,
+                'args': {},
+                'priority': 10,
+            }
+            for i in range(3)
+        ]
+        patch = []
+        for i, step in enumerate(steps):
+            patch.append({'path': '/steps/%d' % i,
+                          'value': step,
+                          'op': 'add'})
+        response = self.patch_json('/deploy_templates/%s' % self.template.uuid,
+                                   patch, headers=self.headers)
+        self.assertEqual('application/json', response.content_type)
+        self.assertEqual(http_client.OK, response.status_code)
+        self.assertEqual(steps, response.json['steps'][:-1])
+        self.assertEqual(_obj_to_api_step(self.template.steps[0]),
+                         response.json['steps'][-1])
+        mock_save.assert_called_once_with(mock.ANY)
+
+
+class TestPost(BaseDeployTemplatesAPITest):
+
+    @mock.patch.object(notification_utils, '_emit_api_notification',
+                       autospec=True)
+    @mock.patch.object(timeutils, 'utcnow', autospec=True)
+    def test_create(self, mock_utcnow, mock_notify):
+        tdict = test_api_utils.post_get_test_deploy_template()
+        test_time = datetime.datetime(2000, 1, 1, 0, 0)
+        mock_utcnow.return_value = test_time
+        response = self.post_json('/deploy_templates', tdict,
+                                  headers=self.headers)
+        self.assertEqual(http_client.CREATED, response.status_int)
+        result = self.get_json('/deploy_templates/%s' % tdict['uuid'],
+                               headers=self.headers)
+        self.assertEqual(tdict['uuid'], result['uuid'])
+        self.assertFalse(result['updated_at'])
+        return_created_at = timeutils.parse_isotime(
+            result['created_at']).replace(tzinfo=None)
+        self.assertEqual(test_time, return_created_at)
+        # Check location header
+        self.assertIsNotNone(response.location)
+        expected_location = '/v1/deploy_templates/%s' % tdict['uuid']
+        self.assertEqual(expected_location,
+                         urlparse.urlparse(response.location).path)
+        mock_notify.assert_has_calls([mock.call(mock.ANY, mock.ANY, 'create',
+                                      obj_fields.NotificationLevel.INFO,
+                                      obj_fields.NotificationStatus.START),
+                                      mock.call(mock.ANY, mock.ANY, 'create',
+                                      obj_fields.NotificationLevel.INFO,
+                                      obj_fields.NotificationStatus.END)])
+
+    def test_create_invalid_api_version(self):
+        tdict = test_api_utils.post_get_test_deploy_template()
+        response = self.post_json(
+            '/deploy_templates', tdict, headers=self.invalid_version_headers,
+            expect_errors=True)
+        self.assertEqual(http_client.NOT_FOUND, response.status_int)
+
+    def test_create_doesnt_contain_id(self):
+        with mock.patch.object(
+                self.dbapi, 'create_deploy_template',
+                wraps=self.dbapi.create_deploy_template) as mock_create:
+            tdict = test_api_utils.post_get_test_deploy_template()
+            self.post_json('/deploy_templates', tdict, headers=self.headers)
+            self.get_json('/deploy_templates/%s' % tdict['uuid'],
+                          headers=self.headers)
+            mock_create.assert_called_once_with(mock.ANY)
+            # Check that 'id' is not in first arg of positional args
+            self.assertNotIn('id', mock_create.call_args[0][0])
+
+    @mock.patch.object(notification_utils.LOG, 'exception', autospec=True)
+    @mock.patch.object(notification_utils.LOG, 'warning', autospec=True)
+    def test_create_generate_uuid(self, mock_warn, mock_except):
+        tdict = test_api_utils.post_get_test_deploy_template()
+        del tdict['uuid']
+        response = self.post_json('/deploy_templates', tdict,
+                                  headers=self.headers)
+        result = self.get_json('/deploy_templates/%s' % response.json['uuid'],
+                               headers=self.headers)
+        self.assertTrue(uuidutils.is_uuid_like(result['uuid']))
+        self.assertFalse(mock_warn.called)
+        self.assertFalse(mock_except.called)
+
+    @mock.patch.object(notification_utils, '_emit_api_notification',
+                       autospec=True)
+    @mock.patch.object(objects.DeployTemplate, 'create', autospec=True)
+    def test_create_error(self, mock_create, mock_notify):
+        mock_create.side_effect = Exception()
+        tdict = test_api_utils.post_get_test_deploy_template()
+        self.post_json('/deploy_templates', tdict, headers=self.headers,
+                       expect_errors=True)
+        mock_notify.assert_has_calls([mock.call(mock.ANY, mock.ANY, 'create',
+                                      obj_fields.NotificationLevel.INFO,
+                                      obj_fields.NotificationStatus.START),
+                                      mock.call(mock.ANY, mock.ANY, 'create',
+                                      obj_fields.NotificationLevel.ERROR,
+                                      obj_fields.NotificationStatus.ERROR)])
+
+    def _test_create_ok(self, tdict):
+        response = self.post_json('/deploy_templates', tdict,
+                                  headers=self.headers)
+        self.assertEqual(http_client.CREATED, response.status_int)
+
+    def _test_create_bad_request(self, tdict, error_msg):
+        response = self.post_json('/deploy_templates', tdict,
+                                  expect_errors=True, headers=self.headers)
+        self.assertEqual(http_client.BAD_REQUEST, response.status_int)
+        self.assertEqual('application/json', response.content_type)
+        self.assertTrue(response.json['error_message'])
+        self.assertIn(error_msg, response.json['error_message'])
+
+    def test_create_long_name(self):
+        name = 'CUSTOM_' + 'X' * 248
+        tdict = test_api_utils.post_get_test_deploy_template(name=name)
+        self._test_create_ok(tdict)
+
+    def test_create_standard_trait_name(self):
+        name = 'HW_CPU_X86_VMX'
+        tdict = test_api_utils.post_get_test_deploy_template(name=name)
+        self._test_create_ok(tdict)
+
+    def test_create_name_invalid_too_long(self):
+        name = 'CUSTOM_' + 'X' * 249
+        tdict = test_api_utils.post_get_test_deploy_template(name=name)
+        self._test_create_bad_request(
+            tdict, 'Deploy template name must be a valid trait')
+
+    def test_create_name_invalid_not_a_trait(self):
+        name = 'not-a-trait'
+        tdict = test_api_utils.post_get_test_deploy_template(name=name)
+        self._test_create_bad_request(
+            tdict, 'Deploy template name must be a valid trait')
+
+    def test_create_steps_invalid_duplicate(self):
+        steps = [
+            {
+                'interface': 'raid',
+                'step': 'create_configuration',
+                'args': {'foo': '%d' % i},
+                'priority': i,
+            }
+            for i in range(2)
+        ]
+        tdict = test_api_utils.post_get_test_deploy_template(steps=steps)
+        self._test_create_bad_request(tdict, "Duplicate deploy steps")
+
+    def _test_create_no_mandatory_field(self, field):
+        tdict = test_api_utils.post_get_test_deploy_template()
+        del tdict[field]
+        self._test_create_bad_request(tdict, "Mandatory field missing")
+
+    def test_create_no_mandatory_field_name(self):
+        self._test_create_no_mandatory_field('name')
+
+    def test_create_no_mandatory_field_steps(self):
+        self._test_create_no_mandatory_field('steps')
+
+    def _test_create_no_mandatory_step_field(self, field):
+        tdict = test_api_utils.post_get_test_deploy_template()
+        del tdict['steps'][0][field]
+        self._test_create_bad_request(tdict, "Mandatory field missing")
+
+    def test_create_no_mandatory_step_field_interface(self):
+        self._test_create_no_mandatory_step_field('interface')
+
+    def test_create_no_mandatory_step_field_step(self):
+        self._test_create_no_mandatory_step_field('step')
+
+    def test_create_no_mandatory_step_field_args(self):
+        self._test_create_no_mandatory_step_field('args')
+
+    def test_create_no_mandatory_step_field_priority(self):
+        self._test_create_no_mandatory_step_field('priority')
+
+    def _test_create_invalid_field(self, field, value, error_msg):
+        tdict = test_api_utils.post_get_test_deploy_template()
+        tdict[field] = value
+        self._test_create_bad_request(tdict, error_msg)
+
+    def test_create_invalid_field_name(self):
+        self._test_create_invalid_field(
+            'name', 42, 'Invalid input for field/attribute name')
+
+    def test_create_invalid_field_name_none(self):
+        self._test_create_invalid_field(
+            'name', None, "Deploy template name cannot be None")
+
+    def test_create_invalid_field_steps(self):
+        self._test_create_invalid_field(
+            'steps', {}, "Invalid input for field/attribute template")
+
+    def test_create_invalid_field_empty_steps(self):
+        self._test_create_invalid_field(
+            'steps', [], "No deploy steps specified")
+
+    def test_create_invalid_field_extra(self):
+        self._test_create_invalid_field(
+            'extra', 42, "Invalid input for field/attribute template")
+
+    def test_create_invalid_field_foo(self):
+        self._test_create_invalid_field(
+            'foo', 'bar', "Unknown attribute for argument template: foo")
+
+    def _test_create_invalid_step_field(self, field, value, error_msg=None):
+        tdict = test_api_utils.post_get_test_deploy_template()
+        tdict['steps'][0][field] = value
+        if error_msg is None:
+            error_msg = "Invalid input for field/attribute"
+        self._test_create_bad_request(tdict, error_msg)
+
+    def test_create_invalid_step_field_interface1(self):
+        self._test_create_invalid_step_field('interface', [3])
+
+    def test_create_invalid_step_field_interface2(self):
+        self._test_create_invalid_step_field('interface', 'foo')
+
+    def test_create_invalid_step_field_step(self):
+        self._test_create_invalid_step_field('step', 42)
+
+    def test_create_invalid_step_field_args1(self):
+        self._test_create_invalid_step_field('args', 'not a dict')
+
+    def test_create_invalid_step_field_args2(self):
+        self._test_create_invalid_step_field('args', [])
+
+    def test_create_invalid_step_field_priority(self):
+        self._test_create_invalid_step_field('priority', 'not a number')
+
+    def test_create_invalid_step_field_negative_priority(self):
+        self._test_create_invalid_step_field('priority', -1)
+
+    def test_create_invalid_step_field_foo(self):
+        self._test_create_invalid_step_field(
+            'foo', 'bar', "Unknown attribute for argument template.steps: foo")
+
+    def test_create_step_string_priority(self):
+        tdict = test_api_utils.post_get_test_deploy_template()
+        tdict['steps'][0]['priority'] = '42'
+        self._test_create_ok(tdict)
+
+    def test_create_complex_step_args(self):
+        tdict = test_api_utils.post_get_test_deploy_template()
+        tdict['steps'][0]['args'] = {'foo': [{'bar': 'baz'}]}
+        self._test_create_ok(tdict)
+
+
+@mock.patch.object(objects.DeployTemplate, 'destroy', autospec=True)
+class TestDelete(BaseDeployTemplatesAPITest):
+
+    def setUp(self):
+        super(TestDelete, self).setUp()
+        self.template = obj_utils.create_test_deploy_template(self.context)
+
+    @mock.patch.object(notification_utils, '_emit_api_notification',
+                       autospec=True)
+    def test_delete_by_uuid(self, mock_notify, mock_destroy):
+        self.delete('/deploy_templates/%s' % self.template.uuid,
+                    headers=self.headers)
+        mock_destroy.assert_called_once_with(mock.ANY)
+        mock_notify.assert_has_calls([mock.call(mock.ANY, mock.ANY, 'delete',
+                                      obj_fields.NotificationLevel.INFO,
+                                      obj_fields.NotificationStatus.START),
+                                      mock.call(mock.ANY, mock.ANY, 'delete',
+                                      obj_fields.NotificationLevel.INFO,
+                                      obj_fields.NotificationStatus.END)])
+
+    def test_delete_by_uuid_with_json(self, mock_destroy):
+        self.delete('/deploy_templates/%s.json' % self.template.uuid,
+                    headers=self.headers)
+        mock_destroy.assert_called_once_with(mock.ANY)
+
+    def test_delete_by_name(self, mock_destroy):
+        self.delete('/deploy_templates/%s' % self.template.name,
+                    headers=self.headers)
+        mock_destroy.assert_called_once_with(mock.ANY)
+
+    def test_delete_by_name_with_json(self, mock_destroy):
+        self.delete('/deploy_templates/%s.json' % self.template.name,
+                    headers=self.headers)
+        mock_destroy.assert_called_once_with(mock.ANY)
+
+    def test_delete_invalid_api_version(self, mock_dpt):
+        response = self.delete('/deploy_templates/%s' % self.template.uuid,
+                               expect_errors=True,
+                               headers=self.invalid_version_headers)
+        self.assertEqual(http_client.NOT_FOUND, response.status_int)
+
+    def test_delete_by_name_non_existent(self, mock_dpt):
+        res = self.delete('/deploy_templates/%s' % 'blah', expect_errors=True,
+                          headers=self.headers)
+        self.assertEqual(http_client.NOT_FOUND, res.status_code)
diff --git a/ironic/tests/unit/api/controllers/v1/test_utils.py b/ironic/tests/unit/api/controllers/v1/test_utils.py
index 7b6411feb..d456ea10f 100644
--- a/ironic/tests/unit/api/controllers/v1/test_utils.py
+++ b/ironic/tests/unit/api/controllers/v1/test_utils.py
@@ -26,6 +26,7 @@ import wsme
 from ironic.api.controllers.v1 import node as api_node
 from ironic.api.controllers.v1 import utils
 from ironic.common import exception
+from ironic.common import policy
 from ironic.common import states
 from ironic import objects
 from ironic.tests import base
@@ -80,6 +81,10 @@ class TestApiUtils(base.TestCase):
         utils.validate_trait(large)
         self.assertRaises(wsme.exc.ClientSideError,
                           utils.validate_trait, large + "1")
+        # Check custom error prefix.
+        self.assertRaisesRegex(wsme.exc.ClientSideError,
+                               "spongebob",
+                               utils.validate_trait, "invalid", "spongebob")
 
     def test_get_patch_values_no_path(self):
         patch = [{'path': '/name', 'op': 'update', 'value': 'node-0'}]
@@ -530,6 +535,13 @@ class TestApiUtils(base.TestCase):
         mock_request.version.minor = 52
         self.assertFalse(utils.allow_port_is_smartnic())
 
+    @mock.patch.object(pecan, 'request', spec_set=['version'])
+    def test_allow_deploy_templates(self, mock_request):
+        mock_request.version.minor = 55
+        self.assertTrue(utils.allow_deploy_templates())
+        mock_request.version.minor = 54
+        self.assertFalse(utils.allow_deploy_templates())
+
 
 class TestNodeIdent(base.TestCase):
 
@@ -717,6 +729,20 @@ class TestVendorPassthru(base.TestCase):
                          sorted(utils.get_controller_reserved_names(
                                 api_node.NodesController)))
 
+    @mock.patch.object(pecan, 'request', spec_set=["context"])
+    @mock.patch.object(policy, 'authorize', spec=True)
+    def test_check_policy(self, mock_authorize, mock_pr):
+        utils.check_policy('fake-policy')
+        cdict = pecan.request.context.to_policy_values()
+        mock_authorize.assert_called_once_with('fake-policy', cdict, cdict)
+
+    @mock.patch.object(pecan, 'request', spec_set=["context"])
+    @mock.patch.object(policy, 'authorize', spec=True)
+    def test_check_policy_forbidden(self, mock_authorize, mock_pr):
+        mock_authorize.side_effect = exception.HTTPForbidden(resource='fake')
+        self.assertRaises(exception.HTTPForbidden,
+                          utils.check_policy, 'fake-policy')
+
 
 class TestPortgroupIdent(base.TestCase):
     def setUp(self):
diff --git a/ironic/tests/unit/api/utils.py b/ironic/tests/unit/api/utils.py
index 36321154b..802e6c45c 100644
--- a/ironic/tests/unit/api/utils.py
+++ b/ironic/tests/unit/api/utils.py
@@ -20,9 +20,11 @@ import hashlib
 import json
 
 from ironic.api.controllers.v1 import chassis as chassis_controller
+from ironic.api.controllers.v1 import deploy_template as dt_controller
 from ironic.api.controllers.v1 import node as node_controller
 from ironic.api.controllers.v1 import port as port_controller
 from ironic.api.controllers.v1 import portgroup as portgroup_controller
+from ironic.api.controllers.v1 import types
 from ironic.api.controllers.v1 import utils as api_utils
 from ironic.api.controllers.v1 import volume_connector as vc_controller
 from ironic.api.controllers.v1 import volume_target as vt_controller
@@ -200,3 +202,24 @@ def allocation_post_data(**kw):
 def fake_event_validator(v):
     """A fake event validator"""
     return v
+
+
+def deploy_template_post_data(**kw):
+    """Return a DeployTemplate object without internal attributes."""
+    template = db_utils.get_test_deploy_template(**kw)
+    # These values are not part of the API object
+    template.pop('version')
+    # Remove internal attributes from each step.
+    step_internal = types.JsonPatchType.internal_attrs()
+    step_internal.append('deploy_template_id')
+    template['steps'] = [remove_internal(step, step_internal)
+                         for step in template['steps']]
+    # Remove internal attributes from the template.
+    dt_patch = dt_controller.DeployTemplatePatchType
+    internal = dt_patch.internal_attrs()
+    return remove_internal(template, internal)
+
+
+def post_get_test_deploy_template(**kw):
+    """Return a DeployTemplate object with appropriate attributes."""
+    return deploy_template_post_data(**kw)
diff --git a/ironic/tests/unit/objects/test_objects.py b/ironic/tests/unit/objects/test_objects.py
index 0717cc42d..865b597d6 100644
--- a/ironic/tests/unit/objects/test_objects.py
+++ b/ironic/tests/unit/objects/test_objects.py
@@ -718,6 +718,8 @@ expected_object_fingerprints = {
     'AllocationCRUDNotification': '1.0-59acc533c11d306f149846f922739c15',
     'AllocationCRUDPayload': '1.0-a82389d019f37cfe54b50049f73911b3',
     'DeployTemplate': '1.1-4e30c8e9098595e359bb907f095bf1a9',
+    'DeployTemplateCRUDNotification': '1.0-59acc533c11d306f149846f922739c15',
+    'DeployTemplateCRUDPayload': '1.0-200857e7e715f58a5b6d6b700ab73a3b',
 }