diff --git a/tripleoclient/tests/test_utils.py b/tripleoclient/tests/test_utils.py
index f1b6e2b79..51e23c7d6 100644
--- a/tripleoclient/tests/test_utils.py
+++ b/tripleoclient/tests/test_utils.py
@@ -1224,16 +1224,23 @@ class ProcessMultipleEnvironments(TestCase):
                                             self.user_tht_root)
 
         mock_hc_process.assert_has_calls([
-            mock.call(env_path='./inside.yaml'),
-            mock.call(env_path='/twd/templates/abs.yaml'),
-            mock.call(env_path='/twd/templates/puppet/foo.yaml'),
-            mock.call(env_path='/twd/templates/environments/myenv.yaml'),
-            mock.call(env_path='/tmp/thtroot42/notouch.yaml'),
-            mock.call(env_path='./tmp/thtroot/notouch2.yaml'),
-            mock.call(env_path='../outside.yaml')])
+            mock.call(env_path='./inside.yaml',
+                      include_env_in_files=False),
+            mock.call(env_path='/twd/templates/abs.yaml',
+                      include_env_in_files=False),
+            mock.call(env_path='/twd/templates/puppet/foo.yaml',
+                      include_env_in_files=False),
+            mock.call(env_path='/twd/templates/environments/myenv.yaml',
+                      include_env_in_files=False),
+            mock.call(env_path='/tmp/thtroot42/notouch.yaml',
+                      include_env_in_files=False),
+            mock.call(env_path='./tmp/thtroot/notouch2.yaml',
+                      include_env_in_files=False),
+            mock.call(env_path='../outside.yaml',
+                      include_env_in_files=False)])
 
     @mock.patch('heatclient.common.template_utils.'
-                'process_environment_and_files', return_value=({}, {}),
+                'process_environment_and_files',
                 autospec=True)
     @mock.patch('heatclient.common.template_utils.'
                 'get_template_contents', return_value=({}, {}),
@@ -1282,7 +1289,7 @@ class ProcessMultipleEnvironments(TestCase):
 
         utils.process_multiple_environments(self.created_env_files,
                                             self.tht_root,
-                                            self.user_tht_root, False)
+                                            self.user_tht_root, None, False)
 
         mock_yaml_dump.assert_has_calls([mock.call(rewritten_env,
                                         default_flow_style=False)])
diff --git a/tripleoclient/tests/v1/overcloud_deploy/test_overcloud_deploy.py b/tripleoclient/tests/v1/overcloud_deploy/test_overcloud_deploy.py
index 2dbb5d004..7cc360a10 100644
--- a/tripleoclient/tests/v1/overcloud_deploy/test_overcloud_deploy.py
+++ b/tripleoclient/tests/v1/overcloud_deploy/test_overcloud_deploy.py
@@ -20,7 +20,6 @@ import six
 import tempfile
 import yaml
 
-from heatclient import exc as hc_exc
 import mock
 import openstack
 from osc_lib import exceptions as oscexc
@@ -177,6 +176,15 @@ class TestDeployOvercloud(fakes.TestDeployOvercloud):
 
     @mock.patch('tripleoclient.utils.get_rc_params',
                 autospec=True)
+    @mock.patch('tripleo_common.utils.plan.generate_passwords',
+                return_value={})
+    @mock.patch(
+        'tripleo_common.image.kolla_builder.container_images_prepare_multi',
+        return_value={})
+    @mock.patch('tripleoclient.workflows.roles.get_roles_data',
+                autospec=True, return_value={})
+    @mock.patch('heatclient.common.template_utils.'
+                'process_environment_and_files', autospec=True)
     @mock.patch('tripleoclient.utils.check_nic_config_with_ansible')
     @mock.patch('tripleoclient.v1.overcloud_deploy.DeployOvercloud.'
                 '_get_ctlplane_attrs', autospec=True, return_value={})
@@ -203,6 +211,8 @@ class TestDeployOvercloud(fakes.TestDeployOvercloud):
                        mock_ceph_fsid,
                        mock_get_undercloud_host_entry, mock_copy,
                        mock_get_ctlplane_attrs, mock_nic_ansiblei,
+                       mock_process_env, mock_roles_data,
+                       mock_container_prepare, mock_generate_password,
                        mock_rc_params):
         fixture = deployment.DeploymentWorkflowFixture()
         self.useFixture(fixture)
@@ -225,7 +235,7 @@ class TestDeployOvercloud(fakes.TestDeployOvercloud):
         mock_event = mock.Mock()
         mock_event.id = '1234'
         mock_events.return_value = [mock_events]
-
+        mock_roles_data.return_value = []
         object_client = clients.tripleoclient.object_store
         object_client.get_object = mock.Mock()
         object_client.put_container = mock.Mock()
@@ -269,16 +279,16 @@ class TestDeployOvercloud(fakes.TestDeployOvercloud):
             return parameter_defaults
 
         mock_create_parameters_env.side_effect = _custom_create_params_env
-
         mock_rc_params.return_value = {'password': 'password',
                                        'region': 'region1'}
+        mock_process_env.return_value = {}, {
+            'parameter_defaults': expected_parameters}
         self.cmd.take_action(parsed_args)
 
-        self.assertFalse(orchestration_client.stacks.update.called)
+        self.assertTrue(orchestration_client.stacks.update.called)
 
         mock_get_template_contents.assert_called_with(
-            object_request=mock.ANY,
-            template_object=constants.OVERCLOUD_YAML_NAME)
+            template_file=mock.ANY)
 
         mock_create_tempest_deployer_input.assert_called_with()
         mock_copy.assert_called_once()
@@ -286,6 +296,15 @@ class TestDeployOvercloud(fakes.TestDeployOvercloud):
             'overcloud', headers={'x-container-meta-usage-tripleo': 'plan'})
 
     @mock.patch('tripleoclient.utils.get_rc_params', autospec=True)
+    @mock.patch('tripleo_common.utils.plan.generate_passwords',
+                return_value={})
+    @mock.patch(
+        'tripleo_common.image.kolla_builder.container_images_prepare_multi',
+        return_value={})
+    @mock.patch('tripleoclient.workflows.roles.get_roles_data',
+                autospec=True, return_value={})
+    @mock.patch('heatclient.common.template_utils.'
+                'process_environment_and_files', autospec=True)
     @mock.patch('tripleoclient.v1.overcloud_deploy.DeployOvercloud.'
                 '_get_ctlplane_attrs', autospec=True, return_value={})
     @mock.patch('tripleoclient.workflows.deployment.create_overcloudrc',
@@ -313,7 +332,10 @@ class TestDeployOvercloud(fakes.TestDeployOvercloud):
                         mock_invoke_plan_env_wf,
                         mock_get_undercloud_host_entry,
                         mock_copy, mock_overcloudrc,
-                        mock_get_ctlplane_attrs, mock_rc_params):
+                        mock_get_ctlplane_attrs,
+                        mock_process_env, mock_roles_data,
+                        mock_container_prepare, mock_generate_password,
+                        mock_rc_params):
         fixture = deployment.DeploymentWorkflowFixture()
         self.useFixture(fixture)
         plane_management_fixture = deployment.PlanManagementFixture()
@@ -365,16 +387,15 @@ class TestDeployOvercloud(fakes.TestDeployOvercloud):
                 'UndercloudHostsEntries':
                     ['192.168.0.1 uc.ctlplane.localhost uc.ctlplane'],
                 'CtlplaneNetworkAttributes': {}}}
-
+        mock_process_env.return_value = {}, parameters_env
         mock_open_context = mock.mock_open()
         with mock.patch('six.moves.builtins.open', mock_open_context):
             self.cmd.take_action(parsed_args)
 
-        self.assertFalse(orchestration_client.stacks.create.called)
+        self.assertTrue(orchestration_client.stacks.create.called)
 
         mock_get_template_contents.assert_called_with(
-            object_request=mock.ANY,
-            template_object=constants.OVERCLOUD_YAML_NAME)
+            template_file=mock.ANY)
 
         utils_overcloud_fixture.mock_deploy_tht.assert_called_with()
 
@@ -386,20 +407,21 @@ class TestDeployOvercloud(fakes.TestDeployOvercloud):
             mock.call('overcloud',
                       'user-environments/tripleoclient-parameters.yaml',
                       yaml.safe_dump(parameters_env,
-                                     default_flow_style=False)),
-            mock.call('overcloud',
-                      'user-environment.yaml',
-                      yaml.safe_dump({}, default_flow_style=False)),
-            mock.call('overcloud',
-                      'plan-environment.yaml',
-                      yaml.safe_dump({'environments':
-                                      [{'path': 'user-environment.yaml'}]},
                                      default_flow_style=False))]
 
         object_client = clients.tripleoclient.object_store
         object_client.put_object.assert_has_calls(calls)
 
     @mock.patch('tripleoclient.utils.get_rc_params', autospec=True)
+    @mock.patch('tripleo_common.utils.plan.generate_passwords',
+                return_value={})
+    @mock.patch(
+        'tripleo_common.image.kolla_builder.container_images_prepare_multi',
+        return_value={})
+    @mock.patch('tripleoclient.workflows.roles.get_roles_data',
+                autospec=True, return_value={})
+    @mock.patch('heatclient.common.template_utils.'
+                'process_environment_and_files', autospec=True)
     @mock.patch('tripleoclient.workflows.deployment.create_overcloudrc',
                 autospec=True)
     @mock.patch('os.chdir')
@@ -431,6 +453,8 @@ class TestDeployOvercloud(fakes.TestDeployOvercloud):
             mock_postconfig, mock_deprecated_params, mock_stack_network_check,
             mock_ceph_fsid, mock_get_undercloud_host_entry, mock_copy,
             mock_chdir, mock_overcloudrc,
+            mock_process_env, mock_roles_data,
+            mock_image_prepare, mock_generate_password,
             mock_rc_params):
         fixture = deployment.DeploymentWorkflowFixture()
         self.useFixture(fixture)
@@ -470,6 +494,7 @@ class TestDeployOvercloud(fakes.TestDeployOvercloud):
         }
         mock_get_template_contents.return_value = [{}, "template"]
 
+        mock_process_env.return_value = {}, {}
         parsed_args = self.check_parser(self.cmd, arglist, verifylist)
 
         baremetal = clients.baremetal
@@ -483,6 +508,15 @@ class TestDeployOvercloud(fakes.TestDeployOvercloud):
             'overcloud', headers={'x-container-meta-usage-tripleo': 'plan'})
 
     @mock.patch('tripleoclient.utils.get_rc_params', autospec=True)
+    @mock.patch('tripleo_common.utils.plan.generate_passwords',
+                return_value={})
+    @mock.patch(
+        'tripleo_common.image.kolla_builder.container_images_prepare_multi',
+        return_value={})
+    @mock.patch('tripleoclient.workflows.roles.get_roles_data',
+                autospec=True, return_value={})
+    @mock.patch('heatclient.common.template_utils.'
+                'process_environment_and_files', autospec=True)
     @mock.patch('tripleoclient.utils.check_nic_config_with_ansible')
     @mock.patch('tripleoclient.utils.copy_clouds_yaml')
     @mock.patch('tripleoclient.v1.overcloud_deploy.DeployOvercloud.'
@@ -507,6 +541,10 @@ class TestDeployOvercloud(fakes.TestDeployOvercloud):
                                      mock_ceph_fsid,
                                      mock_get_undercloud_host_entry,
                                      mock_copy, mock_nic_ansible,
+                                     mock_process_env,
+                                     mock_roles_data,
+                                     mock_image_prepare,
+                                     mock_generate_password,
                                      mock_rc_params):
         fixture = deployment.DeploymentWorkflowFixture()
         self.useFixture(fixture)
@@ -540,20 +578,20 @@ class TestDeployOvercloud(fakes.TestDeployOvercloud):
         object_client = clients.tripleoclient.object_store
         object_client.get_object = mock.Mock()
         object_client.put_container = mock.Mock()
-        mock_env = yaml.safe_dump({'environments': []})
-        object_client.get_object.return_value = ({}, mock_env)
 
+        env = {'parameter_defaults': {},
+               'resource_registry': {}}
+        mock_process_env.return_value = {}, env
         with mock.patch('tempfile.mkstemp') as mkstemp:
             mkstemp.return_value = (os.open(self.parameter_defaults_env_file,
                                             os.O_RDWR),
                                     self.parameter_defaults_env_file)
             self.cmd.take_action(parsed_args)
 
-        self.assertFalse(orchestration_client.stacks.update.called)
+        self.assertTrue(orchestration_client.stacks.update.called)
 
         mock_get_template_contents.assert_called_with(
-            object_request=mock.ANY,
-            template_object=constants.OVERCLOUD_YAML_NAME)
+            template_file=mock.ANY)
 
         mock_create_tempest_deployer_input.assert_called_with()
         mock_copy.assert_called_once()
@@ -578,6 +616,8 @@ class TestDeployOvercloud(fakes.TestDeployOvercloud):
         self.assertFalse(mock_deploy_tht.called)
 
     @mock.patch('tripleoclient.utils.get_rc_params', autospec=True)
+    @mock.patch('heatclient.common.template_utils.'
+                'process_environment_and_files', autospec=True)
     @mock.patch('tripleoclient.utils.check_nic_config_with_ansible')
     @mock.patch('tripleoclient.utils.copy_clouds_yaml')
     @mock.patch('tripleoclient.utils.check_stack_network_matches_env_files')
@@ -592,7 +632,7 @@ class TestDeployOvercloud(fakes.TestDeployOvercloud):
                               mock_update_parameters, mock_post_config,
                               mock_stack_network_check, mock_ceph_fsid,
                               mock_copy, mock_nic_ansible,
-                              mock_rc_params):
+                              mock_process_env, mock_rc_params):
         fixture = deployment.DeploymentWorkflowFixture()
         self.useFixture(fixture)
         plane_management_fixture = deployment.PlanManagementFixture()
@@ -612,6 +652,13 @@ class TestDeployOvercloud(fakes.TestDeployOvercloud):
         env_dirs = [os.path.join(os.environ.get('HOME', ''), '.tripleo',
                     'environments'), self.tmp_dir.path]
 
+        env = {'parameter_defaults': {},
+               'resource_registry': {
+                   'Test': 'OS::Heat::None',
+                   'resources': {'*': {'*': {
+                       'UpdateDeployment': {'hooks': []}}}}}}
+
+        mock_process_env.return_value = {}, env
         with open(test_env, 'w') as temp_file:
             temp_file.write('resource_registry:\n  Test: OS::Heat::None')
 
@@ -627,10 +674,10 @@ class TestDeployOvercloud(fakes.TestDeployOvercloud):
         def _fake_heat_deploy(self, stack, stack_name, template_path,
                               parameters, environments, timeout, tht_root,
                               env, run_validations,
-                              skip_deploy_identifier, plan_env_file,
+                              env_files_tracker=None,
                               deployment_options=None):
             assertEqual(
-                {'parameter_defaults': {'NovaComputeLibvirtType': 'qemu'},
+                {'parameter_defaults': {},
                  'resource_registry': {
                      'Test': 'OS::Heat::None',
                      'resources': {'*': {'*': {
@@ -642,69 +689,12 @@ class TestDeployOvercloud(fakes.TestDeployOvercloud):
         object_client = clients.tripleoclient.object_store
         object_client.get_object = mock.Mock()
         object_client.put_container = mock.Mock()
-        mock_env = yaml.safe_dump({'parameter_defaults':
-                                  {'NovaComputeLibvirtType': 'qemu'}})
-        object_client.get_object.return_value = ({}, mock_env)
-
         parsed_args = self.check_parser(self.cmd, arglist, verifylist)
         self.cmd.take_action(parsed_args)
         mock_copy.assert_called_once()
         object_client.put_container.assert_called_once_with(
             'overcloud', headers={'x-container-meta-usage-tripleo': 'plan'})
 
-    @mock.patch('tripleoclient.v1.overcloud_deploy.DeployOvercloud.'
-                '_deploy_postconfig', autospec=True)
-    @mock.patch('tripleoclient.v1.overcloud_deploy.DeployOvercloud.'
-                '_update_parameters', autospec=True)
-    @mock.patch('tripleoclient.v1.overcloud_deploy.DeployOvercloud.'
-                '_heat_deploy', autospec=True)
-    def test_environment_dirs_env_files_not_found(self, mock_deploy_heat,
-                                                  mock_update_parameters,
-                                                  mock_post_config):
-        plane_management_fixture = deployment.PlanManagementFixture()
-        self.useFixture(plane_management_fixture)
-        utils_fixture = deployment.UtilsOvercloudFixture()
-        self.useFixture(utils_fixture)
-        clients = self.app.client_manager
-
-        mock_update_parameters.return_value = {}
-        utils_fixture.mock_utils_endpoint.return_value = 'foo.bar'
-        os.mkdir(self.tmp_dir.join('env'))
-        os.mkdir(self.tmp_dir.join('common'))
-
-        test_env = self.tmp_dir.join('env/foo2.yaml')
-
-        with open(test_env, 'w') as temp_file:
-            temp_file.write('resource_registry:\n  '
-                            'Test1: ../common/bar.yaml\n  '
-                            'Test2: /tmp/doesnexit.yaml')
-
-        test_sub_env = self.tmp_dir.join('common/bar.yaml')
-        with open(test_sub_env, 'w') as temp_file:
-            temp_file.write('outputs:\n  data:\n    value: 1')
-
-        arglist = ['--templates']
-        verifylist = [
-            ('templates', '/usr/share/openstack-tripleo-heat-templates/'),
-        ]
-        self.useFixture(
-            fixtures.EnvironmentVariable('TRIPLEO_ENVIRONMENT_DIRECTORY',
-                                         self.tmp_dir.join('env')))
-
-        object_client = clients.tripleoclient.object_store
-        object_client.get_object = mock.Mock()
-        object_client.put_container = mock.Mock()
-        mock_env = yaml.safe_dump({'parameter_defaults':
-                                  {'NovaComputeLibvirtType': 'qemu'}})
-        object_client.get_object.return_value = ({}, mock_env)
-
-        parsed_args = self.check_parser(self.cmd, arglist, verifylist)
-        error = self.assertRaises(hc_exc.CommandError, self.cmd.take_action,
-                                  parsed_args)
-        self.assertIn('tmp/doesnexit.yaml', str(error))
-        object_client.put_container.assert_called_once_with(
-            'overcloud', headers={'x-container-meta-usage-tripleo': 'plan'})
-
     @mock.patch('tripleoclient.v1.overcloud_deploy.DeployOvercloud.'
                 '_deploy_postconfig', autospec=True)
     @mock.patch('tripleoclient.v1.overcloud_deploy.DeployOvercloud.'
@@ -770,14 +760,14 @@ class TestDeployOvercloud(fakes.TestDeployOvercloud):
             self, mock_heat_deploy_func):
         result = self.cmd._try_overcloud_deploy_with_compat_yaml(
             '/fake/path', {}, 'overcloud', {}, ['~/overcloud-env.json'], 1,
-            {}, False, True, False, None)
+            {}, False, None, None)
         # If it returns None it succeeded
         self.assertIsNone(result)
         mock_heat_deploy_func.assert_called_once_with(
             self.cmd, {}, 'overcloud',
             '/fake/path/' + constants.OVERCLOUD_YAML_NAME, {},
-            ['~/overcloud-env.json'], 1, '/fake/path', {}, False, True,
-            False, deployment_options=None)
+            ['~/overcloud-env.json'], 1, '/fake/path', {}, False,
+            deployment_options=None, env_files_tracker=None)
 
     @mock.patch('tripleoclient.v1.overcloud_deploy.DeployOvercloud.'
                 '_heat_deploy', autospec=True)
@@ -788,7 +778,7 @@ class TestDeployOvercloud(fakes.TestDeployOvercloud):
                           self.cmd._try_overcloud_deploy_with_compat_yaml,
                           '/fake/path', mock.ANY, mock.ANY, mock.ANY,
                           mock.ANY, mock.ANY, mock.ANY, mock.ANY, mock.ANY,
-                          mock.ANY, None)
+                          None)
 
     @mock.patch('tripleoclient.v1.overcloud_deploy.DeployOvercloud.'
                 '_heat_deploy', autospec=True)
@@ -799,7 +789,7 @@ class TestDeployOvercloud(fakes.TestDeployOvercloud):
         try:
             self.cmd._try_overcloud_deploy_with_compat_yaml(
                 '/fake/path', mock.ANY, mock.ANY, mock.ANY,
-                mock.ANY, mock.ANY, mock.ANY, mock.ANY, mock.ANY, mock.ANY,
+                mock.ANY, mock.ANY, mock.ANY, mock.ANY, mock.ANY,
                 None)
         except ValueError as value_error:
             self.assertIn('/fake/path', str(value_error))
@@ -829,6 +819,13 @@ class TestDeployOvercloud(fakes.TestDeployOvercloud):
         self.assertFalse(mock_create_tempest_deployer_input.called)
 
     @mock.patch('tripleoclient.utils.get_rc_params', autospec=True)
+    @mock.patch('tripleo_common.utils.plan.generate_passwords',
+                return_value={})
+    @mock.patch(
+        'tripleo_common.image.kolla_builder.container_images_prepare_multi',
+        return_value={})
+    @mock.patch('tripleoclient.workflows.roles.get_roles_data',
+                autospec=True, return_value={})
     @mock.patch('tripleoclient.utils.check_nic_config_with_ansible')
     @mock.patch('tripleoclient.utils.copy_clouds_yaml')
     @mock.patch('tripleoclient.v1.overcloud_deploy.DeployOvercloud.'
@@ -844,7 +841,8 @@ class TestDeployOvercloud(fakes.TestDeployOvercloud):
                           mock_heat_deploy, mock_stack_network_check,
                           mock_ceph_fsid, mock_get_undercloud_host_entry,
                           mock_copy, mock_nic_ansible,
-                          mock_rc_params):
+                          mock_roles_data, mock_image_prepare,
+                          mock_generate_password, mock_rc_params):
         fixture = deployment.DeploymentWorkflowFixture()
         self.useFixture(fixture)
         clients = self.app.client_manager
@@ -873,6 +871,12 @@ class TestDeployOvercloud(fakes.TestDeployOvercloud):
         with open(test_env2, 'w') as temp_file:
             temp_file.write('resource_registry:\n  Test2: OS::Heat::None')
 
+        os.makedirs(self.tmp_dir.join('tripleo-heat-templates'))
+        reg_file = self.tmp_dir.join(
+            'tripleo-heat-templates/overcloud-resource-registry-puppet.yaml')
+        with open(reg_file, 'w+') as temp_file:
+            temp_file.write('resource_registry:\n  Test2: OS::Heat::None')
+
         test_answerfile = self.tmp_dir.join('answerfile')
         with open(test_answerfile, 'w') as answerfile:
             yaml.dump(
@@ -922,6 +926,13 @@ class TestDeployOvercloud(fakes.TestDeployOvercloud):
             'overcloud', headers={'x-container-meta-usage-tripleo': 'plan'})
 
     @mock.patch('tripleoclient.utils.get_rc_params', autospec=True)
+    @mock.patch('tripleo_common.utils.plan.generate_passwords',
+                return_value={})
+    @mock.patch(
+        'tripleo_common.image.kolla_builder.container_images_prepare_multi',
+        return_value={})
+    @mock.patch('tripleoclient.workflows.roles.get_roles_data',
+                autospec=True, return_value={})
     @mock.patch('tripleoclient.utils.check_nic_config_with_ansible')
     @mock.patch('tripleoclient.v1.overcloud_deploy.DeployOvercloud.'
                 '_get_undercloud_host_entry', autospec=True,
@@ -941,6 +952,9 @@ class TestDeployOvercloud(fakes.TestDeployOvercloud):
                                   mock_ceph_fsid,
                                   mock_get_undercloud_host_entry,
                                   mock_nic_ansible,
+                                  mock_roles_data,
+                                  mock_image_prepare,
+                                  mock_generate_password,
                                   mock_rc_params):
         plane_management_fixture = deployment.PlanManagementFixture()
         self.useFixture(plane_management_fixture)
@@ -983,6 +997,13 @@ class TestDeployOvercloud(fakes.TestDeployOvercloud):
             'overcloud', headers={'x-container-meta-usage-tripleo': 'plan'})
 
     @mock.patch('tripleoclient.utils.get_rc_params', autospec=True)
+    @mock.patch('tripleo_common.utils.plan.generate_passwords',
+                return_value={})
+    @mock.patch(
+        'tripleo_common.image.kolla_builder.container_images_prepare_multi',
+        return_value={})
+    @mock.patch('tripleoclient.workflows.roles.get_roles_data',
+                autospec=True, return_value={})
     @mock.patch('tripleoclient.v1.overcloud_deploy.DeployOvercloud.'
                 '_get_ctlplane_attrs', autospec=True, return_value={})
     @mock.patch('tripleoclient.utils.copy_clouds_yaml')
@@ -1015,6 +1036,9 @@ class TestDeployOvercloud(fakes.TestDeployOvercloud):
                                  mock_ceph_fsid,
                                  mock_get_undercloud_host_entry, mock_copy,
                                  mock_get_ctlplane_attrs,
+                                 mock_roles_data,
+                                 mock_image_prepare,
+                                 mock_generate_password,
                                  mock_rc_params):
         fixture = deployment.DeploymentWorkflowFixture()
         self.useFixture(fixture)
@@ -1094,8 +1118,7 @@ class TestDeployOvercloud(fakes.TestDeployOvercloud):
         self.cmd.take_action(parsed_args)
 
         mock_get_template_contents.assert_called_with(
-            object_request=mock.ANY,
-            template_object=constants.OVERCLOUD_YAML_NAME)
+            template_file=mock.ANY)
 
         mock_create_tempest_deployer_input.assert_called_with()
 
@@ -1295,6 +1318,15 @@ class TestDeployOvercloud(fakes.TestDeployOvercloud):
         mock_copy.assert_called_once()
 
     @mock.patch('tripleoclient.utils.get_rc_params', autospec=True)
+    @mock.patch('tripleo_common.utils.plan.generate_passwords',
+                return_value={})
+    @mock.patch(
+        'tripleo_common.image.kolla_builder.container_images_prepare_multi',
+        return_value={})
+    @mock.patch('tripleoclient.workflows.roles.get_roles_data',
+                autospec=True, return_value={})
+    @mock.patch('heatclient.common.template_utils.'
+                'process_environment_and_files', autospec=True)
     @mock.patch('tripleoclient.utils.check_nic_config_with_ansible')
     @mock.patch('tripleoclient.v1.overcloud_deploy.DeployOvercloud.'
                 '_get_ctlplane_attrs', autospec=True, return_value={})
@@ -1314,6 +1346,8 @@ class TestDeployOvercloud(fakes.TestDeployOvercloud):
             mock_ceph_fsid, mock_hd,
             mock_get_undercloud_host_entry, mock_copy,
             mock_get_ctlplane_attrs, mock_nic_ansible,
+            mock_process_env, mock_roles_data,
+            mock_container_prepare, mock_generate_password,
             mock_rc_params):
         fixture = deployment.DeploymentWorkflowFixture()
         self.useFixture(fixture)
@@ -1329,6 +1363,7 @@ class TestDeployOvercloud(fakes.TestDeployOvercloud):
             ('templates', '/usr/share/openstack-tripleo-heat-templates/'),
             ('overcloud_ssh_port_timeout', 42), ('timeout', 451)
         ]
+        mock_process_env.return_value = {}, {}
         parsed_args = self.check_parser(self.cmd, arglist, verifylist)
 
         mock_rc_params.return_value = {'password': 'password',
@@ -1345,8 +1380,9 @@ class TestDeployOvercloud(fakes.TestDeployOvercloud):
                         'RootStackName': 'overcloud',
                         'UndercloudHostsEntries':
                            ['192.168.0.1 uc.ctlplane.localhost uc.ctlplane'],
-                        'CtlplaneNetworkAttributes': {}}, {}, 451, mock.ANY,
-                       {}, False, False, None, deployment_options={})],
+                        'CtlplaneNetworkAttributes': {}}, mock.ANY,
+                       451, mock.ANY, mock.ANY, False,
+                       deployment_options={}, env_files_tracker=mock.ANY)],
             mock_hd.mock_calls)
         self.assertIn(
             [mock.call(mock.ANY, mock.ANY, mock.ANY, 'ctlplane', None, None,
diff --git a/tripleoclient/tests/v1/overcloud_upgrade/test_overcloud_upgrade.py b/tripleoclient/tests/v1/overcloud_upgrade/test_overcloud_upgrade.py
index 27c957af8..86297ace0 100644
--- a/tripleoclient/tests/v1/overcloud_upgrade/test_overcloud_upgrade.py
+++ b/tripleoclient/tests/v1/overcloud_upgrade/test_overcloud_upgrade.py
@@ -133,7 +133,7 @@ class TestOvercloudUpgradePrepare(fakes.TestOvercloudUpgradePrepare):
         check_mech.return_value = 'Wrong mech'
         mock_stack = mock.Mock(parameters={'DeployIdentifier': ''})
         argslist = (mock_stack, 'mock_stack', '/tmp', {},
-                    {}, 1, '/tmp', {}, True, False, False, None)
+                    {}, 1, '/tmp', {}, True, False, None)
         self.cmd.object_client = mock.Mock()
         self.assertRaises(oscexc.CommandError,
                           self.cmd._heat_deploy, *argslist)
diff --git a/tripleoclient/tests/v1/tripleo/test_tripleo_deploy.py b/tripleoclient/tests/v1/tripleo/test_tripleo_deploy.py
index 228a609ba..123bc5e4b 100644
--- a/tripleoclient/tests/v1/tripleo/test_tripleo_deploy.py
+++ b/tripleoclient/tests/v1/tripleo/test_tripleo_deploy.py
@@ -385,12 +385,18 @@ class TestDeployUndercloud(TestPluginV1):
         self.cmd._deploy_tripleo_heat_templates(self.orc, parsed_args)
 
         mock_hc_process.assert_has_calls([
-            mock.call(env_path='./inside.yaml'),
-            mock.call(env_path='/twd/templates/abs.yaml'),
-            mock.call(env_path='/twd/templates/puppet/foo.yaml'),
-            mock.call(env_path='/twd/templates/environments/myenv.yaml'),
-            mock.call(env_path='/tmp/thtroot42/notouch.yaml'),
-            mock.call(env_path='../outside.yaml')])
+            mock.call(env_path='./inside.yaml',
+                      include_env_in_files=False),
+            mock.call(env_path='/twd/templates/abs.yaml',
+                      include_env_in_files=False),
+            mock.call(env_path='/twd/templates/puppet/foo.yaml',
+                      include_env_in_files=False),
+            mock.call(env_path='/twd/templates/environments/myenv.yaml',
+                      include_env_in_files=False),
+            mock.call(env_path='/tmp/thtroot42/notouch.yaml',
+                      include_env_in_files=False),
+            mock.call(env_path='../outside.yaml',
+                      include_env_in_files=False)])
 
     @mock.patch('tripleoclient.utils.rel_or_abs_path')
     @mock.patch('heatclient.common.template_utils.'
diff --git a/tripleoclient/tests/workflows/test_parameters.py b/tripleoclient/tests/workflows/test_parameters.py
index 449740368..5c810013b 100644
--- a/tripleoclient/tests/workflows/test_parameters.py
+++ b/tripleoclient/tests/workflows/test_parameters.py
@@ -95,7 +95,10 @@ class TestParameterWorkflows(utils.TestCommand):
         parameters.invoke_plan_env_workflows(
             self.app.client_manager,
             'overcloud',
-            'the-plan-environment.yaml'
+            'the-plan-environment.yaml',
+            stack_data=mock.Mock(),
+            role_list=mock.Mock(),
+            derived_environment_path=mock.Mock()
         )
         calls = [
             mock.call(
@@ -104,7 +107,11 @@ class TestParameterWorkflows(utils.TestCommand):
                 workdir=mock.ANY,
                 playbook_dir=mock.ANY,
                 verbosity=0,
-                extra_vars={'num_phy_cores_per_numa_node_for_pmd': 2}
+                extra_vars={'num_phy_cores_per_numa_node_for_pmd': 2,
+                            'tripleo_get_flatten_params': {
+                                'stack_data': mock.ANY},
+                            'tripleo_role_list': mock.ANY,
+                            'derived_environment_path': mock.ANY}
             )
         ]
         mock_playbook.assert_has_calls(calls, any_order=True)
@@ -134,7 +141,10 @@ class TestParameterWorkflows(utils.TestCommand):
         parameters.invoke_plan_env_workflows(
             self.app.client_manager,
             'overcloud',
-            'the-plan-environment.yaml'
+            'the-plan-environment.yaml',
+            stack_data=mock.Mock(),
+            role_list=mock.Mock(),
+            derived_environment_path=mock.Mock()
         )
         calls = [
             mock.call(
@@ -143,7 +153,11 @@ class TestParameterWorkflows(utils.TestCommand):
                 workdir=mock.ANY,
                 playbook_dir=mock.ANY,
                 verbosity=0,
-                extra_vars={'num_phy_cores_per_numa_node_for_pmd': 2}
+                extra_vars={'num_phy_cores_per_numa_node_for_pmd': 2,
+                            'tripleo_get_flatten_params': {
+                                'stack_data': mock.ANY},
+                            'tripleo_role_list': mock.ANY,
+                            'derived_environment_path': mock.ANY}
             ),
             mock.call(
                 playbook='sample-playbook-2.yaml',
@@ -151,7 +165,11 @@ class TestParameterWorkflows(utils.TestCommand):
                 workdir=mock.ANY,
                 playbook_dir='/playbook/dir-1',
                 verbosity=0,
-                extra_vars={'some_opt': 0}
+                extra_vars={'some_opt': 0,
+                            'tripleo_get_flatten_params': {
+                                'stack_data': mock.ANY},
+                            'tripleo_role_list': mock.ANY,
+                            'derived_environment_path': mock.ANY}
             )
         ]
         mock_playbook.assert_has_calls(calls, any_order=True)
diff --git a/tripleoclient/utils.py b/tripleoclient/utils.py
index 0fe8b4689..efbcf026b 100644
--- a/tripleoclient/utils.py
+++ b/tripleoclient/utils.py
@@ -59,6 +59,7 @@ from heatclient import exc as hc_exc
 from six.moves.urllib import error as url_error
 from six.moves.urllib import request
 
+from tripleo_common.utils import stack as stack_utils
 from tripleo_common.utils import swift as swiftutils
 from tripleoclient import constants
 from tripleoclient import exceptions
@@ -1616,11 +1617,48 @@ def cleanup_tripleo_ansible_inventory_file(path):
         processutils.execute('/usr/bin/rm', '-f', path)
 
 
+def get_roles_file_path(roles_file, tht_root):
+    roles_file = roles_file or os.path.join(
+        tht_root, constants.OVERCLOUD_ROLES_FILE)
+    return os.path.abspath(roles_file)
+
+
+def get_networks_file_path(networks_file, tht_root):
+    networks_file = networks_file or os.path.join(
+        tht_root, constants.OVERCLOUD_NETWORKS_FILE)
+    return os.path.abspath(networks_file)
+
+
+def build_stack_data(clients, stack_name, template,
+                     files, env_files):
+    orchestration_client = clients.orchestration
+    fields = {
+        'template': template,
+        'files': files,
+        'environment_files': env_files,
+        'show_nested': True
+    }
+    stack_data = {}
+    result = orchestration_client.stacks.validate(**fields)
+
+    if result:
+        stack_data['environment_parameters'] = result.get(
+            'Environment', {}).get('parameter_defaults')
+        flattened = {'resources': {}, 'parameters': {}}
+        stack_utils._flat_it(flattened, 'Root', result)
+        stack_data['heat_resource_tree'] = flattened
+
+    return stack_data
+
+
 def process_multiple_environments(created_env_files, tht_root,
-                                  user_tht_root, cleanup=True):
+                                  user_tht_root,
+                                  env_files_tracker=None,
+                                  cleanup=True):
     log = logging.getLogger(__name__ + ".process_multiple_environments")
     env_files = {}
     localenv = {}
+    include_env_in_files = env_files_tracker is not None
     # Normalize paths for full match checks
     user_tht_root = os.path.normpath(user_tht_root)
     tht_root = os.path.normpath(tht_root)
@@ -1639,7 +1677,10 @@ def process_multiple_environments(created_env_files, tht_root,
             env_path = new_env_path
         try:
             files, env = template_utils.process_environment_and_files(
-                env_path=env_path)
+                env_path=env_path, include_env_in_files=include_env_in_files)
+            if env_files_tracker is not None:
+                env_files_tracker.append(
+                    heat_utils.normalise_file_path_to_url(env_path))
         except hc_exc.CommandError as ex:
             # This provides fallback logic so that we can reference files
             # inside the resource_registry values that may be rendered via
@@ -1687,7 +1728,10 @@ def process_multiple_environments(created_env_files, tht_root,
                 f.write(yaml.safe_dump(env_map, default_flow_style=False))
                 f.flush()
                 files, env = template_utils.process_environment_and_files(
-                    env_path=f.name)
+                    env_path=f.name, include_env_in_files=include_env_in_files)
+                if env_files_tracker is not None:
+                    env_files_tracker.append(
+                        heat_utils.normalise_file_path_to_url(f.name))
         if files:
             log.debug("Adding files %s for %s" % (files, env_path))
             env_files.update(files)
diff --git a/tripleoclient/v1/overcloud_deploy.py b/tripleoclient/v1/overcloud_deploy.py
index b6609b52d..79869e95c 100644
--- a/tripleoclient/v1/overcloud_deploy.py
+++ b/tripleoclient/v1/overcloud_deploy.py
@@ -34,7 +34,9 @@ from keystoneauth1.exceptions.catalog import EndpointNotFound
 import openstack
 from osc_lib import exceptions as oscexc
 from osc_lib.i18n import _
+from tripleo_common.image import kolla_builder
 from tripleo_common import update
+from tripleo_common.utils import plan as plan_utils
 
 from tripleoclient import command
 from tripleoclient import constants
@@ -43,6 +45,7 @@ from tripleoclient import utils
 from tripleoclient.workflows import deployment
 from tripleoclient.workflows import parameters as workflow_params
 from tripleoclient.workflows import plan_management
+from tripleoclient.workflows import roles
 
 CONF = cfg.CONF
 
@@ -73,7 +76,6 @@ class DeployOvercloud(command.Command):
                 parsed_args.deployed_server = True
 
     def _update_args_from_answers_file(self, args):
-        # Update parameters from answers file:
         if args.answers_file is not None:
             with open(args.answers_file, 'r') as answers_file:
                 answers = yaml.safe_load(answers_file)
@@ -85,7 +87,7 @@ class DeployOvercloud(command.Command):
                     answers['environments'].extend(args.environment_files)
                 args.environment_files = answers['environments']
 
-    def _update_parameters(self, args, stack):
+    def _update_parameters(self, args, stack, tht_root, user_tht_root):
         parameters = {}
 
         stack_is_new = stack is None
@@ -98,6 +100,29 @@ class DeployOvercloud(command.Command):
 
         parameters['StackAction'] = 'CREATE' if stack_is_new else 'UPDATE'
 
+        # We need the processed env for the image parameters atm
+        env_files = []
+        if args.environment_directories:
+            env_files.extend(utils.load_environment_directories(
+                args.environment_directories))
+        if args.environment_files:
+            env_files.extend(args.environment_files)
+
+        _, env = utils.process_multiple_environments(
+            env_files, tht_root, user_tht_root,
+            cleanup=(not args.no_cleanup))
+        image_params = kolla_builder.container_images_prepare_multi(
+            env, roles.get_roles_data(args.roles_file,
+                                      tht_root), dry_run=True)
+        if image_params:
+            parameters.update(image_params)
+
+        password_params = plan_utils.generate_passwords(
+            self.object_client, self.orchestration_client,
+            args.stack)
+
+        parameters.update(password_params)
+
         param_args = (
             ('NtpServer', 'ntp_server'),
         )
@@ -186,21 +211,21 @@ class DeployOvercloud(command.Command):
     def _create_breakpoint_cleanup_env(self, tht_root, container_name):
         bp_env = {}
         update.add_breakpoints_cleanup_into_env(bp_env)
-        env_path, swift_path = self._write_user_environment(
+        env_path, _ = self._write_user_environment(
             bp_env,
             'tripleoclient-breakpoint-cleanup.yaml',
             tht_root,
             container_name)
-        return bp_env
+        return [env_path]
 
     def _create_parameters_env(self, parameters, tht_root, container_name):
         parameter_defaults = {"parameter_defaults": parameters}
-        env_path, swift_path = self._write_user_environment(
+        env_path, _ = self._write_user_environment(
             parameter_defaults,
             'tripleoclient-parameters.yaml',
             tht_root,
             container_name)
-        return parameter_defaults
+        return [env_path]
 
     def _check_limit_skiplist_warning(self, env):
         if env.get('parameter_defaults').get('DeploymentServerBlacklist'):
@@ -242,7 +267,8 @@ class DeployOvercloud(command.Command):
 
     def _heat_deploy(self, stack, stack_name, template_path, parameters,
                      env_files, timeout, tht_root, env,
-                     run_validations, skip_deploy_identifier, plan_env_file,
+                     run_validations,
+                     env_files_tracker=None,
                      deployment_options=None):
         """Verify the Baremetal nodes are available and do a stack update"""
 
@@ -256,162 +282,19 @@ class DeployOvercloud(command.Command):
                 raise oscexc.CommandError(msg)
 
         self.log.debug("Getting template contents from plan %s" % stack_name)
-        # We need to reference the plan here, not the local
-        # tht root, as we need template_object to refer to
-        # the rendered overcloud.yaml, not the tht_root overcloud.j2.yaml
-        # FIXME(shardy) we need to move more of this into mistral actions
-        plan_yaml_path = os.path.relpath(template_path, tht_root)
-
-        # heatclient template_utils needs a function that can
-        # retrieve objects from a container by name/path
-        def do_object_request(method='GET', object_path=None):
-            obj = self.object_client.get_object(stack_name, object_path)
-            return obj and obj[1]
 
         template_files, template = template_utils.get_template_contents(
-            template_object=plan_yaml_path,
-            object_request=do_object_request)
-
+            template_file=template_path)
         files = dict(list(template_files.items()) + list(env_files.items()))
 
-        moved_files = self._upload_missing_files(
-            stack_name, files, tht_root)
-        self._process_and_upload_environment(
-            stack_name, env, moved_files, tht_root)
-
-        # Invokes the workflows specified in plan environment file
-        if plan_env_file:
-            workflow_params.invoke_plan_env_workflows(
-                self.clients,
-                stack_name,
-                plan_env_file,
-                verbosity=utils.playbook_verbosity(self=self)
-            )
-
         workflow_params.check_deprecated_parameters(self.clients, stack_name)
 
         self.log.info("Deploying templates in the directory {0}".format(
             os.path.abspath(tht_root)))
-        deployment.deploy_and_wait(
-            log=self.log,
-            clients=self.clients,
-            stack=stack,
-            plan_name=stack_name,
-            verbose_level=utils.playbook_verbosity(self=self),
-            timeout=timeout,
-            run_validations=run_validations,
-            skip_deploy_identifier=skip_deploy_identifier,
-            deployment_options=deployment_options
-        )
-
-    def _process_and_upload_environment(self, container_name,
-                                        env, moved_files, tht_root):
-        """Process the environment and upload to Swift
-
-        The environment at this point should be the result of the merged
-        custom user environments. We need to look at the paths in the
-        environment and update any that changed when they were uploaded to
-        swift.
-        """
-
-        file_prefix = "file://"
-
-        if env.get('resource_registry'):
-            for name, path in env['resource_registry'].items():
-                if not isinstance(path, six.string_types):
-                    continue
-                if path in moved_files:
-                    new_path = moved_files[path]
-                    env['resource_registry'][name] = new_path
-                elif path.startswith(file_prefix):
-                    path = path[len(file_prefix):]
-                    if path.startswith(tht_root):
-                        path = path[len(tht_root):]
-                    # We want to make sure all the paths are relative.
-                    if path.startswith("/"):
-                        path = path[1:]
-                    env['resource_registry'][name] = path
-
-        # Parameters are removed from the environment
-        params = env.pop('parameter_defaults', None)
-
-        contents = yaml.safe_dump(env, default_flow_style=False)
-
-        # Until we have a well defined plan update workflow in tripleo-common
-        # we need to manually add an environment in swift and for users
-        # custom environments passed to the deploy command.
-        # See bug: https://bugs.launchpad.net/tripleo/+bug/1623431
-        # Update plan env.
-        swift_path = "user-environment.yaml"
-        self.object_client.put_object(container_name, swift_path, contents)
-
-        env = yaml.safe_load(self.object_client.get_object(
-            container_name, constants.PLAN_ENVIRONMENT)[1])
-
-        user_env = {'path': swift_path}
-        if user_env not in env['environments']:
-            env['environments'].append(user_env)
-            yaml_string = yaml.safe_dump(env, default_flow_style=False)
-            self.object_client.put_object(
-                container_name, constants.PLAN_ENVIRONMENT, yaml_string)
-
-        # Parameters are sent to the update parameters action, this stores them
-        # in the plan environment and means the UI can find them.
-        if params:
-            with utils.TempDirs() as tmp:
-                utils.run_ansible_playbook(
-                    playbook='cli-update-params.yaml',
-                    inventory='localhost,',
-                    workdir=tmp,
-                    playbook_dir=constants.ANSIBLE_TRIPLEO_PLAYBOOKS,
-                    verbosity=utils.playbook_verbosity(self=self),
-                    extra_vars={
-                        "container": container_name
-                    },
-                    extra_vars_file={
-                        "parameters": params
-                    }
-                )
-
-    def _upload_missing_files(self, container_name, files_dict, tht_root):
-        """Find the files referenced in custom environments and upload them
-
-        Heat environments can be passed to be included in the deployment, these
-        files can include references to other files anywhere on the local
-        file system. These need to be discovered and uploaded to Swift. When
-        they have been uploaded to Swift the path to them will be different,
-        the new paths are store din the file_relocation dict, which is returned
-        and used by _process_and_upload_environment which will merge the
-        environment and update paths to the relative Swift path.
-        """
-
-        file_relocation = {}
-        file_prefix = "file://"
-
-        # select files files for relocation & upload
-        for fullpath in files_dict.keys():
-
-            if not fullpath.startswith(file_prefix):
-                continue
-
-            path = fullpath[len(file_prefix):]
-
-            if path.startswith(tht_root):
-                # This should already be uploaded.
-                continue
-
-            file_relocation[fullpath] = "user-files/{}".format(
-                os.path.normpath(path[1:]))
-
-        # make sure links within files point to new locations, and upload them
-        for orig_path, reloc_path in file_relocation.items():
-            link_replacement = utils.relative_link_replacement(
-                file_relocation, os.path.dirname(reloc_path))
-            contents = utils.replace_links_in_template_contents(
-                files_dict[orig_path], link_replacement)
-            self.object_client.put_object(container_name, reloc_path, contents)
-
-        return file_relocation
+        deployment.deploy_without_plan(
+            self.clients, stack, stack_name,
+            template, files, env_files_tracker,
+            self.log)
 
     def _download_missing_files_from_plan(self, tht_dir, plan_name):
         # get and download missing files into tmp directory
@@ -486,27 +369,23 @@ class DeployOvercloud(command.Command):
             os.path.abspath(tht_root)))
 
         self.log.debug("Creating Environment files")
-        env = {}
         created_env_files = []
 
+        created_env_files.append(
+            os.path.join(tht_root, 'overcloud-resource-registry-puppet.yaml'))
         created_env_files.extend(
             self._provision_baremetal(parsed_args, tht_root))
 
         if parsed_args.environment_directories:
             created_env_files.extend(utils.load_environment_directories(
                 parsed_args.environment_directories))
+
         parameters = {}
-        if stack:
-            try:
-                # If user environment already exist then keep it
-                user_env = yaml.safe_load(self.object_client.get_object(
-                    parsed_args.stack, constants.USER_ENVIRONMENT)[1])
-                template_utils.deep_update(env, user_env)
-            except Exception:
-                pass
-        parameters.update(self._update_parameters(parsed_args, stack))
-        template_utils.deep_update(env, self._create_parameters_env(
-            parameters, tht_root, parsed_args.stack))
+        parameters.update(self._update_parameters(
+            parsed_args, stack, tht_root, user_tht_root))
+        param_env = self._create_parameters_env(
+            parameters, tht_root, parsed_args.stack)
+        created_env_files.extend(param_env)
 
         if parsed_args.deployed_server:
             created_env_files.append(
@@ -520,11 +399,34 @@ class DeployOvercloud(command.Command):
             deployment_options['ansible_python_interpreter'] = \
                 parsed_args.deployment_python_interpreter
 
+        if stack:
+            env_path = self._create_breakpoint_cleanup_env(
+                tht_root, parsed_args.stack)
+            created_env_files.extend(env_path)
+
         self.log.debug("Processing environment files %s" % created_env_files)
-        env_files, localenv = utils.process_multiple_environments(
+        env_files_tracker = []
+        env_files, env = utils.process_multiple_environments(
             created_env_files, tht_root, user_tht_root,
+            env_files_tracker=env_files_tracker,
             cleanup=(not parsed_args.no_cleanup))
-        template_utils.deep_update(env, localenv)
+
+        # Invokes the workflows specified in plan environment file
+        if parsed_args.plan_environment_file:
+            output_path = self._user_env_path(
+                'derived_parameters.yaml', tht_root)
+            workflow_params.build_derived_params_environment(
+                self.clients, parsed_args.stack, tht_root, env_files,
+                env_files_tracker, parsed_args.roles_file,
+                parsed_args.plan_environment_file,
+                output_path, utils.playbook_verbosity(self=self))
+
+            created_env_files.append(output_path)
+            env_files_tracker = []
+            env_files, env = utils.process_multiple_environments(
+                created_env_files, tht_root, user_tht_root,
+                env_files_tracker=env_files_tracker,
+                cleanup=(not parsed_args.no_cleanup))
 
         if parsed_args.limit:
             # check if skip list is defined while using --limit and throw a
@@ -545,10 +447,6 @@ class DeployOvercloud(command.Command):
                 if (ceph_deployed != "OS::Heat::None"
                         or ceph_external != "OS::Heat::None"):
                     utils.check_ceph_fsid_matches_env_files(stack, env)
-            bp_cleanup = self._create_breakpoint_cleanup_env(
-                tht_root, parsed_args.stack)
-            template_utils.deep_update(env, bp_cleanup)
-
         # check migration to new nic config with ansible
         utils.check_nic_config_with_ansible(stack, env)
 
@@ -563,10 +461,11 @@ class DeployOvercloud(command.Command):
                     '(with HA).')
 
         self._try_overcloud_deploy_with_compat_yaml(
-            tht_root, stack, parsed_args.stack, parameters, env_files,
+            tht_root, stack,
+            parsed_args.stack, parameters, env_files,
             parsed_args.timeout, env,
-            parsed_args.run_validations, parsed_args.skip_deploy_identifier,
-            parsed_args.plan_environment_file,
+            parsed_args.run_validations,
+            env_files_tracker=env_files_tracker,
             deployment_options=deployment_options)
 
         self._unprovision_baremetal(parsed_args)
@@ -575,16 +474,15 @@ class DeployOvercloud(command.Command):
                                                stack_name, parameters,
                                                env_files, timeout,
                                                env, run_validations,
-                                               skip_deploy_identifier,
-                                               plan_env_file,
+                                               env_files_tracker=None,
                                                deployment_options=None):
         overcloud_yaml = os.path.join(tht_root, constants.OVERCLOUD_YAML_NAME)
         try:
             self._heat_deploy(stack, stack_name, overcloud_yaml,
                               parameters, env_files, timeout,
                               tht_root, env,
-                              run_validations, skip_deploy_identifier,
-                              plan_env_file,
+                              run_validations,
+                              env_files_tracker=env_files_tracker,
                               deployment_options=deployment_options)
         except Exception as e:
             messages = 'Failed to deploy: %s' % str(e)
diff --git a/tripleoclient/workflows/deployment.py b/tripleoclient/workflows/deployment.py
index c02e39dde..3f8abb67c 100644
--- a/tripleoclient/workflows/deployment.py
+++ b/tripleoclient/workflows/deployment.py
@@ -131,14 +131,68 @@ def create_overcloudrc(stack, rc_params, no_proxy='',
         rc_params['password'],
         rc_params['region'])
     rcpath = os.path.join(output_dir, '%src' % stack.stack_name)
-
     with open(rcpath, 'w') as rcfile:
         rcfile.write(overcloudrcs['overcloudrc'])
     os.chmod(rcpath, 0o600)
-
     return os.path.abspath(rcpath)
 
 
+def deploy_without_plan(clients, stack, stack_name, template,
+                        files, env_files,
+                        log):
+    orchestration_client = clients.orchestration
+    if stack is None:
+        log.info("Performing Heat stack create")
+        action = 'CREATE'
+        marker = None
+    else:
+        log.info("Performing Heat stack update")
+        # Make sure existing parameters for stack are reused
+        # Find the last top-level event to use for the first marker
+        events = event_utils.get_events(orchestration_client,
+                                        stack_id=stack_name,
+                                        event_args={'sort_dir': 'desc',
+                                                    'limit': 1})
+        marker = events[0].id if events else None
+        action = 'UPDATE'
+
+    set_deployment_status(clients=clients,
+                          plan=stack_name,
+                          status='DEPLOYING')
+    stack_args = {
+        'stack_name': stack_name,
+        'template': template,
+        'environment_files': env_files,
+        'files': files}
+    try:
+        if stack:
+            stack_args['existing'] = True
+            orchestration_client.stacks.update(stack.id, **stack_args)
+        else:
+            stack = orchestration_client.stacks.create(**stack_args)
+
+        print("Success.")
+    except Exception:
+        set_deployment_status(clients=clients,
+                              plan=stack_name,
+                              status='DEPLOY_FAILED')
+        raise
+
+    create_result = utils.wait_for_stack_ready(
+        orchestration_client, stack_name, marker, action)
+    if not create_result:
+        shell.OpenStackShell().run(["stack", "failures", "list", stack_name])
+        set_deployment_status(
+            clients=clients,
+            plan=stack_name,
+            status='DEPLOY_FAILED'
+        )
+        if stack is None:
+            raise exceptions.DeploymentError("Heat Stack create failed.")
+        else:
+            raise exceptions.DeploymentError("Heat Stack update failed.")
+
+
 def get_overcloud_hosts(stack, ssh_network):
     ips = []
     role_net_ip_map = utils.get_role_net_ip_map(stack)
diff --git a/tripleoclient/workflows/parameters.py b/tripleoclient/workflows/parameters.py
index 1335dc97b..abde133b3 100644
--- a/tripleoclient/workflows/parameters.py
+++ b/tripleoclient/workflows/parameters.py
@@ -15,9 +15,11 @@ import os
 import re
 import yaml
 
+from heatclient.common import template_utils
 from tripleo_common.utils import stack_parameters as stk_parameters
 
 from tripleoclient.constants import ANSIBLE_TRIPLEO_PLAYBOOKS
+from tripleoclient.constants import OVERCLOUD_YAML_NAME
 from tripleoclient.constants import UNUSED_PARAMETER_EXCLUDES_RE
 from tripleoclient import exceptions
 from tripleoclient import utils
@@ -28,6 +30,8 @@ LOG = logging.getLogger(__name__)
 
 
 def invoke_plan_env_workflows(clients, stack_name, plan_env_file,
+                              stack_data, role_list,
+                              derived_environment_path,
                               verbosity=0):
     """Invokes the workflows in plan environment file"""
 
@@ -57,6 +61,9 @@ def invoke_plan_env_workflows(clients, stack_name, plan_env_file,
                     pb_vars
                 )
             )
+            pb_vars['tripleo_get_flatten_params'] = {'stack_data': stack_data}
+            pb_vars['tripleo_role_list'] = role_list
+            pb_vars['derived_environment_path'] = derived_environment_path
             playbook_dir = os.path.dirname(pb)
             if not playbook_dir:
                 playbook_dir = ANSIBLE_TRIPLEO_PLAYBOOKS
@@ -71,6 +78,39 @@ def invoke_plan_env_workflows(clients, stack_name, plan_env_file,
             )
 
 
+def build_derived_params_environment(clients, stack_name,
+                                     tht_root, env_files,
+                                     env_files_tracker,
+                                     roles_file,
+                                     plan_env_file,
+                                     derived_env_file,
+                                     verbosity):
+    template_path = os.path.join(tht_root, OVERCLOUD_YAML_NAME)
+    template_files, template = template_utils.get_template_contents(
+        template_file=template_path)
+    files = dict(list(template_files.items()) + list(
+        env_files.items()))
+    # Build stack_data
+    stack_data = utils.build_stack_data(
+        clients, stack_name, template,
+        files, env_files_tracker)
+
+    # Get role list
+    role_list = roles.get_roles(
+        clients, roles_file, template, files,
+        env_files_tracker, detail=False, valid=True)
+
+    invoke_plan_env_workflows(
+            clients,
+            stack_name,
+            plan_env_file,
+            stack_data=stack_data,
+            role_list=role_list,
+            derived_environment_path=derived_env_file,
+            verbosity=verbosity
+        )
+
+
 def check_deprecated_parameters(clients, container):
     """Checks for deprecated parameters in plan and adds warning if present.
 
diff --git a/tripleoclient/workflows/roles.py b/tripleoclient/workflows/roles.py
index 9f55f0d65..fc5a4f594 100644
--- a/tripleoclient/workflows/roles.py
+++ b/tripleoclient/workflows/roles.py
@@ -18,10 +18,50 @@ from tripleo_common.actions import plan
 # TODO(cloudnull): Convert to a swiftutils in tripleo-common
 # from tripleo_common.utils import swift as swiftutils
 
+from tripleoclient import utils
 
 LOG = logging.getLogger(__name__)
 
 
+def get_roles_data(roles_file, tht_root):
+    abs_roles_file = utils.get_roles_file_path(
+        roles_file, tht_root)
+    roles_data = None
+    with open(abs_roles_file, 'r') as fp:
+        roles_data = yaml.safe_load(fp)
+    return roles_data
+
+
+def get_roles(clients, roles_file, tht_root,
+              stack_name,
+              template,
+              files,
+              env_files,
+              detail=False, valid=False):
+    roles_data = get_roles_data(roles_file, tht_root)
+
+    if detail:
+        return roles_data
+
+    role_names = [role['name'] for role in roles_data]
+
+    if not valid:
+        return role_names
+
+    stack_data = utils.build_stack_data(
+        clients, stack_name, template,
+        files, env_files)
+
+    valid_roles = []
+    for name in role_names:
+        role_count = stack_data['parameters'].get(
+            name + 'Count', {}).get('default', 0)
+        if role_count > 0:
+            valid_roles.append(name)
+
+    return valid_roles
+
+
 def list_available_roles(clients, container='overcloud'):
     """Return a list of available roles.