Terraform Infra-Driver Change VNF Package

This patch provides Change current VNF package API and
Rollback of its operation for the Terraform infra-driver.

In this API for Terraform infra-driver, overwrite old tf files
with new tf files in new vnf package
(Rollback is the opposite of this).
Therefore, there are no limitation in tf file.

This patch also includes tiny refactoring of
tacker/sol_refactored/infra_drivers/terraform/terraform.py
as follows:

- Changing the implementation regarding Terraform commands
  because the options given differ depending on the terraform
  command or LCM operation.
- Changing _exec_cmd to use the built-in error handling option
  of the subprocess module [1]
- Changing the error handling to the same strategy as the helm
  infra-driver [2]

[1] https://docs.python.org/ja/3/library/subprocess.html
[2] https://github.com/openstack/tacker/blob/master/tacker/sol_refactored/infra_drivers/kubernetes/helm_utils.py#L33C1-L41C22

Implements: blueprint terraform-infra-driver-ccvp

Change-Id: I00d883f879d9e4a0d72d4e78cb4d0071649fb5ad

releasenotes/notes/add-terraform-infra-driver-change-vnfpkg-087413812aca9a1f.yaml

@@ -0,0 +1,7 @@
+---
+features:
+  - |
+    Add Change Current VNF Package API for Terraform infra-driver.
+    Using this API, Tacker updates virtual resources deployed by
+    Terraform infra-driver instantiation operations. Currently,
+    it only supports "RollingUpdate" out of several methods for a update.

tacker/sol_refactored/common/config.py

@@ -132,7 +132,7 @@ VNFM_OPTS = [
                help=_('Specifies the root CA certificate to use when the '
                       'heat_verify_cert option is True.')),
     cfg.StrOpt('tf_file_dir',
-             default='/var/lib/tacker/terraform/',
+             default='/var/lib/tacker/terraform',
              help=_('Temporary directory for Terraform infra-driver to '
                     'store terraform config files'))
 ]

tacker/sol_refactored/conductor/vnflcm_driver_v2.py

@@ -1178,6 +1178,9 @@ class VnfLcmDriverV2(object):
         elif vim_info.vimType == 'ETSINFV.HELM.V_3':
             driver = helm.Helm()
             driver.change_vnfpkg(req, inst, grant_req, grant, vnfd)
+        elif vim_info.vimType == 'TERRAFORM.V1':
+            driver = terraform.Terraform()
+            driver.change_vnfpkg(req, inst, grant_req, grant, vnfd)
         else:
             # should not occur
             raise sol_ex.SolException(sol_detail='not support vim type')
@@ -1204,6 +1207,9 @@ class VnfLcmDriverV2(object):
         elif vim_info.vimType == 'ETSINFV.HELM.V_3':
             driver = helm.Helm()
             driver.change_vnfpkg_rollback(req, inst, grant_req, grant, vnfd)
+        elif vim_info.vimType == 'TERRAFORM.V1':
+            driver = terraform.Terraform()
+            driver.change_vnfpkg_rollback(req, inst, grant_req, grant, vnfd)
         else:
             # should not occur
             raise sol_ex.SolException(sol_detail='not support vim type')

tacker/sol_refactored/infra_drivers/terraform/terraform.py

@@ -21,7 +21,7 @@ from tacker.sol_refactored.common import config
 from tacker.sol_refactored.common import exceptions as sol_ex
 from tacker.sol_refactored.common import vnf_instance_utils as inst_utils
 from tacker.sol_refactored import objects
-
+from tacker.sol_refactored.objects.v2 import fields as v2fields
 
 import json
 import os
@@ -53,28 +53,24 @@ class Terraform():
         self._instantiate(vim_conn_info, working_dir, tf_var_path)
         self._make_instantiated_vnf_info(req, inst, grant_req,
                                          grant, vnfd, working_dir,
-                                         tf_var_path)
+                                         tf_dir_path, tf_var_path)
 
     def _instantiate(self, vim_conn_info, working_dir, tf_var_path):
         '''Executes terraform init, terraform plan, and terraform apply'''
 
         access_info = vim_conn_info.get('accessInfo', {})
 
-        try:
-            init_cmd = self._gen_tf_cmd("init")
-            self._exec_cmd(init_cmd, cwd=working_dir)
-            LOG.info("Terraform init completed successfully.")
+        init_cmd = ['terraform', 'init']
+        self._exec_cmd(init_cmd, cwd=working_dir)
+        LOG.info("Terraform init completed successfully.")
 
-            plan_cmd = self._gen_tf_cmd('plan', access_info, tf_var_path)
-            self._exec_cmd(plan_cmd, cwd=working_dir)
-            LOG.info("Terraform plan completed successfully.")
+        plan_cmd = self._gen_plan_cmd(access_info, tf_var_path)
+        self._exec_cmd(plan_cmd, cwd=working_dir)
+        LOG.info("Terraform plan completed successfully.")
 
-            apply_cmd = self._gen_tf_cmd('apply', access_info, tf_var_path)
-            self._exec_cmd(apply_cmd, cwd=working_dir)
-            LOG.info("Terraform apply completed successfully.")
-
-        except subprocess.CalledProcessError as error:
-            raise sol_ex.TerraformOperationFailed(sol_detail=str(error))
+        apply_cmd = self._gen_apply_cmd(access_info, tf_var_path)
+        self._exec_cmd(apply_cmd, cwd=working_dir)
+        LOG.info("Terraform apply completed successfully.")
 
     def terminate(self, req, inst, grant_req, grant, vnfd):
         '''Terminates the terraform resources managed by the current project'''
@@ -89,17 +85,9 @@ class Terraform():
 
         access_info = vim_conn_info.get('accessInfo', {})
 
-        try:
-            # Execute the terraform destroy command (auto-approve)
-            destroy_cmd = self._gen_tf_cmd('destroy', access_info, tf_var_path)
-            self._exec_cmd(destroy_cmd, cwd=working_dir)
-            LOG.info("Terraform destroy completed successfully.")
-
-        except subprocess.CalledProcessError as error:
-            failed_process = error.cmd[0].capitalize()
-            LOG.error(f"Error running {failed_process}: {error}")
-            # raise error and leave working_dir for retry
-            raise sol_ex.TerraformOperationFailed(sol_detail=str(error))
+        destroy_cmd = self._gen_destroy_cmd(access_info, tf_var_path)
+        self._exec_cmd(destroy_cmd, cwd=working_dir)
+        LOG.info("Terraform destroy completed successfully.")
 
         try:
             # Remove the working directory and its contents
@@ -113,18 +101,113 @@ class Terraform():
         '''Calls terminate'''
         self.terminate(req, inst, grant_req, grant, vnfd)
 
-    def _make_instantiated_vnf_info(self, req, inst, grant_req,
-                                    grant, vnfd, working_dir, tf_var_path):
+    def change_vnfpkg(self, req, inst, grant_req, grant, vnfd):
+        '''Calls Terraform Apply and replicates new files'''
+
+        vim_conn_info = inst_utils.select_vim_info(inst.vimConnectionInfo)
+        tf_dir_path = req.additionalParams.get('tf_dir_path')
+        tf_var_path = req.additionalParams.get('tf_var_path')
+        working_dir = f"{CONF.v2_vnfm.tf_file_dir}/{inst.id}"
+
+        if req.additionalParams['upgrade_type'] == 'RollingUpdate':
+            self._change_vnfpkg_rolling_update(vim_conn_info, working_dir,
+                                               req.vnfdId, tf_dir_path,
+                                               tf_var_path)
+
+        self._make_instantiated_vnf_info(req, inst, grant_req,
+                                         grant, vnfd, working_dir,
+                                         tf_dir_path, tf_var_path)
+
+    def _change_vnfpkg_rolling_update(self, vim_conn_info, working_dir,
+                                      vnfd_id, tf_dir_path, tf_var_path):
+        '''Calls Terraform Apply'''
+
+        excluded_files = ['.terraform', '.terraform.lock.hcl',
+                  'terraform.tfstate', 'provider.tf', 'provider.tf.json']
+
+        # Delete old files (e.g main.tf, variables.tf, modules)
+        for root, dirs, files in os.walk(working_dir):
+            for file_name in files:
+                if file_name not in excluded_files:
+                    file_path = os.path.join(root, file_name)
+                    if os.path.exists(file_path):
+                        os.remove(file_path)
+            for dir_name in dirs:
+                dir_path = os.path.join(root, dir_name)
+                if os.path.exists(dir_path):
+                    shutil.rmtree(dir_path)
+
+        # Duplicate tf files from new VNF Package
+        context = tacker.context.get_admin_context()
+        try:
+            pkg_vnfd = vnf_package_vnfd.VnfPackageVnfd().get_by_id(
+                context, vnfd_id)
+        except exceptions.VnfPackageVnfdNotFound as exc:
+            raise sol_ex.VnfdIdNotFound(vnfd_id=vnfd_id) from exc
+        csar_path = os.path.join(CONF.vnf_package.vnf_package_csar_path,
+                                pkg_vnfd.package_uuid)
+        if tf_dir_path is not None:
+            vnf_package_path = f"{csar_path}/{tf_dir_path}"
+        else:
+            vnf_package_path = csar_path
+
+        # Copy files from new VNF Package
+        for file_name in os.listdir(vnf_package_path):
+            if file_name not in excluded_files:
+                source_path = os.path.join(vnf_package_path, file_name)
+                if os.path.exists(source_path):
+                    if os.path.isfile(source_path):
+                        shutil.copy2(source_path, working_dir)
+                    elif os.path.isdir(source_path):
+                        destination_dir = os.path.join(working_dir, file_name)
+                        shutil.copytree(source_path, destination_dir,
+                                        dirs_exist_ok=True)
+
+        access_info = vim_conn_info.get('accessInfo', {})
+
+        init_cmd = ['terraform', 'init', '-upgrade']
+        self._exec_cmd(init_cmd, cwd=working_dir)
+        LOG.info("Terraform init completed successfully.")
+
+        plan_cmd = self._gen_plan_cmd(access_info, tf_var_path)
+        self._exec_cmd(plan_cmd, cwd=working_dir)
+        LOG.info("Terraform plan completed successfully.")
+
+        apply_cmd = self._gen_apply_cmd(access_info, tf_var_path)
+        self._exec_cmd(apply_cmd, cwd=working_dir)
+        LOG.info("Terraform apply completed successfully.")
+
+    def change_vnfpkg_rollback(self, req, inst, grant_req, grant, vnfd):
+        '''Calls _change_vnfpkg_rolling_update function'''
+
+        vim_conn_info = inst_utils.select_vim_info(inst.vimConnectionInfo)
+        tf_dir_path = inst.instantiatedVnfInfo.metadata['tf_dir_path']
+        tf_var_path = inst.instantiatedVnfInfo.metadata['tf_var_path']
+        working_dir = f"{CONF.v2_vnfm.tf_file_dir}/{inst.id}"
+
+        if req.additionalParams['upgrade_type'] == 'RollingUpdate':
+            self._change_vnfpkg_rolling_update(vim_conn_info, working_dir,
+                                               inst.vnfdId, tf_dir_path,
+                                               tf_var_path)
+
+        self._make_instantiated_vnf_info(req, inst, grant_req,
+                                         grant, vnfd, working_dir,
+                                         tf_dir_path, tf_var_path)
+
+    def _make_instantiated_vnf_info(self, req, inst, grant_req, grant, vnfd,
+                                    working_dir, tf_dir_path, tf_var_path):
         '''Updates Tacker with information on the VNF state'''
 
-        # Define inst_vnf_info
-        flavour_id = req.flavourId
-        inst.instantiatedVnfInfo = objects.VnfInstanceV2_InstantiatedVnfInfo(
+        op = grant_req.operation
+        if op == v2fields.LcmOperationType.INSTANTIATE:
+            flavour_id = req.flavourId
+        else:
+            flavour_id = inst.instantiatedVnfInfo.flavourId
+
+        # make new instantiatedVnfInfo and replace
+        inst_vnf_info = objects.VnfInstanceV2_InstantiatedVnfInfo(
             flavourId=flavour_id,
             vnfState='STARTED',
-            metadata={
-                'tf_var_path': tf_var_path
-            }
         )
 
         # Specify the path to the terraform.tfstate file
@@ -167,8 +250,24 @@ class Terraform():
             for vnfc_res_info in vnfc_resource_info_list
         ]
 
-        inst.instantiatedVnfInfo.vnfcResourceInfo = vnfc_resource_info_list
-        inst.instantiatedVnfInfo.vnfcInfo = vnfc_info_list
+        inst_vnf_info.vnfcResourceInfo = vnfc_resource_info_list
+        inst_vnf_info.vnfcInfo = vnfc_info_list
+
+        inst_vnf_info.metadata = {}
+        # Restore metadata
+        if (inst.obj_attr_is_set('instantiatedVnfInfo') and
+                inst.instantiatedVnfInfo.obj_attr_is_set('metadata')):
+            inst_vnf_info.metadata.update(inst.instantiatedVnfInfo.metadata)
+
+        # Store tf_dir_path
+        if op == v2fields.LcmOperationType.INSTANTIATE or tf_dir_path:
+            inst_vnf_info.metadata['tf_dir_path'] = tf_dir_path
+
+        # Store tf_var_path
+        if op == v2fields.LcmOperationType.INSTANTIATE or tf_var_path:
+            inst_vnf_info.metadata['tf_var_path'] = tf_var_path
+
+        inst.instantiatedVnfInfo = inst_vnf_info
 
     def _get_tf_vnfpkg(self, vnf_instance_id, vnfd_id, tf_dir_path):
         """Create a VNF package with given IDs
@@ -239,6 +338,33 @@ class Terraform():
 
         return provider_tf_path
 
+    def _gen_cmd_args(self, access_info, tf_var_path):
+        args = []
+        for key, value in access_info.items():
+            if key == "endpoints":
+                continue
+            args.extend(['-var', f'{key}={value}'])
+        if tf_var_path:
+            args.extend(['-var-file', tf_var_path])
+        return args
+
+    def _gen_plan_cmd(self, access_info, tf_var_path, extra_args=None):
+        cmd = ['terraform', 'plan']
+        args = self._gen_cmd_args(access_info, tf_var_path)
+        if extra_args:
+            args.extend(extra_args)
+        return cmd + args
+
+    def _gen_apply_cmd(self, access_info, tf_var_path):
+        cmd = ['terraform', 'apply', '-auto-approve']
+        args = self._gen_cmd_args(access_info, tf_var_path)
+        return cmd + args
+
+    def _gen_destroy_cmd(self, access_info, tf_var_path):
+        cmd = ['terraform', 'destroy', '-auto-approve']
+        args = self._gen_cmd_args(access_info, tf_var_path)
+        return cmd + args
+
     def _exec_cmd(self, cmd, cwd,
                   stdout=subprocess.PIPE, stderr=subprocess.PIPE,
                   check=True, text=True):
@@ -248,42 +374,8 @@ class Terraform():
         commands used in this package. All the args other than self and cmd
         the same as for subprocess.run().
         """
-        res = subprocess.run(cmd, cwd=cwd, stdout=stdout, stderr=stderr,
-                             check=check, text=text)
-        if res.returncode != 0:
-            raise
-        return res
-
-    def _gen_tf_cmd(self, subcmd, access_info=None, tf_var_path=None,
-                    auto_approve=True):
-        """Return terraform command of given subcommand as a list
-
-        The result is intended to be an arg of supprocess.run().
-        """
-
-        # NOTE(yasufum): Only following subcommands are supported.
-        allowed_subcmds = ["init", "plan", "apply", "destroy"]
-        if subcmd not in allowed_subcmds:
-            return []
-
-        if subcmd == "init":
-            return ["terraform", "init"]
-
-        def _gen_tf_cmd_args(access_info, tf_var_path):
-            args = []
-            for key, value in access_info.items():
-                if key == "endpoints":
-                    continue
-                args.extend(['-var', f'{key}={value}'])
-            if tf_var_path:
-                args.extend(['-var-file', tf_var_path])
-            return args
-
-        # list of subcommands accept "-auto-approve" option.
-        accept_ap = ["apply", "destroy"]
-        if auto_approve is True and subcmd in accept_ap:
-            cmd = ["terraform", subcmd, "-auto-approve"]
-        else:
-            cmd = ["terraform", subcmd]
-        args = _gen_tf_cmd_args(access_info, tf_var_path)
-        return cmd + args
+        try:
+            subprocess.run(cmd, cwd=cwd, stdout=stdout, stderr=stderr,
+                           check=check, text=text)
+        except subprocess.CalledProcessError as error:
+            raise sol_ex.TerraformOperationFailed(sol_detail=str(error))

tacker/tests/functional/sol_terraform_v2/base_v2.py

@@ -12,25 +12,12 @@
 #    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
 #    License for the specific language governing permissions and limitations
 #    under the License.
-import os
-import shutil
-import tempfile
-import time
-
 from oslo_config import cfg
-from oslo_utils import uuidutils
 
-from tacker.sol_refactored.common import http_client
 from tacker.sol_refactored import objects
 from tacker.tests.functional import base_v2
-from tacker.tests.functional.common.fake_server import FakeServerManager
-from tacker.tests.functional.sol_v2_common import utils
 from tacker import version
 
-FAKE_SERVER_MANAGER = FakeServerManager()
-
-VNF_PACKAGE_UPLOAD_TIMEOUT = 300
-
 
 class BaseVnfLcmTerraformV2Test(base_v2.BaseTackerTestV2):
 
@@ -43,77 +30,3 @@ class BaseVnfLcmTerraformV2Test(base_v2.BaseTackerTestV2):
                  project='tacker',
                  version='%%prog %s' % version.version_info.release_string())
         objects.register_all()
-
-        vim_info = cls.get_vim_info()
-        auth = http_client.KeystonePasswordAuthHandle(
-            auth_url=vim_info.interfaceInfo['endpoint'],
-            username=vim_info.accessInfo['username'],
-            password=vim_info.accessInfo['password'],
-            project_name=vim_info.accessInfo['project'],
-            user_domain_name=vim_info.accessInfo['userDomain'],
-            project_domain_name=vim_info.accessInfo['projectDomain']
-        )
-        cls.tacker_client = http_client.HttpClient(auth)
-
-    def assert_notification_get(self, callback_url):
-        notify_mock_responses = FAKE_SERVER_MANAGER.get_history(
-            callback_url)
-        FAKE_SERVER_MANAGER.clear_history(
-            callback_url)
-        self.assertEqual(1, len(notify_mock_responses))
-        self.assertEqual(204, notify_mock_responses[0].status_code)
-
-    def _check_notification(self, callback_url, notify_type):
-        notify_mock_responses = FAKE_SERVER_MANAGER.get_history(
-            callback_url)
-        FAKE_SERVER_MANAGER.clear_history(
-            callback_url)
-        self.assertEqual(1, len(notify_mock_responses))
-        self.assertEqual(204, notify_mock_responses[0].status_code)
-        self.assertEqual(notify_type, notify_mock_responses[0].request_body[
-            'notificationType'])
-
-    @classmethod
-    def create_vnf_package(cls, sample_path, user_data=None, image_path=None):
-        vnfd_id = uuidutils.generate_uuid()
-        tmp_dir = tempfile.mkdtemp()
-
-        utils.make_zip(sample_path, tmp_dir, vnfd_id, image_path)
-
-        zip_file_name = os.path.basename(os.path.abspath(sample_path)) + ".zip"
-        zip_file_path = os.path.join(tmp_dir, zip_file_name)
-
-        path = "/vnfpkgm/v1/vnf_packages"
-        if user_data is not None:
-            req_body = {'userDefinedData': user_data}
-        else:
-            req_body = {}
-        resp, body = cls.tacker_client.do_request(
-            path, "POST", expected_status=[201], body=req_body)
-
-        pkg_id = body['id']
-
-        with open(zip_file_path, 'rb') as fp:
-            path = f"/vnfpkgm/v1/vnf_packages/{pkg_id}/package_content"
-            resp, body = cls.tacker_client.do_request(
-                path, "PUT", body=fp, content_type='application/zip',
-                expected_status=[202])
-
-        # wait for onboard
-        timeout = VNF_PACKAGE_UPLOAD_TIMEOUT
-        start_time = int(time.time())
-        path = f"/vnfpkgm/v1/vnf_packages/{pkg_id}"
-        while True:
-            resp, body = cls.tacker_client.do_request(
-                path, "GET", expected_status=[200])
-            if body['onboardingState'] == "ONBOARDED":
-                break
-
-            if (int(time.time()) - start_time) > timeout:
-                raise Exception("Failed to onboard vnf package")
-
-            time.sleep(5)
-
-        shutil.rmtree(tmp_dir)
-
-        return pkg_id, vnfd_id

tacker/tests/functional/sol_terraform_v2/paramgen.py

@@ -77,3 +77,30 @@ def terminate_req():
         "gracefulTerminationTimeout": 5,
         "additionalParams": {"dummy-key": "dummy-val"}
     }
+
+
+def change_vnfpkg_req(vnfd_id):
+    return {
+        "vnfdId": vnfd_id,
+        "additionalParams": {
+            "upgrade_type": "RollingUpdate",
+            "tf_dir_path": "Files/terraform",
+            "vdu_params": [{
+                "vdu_id": "VDU1"
+            }]
+        }
+    }
+
+
+def change_vnfpkg_fail_req(vnfd_id):
+    return {
+        "vnfdId": vnfd_id,
+        "additionalParams": {
+            "upgrade_type": "RollingUpdate",
+            "tf_dir_path": "Files/terraform",
+            "tf_var_path": "Files/terraform/test-tf-fail.tfvars",
+            "vdu_params": [{
+                "vdu_id": "VDU1"
+            }]
+        }
+    }

tacker/tests/functional/sol_terraform_v2/samples/test_terraform_change_vnf_package/contents/Definitions/sample_tf_df_simple.yaml

@@ -0,0 +1,177 @@
+tosca_definitions_version: tosca_simple_yaml_1_2
+
+description: Simple deployment flavour for Sample VNF
+
+imports:
+  - etsi_nfv_sol001_common_types.yaml
+  - etsi_nfv_sol001_vnfd_types.yaml
+  - sample_tf_types.yaml
+
+topology_template:
+  inputs:
+    descriptor_id:
+      type: string
+    descriptor_version:
+      type: string
+    provider:
+      type: string
+    product_name:
+      type: string
+    software_version:
+      type: string
+    vnfm_info:
+      type: list
+      entry_schema:
+        type: string
+    flavour_id:
+      type: string
+    flavour_description:
+      type: string
+
+  substitution_mappings:
+    node_type: company.provider.VNF
+    properties:
+      flavour_id: simple
+    requirements:
+      virtual_link_external: []
+
+  node_templates:
+    VNF:
+      type: company.provider.VNF
+      properties:
+        flavour_description: A simple flavour
+      interfaces:
+        Vnflcm:
+          instantiate_start:
+            implementation: sample-script
+          instantiate_end:
+            implementation: sample-script
+          terminate_start:
+            implementation: sample-script
+          terminate_end:
+            implementation: sample-script
+          scale_start:
+            implementation: sample-script
+          scale_end:
+            implementation: sample-script
+          heal_start:
+            implementation: sample-script
+          heal_end:
+            implementation: sample-script
+          modify_information_start:
+            implementation: sample-script
+          modify_information_end:
+            implementation: sample-script
+      artifacts:
+        sample-script:
+          description: Sample script
+          type: tosca.artifacts.Implementation.Python
+          file: ../Scripts/sample_script.py
+
+    VDU1:
+      type: tosca.nodes.nfv.Vdu.Compute
+      properties:
+        name: vdu1
+        description: VDU1 compute node
+        vdu_profile:
+          min_number_of_instances: 1
+          max_number_of_instances: 3
+
+    VDU2:
+      type: tosca.nodes.nfv.Vdu.Compute
+      properties:
+        name: vdu2
+        description: VDU2 compute node
+        vdu_profile:
+          min_number_of_instances: 1
+          max_number_of_instances: 3
+
+  policies:
+    - scaling_aspects:
+        type: tosca.policies.nfv.ScalingAspects
+        properties:
+          aspects:
+            vdu1_aspect:
+              name: vdu1_aspect
+              description: vdu1 scaling aspect
+              max_scale_level: 2
+              step_deltas:
+                - delta_1
+            vdu2_aspect:
+              name: vdu2_aspect
+              description: vdu2 scaling aspect
+              max_scale_level: 2
+              step_deltas:
+                - delta_1
+
+    - VDU1_initial_delta:
+        type: tosca.policies.nfv.VduInitialDelta
+        properties:
+          initial_delta:
+            number_of_instances: 1
+        targets: [ VDU1 ]
+
+    - VDU1_scaling_aspect_deltas:
+        type: tosca.policies.nfv.VduScalingAspectDeltas
+        properties:
+          aspect: vdu1_aspect
+          deltas:
+            delta_1:
+              number_of_instances: 1
+        targets: [ VDU1 ]
+
+    - VDU2_initial_delta:
+        type: tosca.policies.nfv.VduInitialDelta
+        properties:
+          initial_delta:
+            number_of_instances: 1
+        targets: [ VDU2 ]
+
+    - VDU2_scaling_aspect_deltas:
+        type: tosca.policies.nfv.VduScalingAspectDeltas
+        properties:
+          aspect: vdu2_aspect
+          deltas:
+            delta_1:
+              number_of_instances: 1
+        targets: [ VDU2 ]
+
+    - instantiation_levels:
+        type: tosca.policies.nfv.InstantiationLevels
+        properties:
+          levels:
+            instantiation_level_1:
+              description: Smallest size
+              scale_info:
+                vdu1_aspect:
+                  scale_level: 0
+                vdu2_aspect:
+                  scale_level: 0
+            instantiation_level_2:
+              description: Largest size
+              scale_info:
+                vdu1_aspect:
+                  scale_level: 2
+                vdu2_aspect:
+                  scale_level: 2
+          default_level: instantiation_level_1
+
+    - VDU1_instantiation_levels:
+        type: tosca.policies.nfv.VduInstantiationLevels
+        properties:
+          levels:
+            instantiation_level_1:
+              number_of_instances: 1
+            instantiation_level_2:
+              number_of_instances: 3
+        targets: [ VDU1 ]
+
+    - VDU2_instantiation_levels:
+        type: tosca.policies.nfv.VduInstantiationLevels
+        properties:
+          levels:
+            instantiation_level_1:
+              number_of_instances: 1
+            instantiation_level_2:
+              number_of_instances: 3
+        targets: [ VDU2 ]

tacker/tests/functional/sol_terraform_v2/samples/test_terraform_change_vnf_package/contents/Definitions/sample_tf_top.vnfd.yaml

@@ -0,0 +1,31 @@
+tosca_definitions_version: tosca_simple_yaml_1_2
+
+description: Sample Terraform VNF
+
+imports:
+  - etsi_nfv_sol001_common_types.yaml
+  - etsi_nfv_sol001_vnfd_types.yaml
+  - sample_tf_types.yaml
+  - sample_tf_df_simple.yaml
+
+topology_template:
+  inputs:
+    selected_flavour:
+      type: string
+      description: VNF deployment flavour selected by the consumer. It is provided in the API
+
+  node_templates:
+    VNF:
+      type: company.provider.VNF
+      properties:
+        flavour_id: { get_input: selected_flavour }
+        descriptor_id: b1bb0ce7-ebca-4fa7-95ed-4840d7000000
+        provider: Company
+        product_name: Sample Terraform VNF
+        software_version: '1.0'
+        descriptor_version: '1.0'
+        vnfm_info:
+          - Tacker
+      requirements:
+        #- virtual_link_external # mapped in lower-level templates
+        #- virtual_link_internal # mapped in lower-level templates

tacker/tests/functional/sol_terraform_v2/samples/test_terraform_change_vnf_package/contents/Definitions/sample_tf_types.yaml

@@ -0,0 +1,53 @@
+tosca_definitions_version: tosca_simple_yaml_1_2
+
+description: VNF type definition
+
+imports:
+  - etsi_nfv_sol001_common_types.yaml
+  - etsi_nfv_sol001_vnfd_types.yaml
+
+node_types:
+  company.provider.VNF:
+    derived_from: tosca.nodes.nfv.VNF
+    properties:
+      descriptor_id:
+        type: string
+        constraints: [ valid_values: [ b1bb0ce7-ebca-4fa7-95ed-4840d7000000 ] ]
+        default: b1bb0ce7-ebca-4fa7-95ed-4840d7000000
+      descriptor_version:
+        type: string
+        constraints: [ valid_values: [ '1.0' ] ]
+        default: '1.0'
+      provider:
+        type: string
+        constraints: [ valid_values: [ 'Company' ] ]
+        default: 'Company'
+      product_name:
+        type: string
+        constraints: [ valid_values: [ 'Sample Terraform VNF' ] ]
+        default: 'Sample Terraform VNF'
+      software_version:
+        type: string
+        constraints: [ valid_values: [ '1.0' ] ]
+        default: '1.0'
+      vnfm_info:
+        type: list
+        entry_schema:
+          type: string
+          constraints: [ valid_values: [ Tacker ] ]
+        default: [ Tacker ]
+      flavour_id:
+        type: string
+        constraints: [ valid_values: [ simple ] ]
+        default: simple
+      flavour_description:
+        type: string
+        default: ""
+    requirements:
+      - virtual_link_external:
+          capability: tosca.capabilities.nfv.VirtualLinkable
+      - virtual_link_internal:
+          capability: tosca.capabilities.nfv.VirtualLinkable
+    interfaces:
+      Vnflcm:
+        type: tosca.interfaces.nfv.Vnflcm

tacker/tests/functional/sol_terraform_v2/samples/test_terraform_change_vnf_package/contents/Files/terraform/main.tf

@@ -0,0 +1,8 @@
+resource "aws_instance" "vdu1"{
+  ami           = "ami-785db401"
+  instance_type = "t2.small"
+
+  tags = {
+    Name = "change-vnfpkg-test"
+  }
+}

tacker/tests/functional/sol_terraform_v2/samples/test_terraform_change_vnf_package/contents/Scripts/sample_script.py

@@ -0,0 +1,67 @@
+# Copyright (C) 2023 Nippon Telegraph and Telephone Corporation
+# All Rights Reserved.
+#
+#    Licensed under the Apache License, Version 2.0 (the "License"); you may
+#    not use this file except in compliance with the License. You may obtain
+#    a copy of the License at
+#
+#         http://www.apache.org/licenses/LICENSE-2.0
+#
+#    Unless required by applicable law or agreed to in writing, software
+#    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+#    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+#    License for the specific language governing permissions and limitations
+#    under the License.
+
+import os
+import pickle
+import sys
+
+
+class SampleScript(object):
+
+    def __init__(self, req, inst, grant_req, grant, csar_dir):
+        self.req = req
+        self.inst = inst
+        self.grant_req = grant_req
+        self.grant = grant
+        self.csar_dir = csar_dir
+
+    def instantiate_start(self):
+        pass
+
+    def instantiate_end(self):
+        pass
+
+    def terminate_start(self):
+        pass
+
+    def terminate_end(self):
+        pass
+
+
+def main():
+    script_dict = pickle.load(sys.stdin.buffer)
+
+    operation = script_dict['operation']
+    req = script_dict['request']
+    inst = script_dict['vnf_instance']
+    grant_req = script_dict['grant_request']
+    grant = script_dict['grant_response']
+    csar_dir = script_dict['tmp_csar_dir']
+
+    script = SampleScript(req, inst, grant_req, grant, csar_dir)
+    try:
+        getattr(script, operation)()
+    except AttributeError:
+        raise Exception("{} is not included in the script.".format(operation))
+
+
+if __name__ == "__main__":
+    try:
+        main()
+        os._exit(0)
+    except Exception as ex:
+        sys.stderr.write(str(ex))
+        sys.stderr.flush()
+        os._exit(1)

tacker/tests/functional/sol_terraform_v2/samples/test_terraform_change_vnf_package/contents/TOSCA-Metadata/TOSCA.meta

@@ -0,0 +1,9 @@
+TOSCA-Meta-File-Version: 1.0
+Created-by: dummy_user
+CSAR-Version: 1.1
+Entry-Definitions: Definitions/sample_tf_top.vnfd.yaml
+
+Name: Files/terraform/main.tf
+Content-Type: test-data
+Algorithm: SHA-256
+Hash: c438b48a8bfb577746eff0c3e3d284a55a211430fc6ab80120b58750b0dd8bea

tacker/tests/functional/sol_terraform_v2/samples/test_terraform_change_vnf_package/pkggen.py

@@ -0,0 +1,44 @@
+# Copyright (C) 2023 Nippon Telegraph and Telephone Corporation
+# All Rights Reserved.
+#
+#    Licensed under the Apache License, Version 2.0 (the "License"); you may
+#    not use this file except in compliance with the License. You may obtain
+#    a copy of the License at
+#
+#         http://www.apache.org/licenses/LICENSE-2.0
+#
+#    Unless required by applicable law or agreed to in writing, software
+#    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+#    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+#    License for the specific language governing permissions and limitations
+#    under the License.
+
+import json
+import os
+import shutil
+import tempfile
+
+from oslo_utils import uuidutils
+
+from tacker.tests.functional.sol_terraform_v2 import paramgen as tf_paramgen
+from tacker.tests.functional.sol_v2_common import utils
+
+
+zip_file_name = os.path.basename(os.path.abspath(".")) + '.zip'
+tmp_dir = tempfile.mkdtemp()
+vnfd_id = uuidutils.generate_uuid()
+
+# tacker/tests/functional/sol_terraform_v2/samples/{package_name}
+utils.make_zip(".", tmp_dir, vnfd_id)
+
+shutil.move(os.path.join(tmp_dir, zip_file_name), ".")
+shutil.rmtree(tmp_dir)
+
+change_vnfpkg_req = tf_paramgen.change_vnfpkg_req(vnfd_id)
+change_vnfpkg_fail_req = tf_paramgen.change_vnfpkg_fail_req(vnfd_id)
+
+with open("change_vnfpkg_req", "w", encoding='utf-8') as f:
+    f.write(json.dumps(change_vnfpkg_req, indent=2))
+
+with open("change_vnfpkg_fail_req", "w", encoding='utf-8') as f:
+    f.write(json.dumps(change_vnfpkg_fail_req, indent=2))

tacker/tests/functional/sol_terraform_v2/test_terraform.py

@@ -14,6 +14,7 @@
 #    under the License.
 
 import os
+import time
 
 import tacker.conf
 
@@ -23,6 +24,8 @@ from tacker.tests.functional.sol_terraform_v2 import paramgen as tf_paramgen
 
 CONF = tacker.conf.CONF
 
+WAIT_LCMOCC_UPDATE_TIME = 3
+
 
 class VnfLcmTerraformTest(base_v2.BaseVnfLcmTerraformV2Test):
 
@@ -35,25 +38,21 @@ class VnfLcmTerraformTest(base_v2.BaseVnfLcmTerraformV2Test):
         pkg_dir_path = os.path.join(cur_dir, sample_pkg)
         cls.basic_pkg, cls.basic_vnfd_id = cls.create_vnf_package(pkg_dir_path)
 
+        sample_chg_vnfpkg = "samples/test_terraform_change_vnf_package"
+        chg_vnfpkg_dir_path = os.path.join(cur_dir, sample_chg_vnfpkg)
+        cls.new_pkg, cls.new_vnfd_id = cls.create_vnf_package(
+            chg_vnfpkg_dir_path)
+
     @classmethod
     def tearDownClass(cls):
         super(VnfLcmTerraformTest, cls).tearDownClass()
 
         cls.delete_vnf_package(cls.basic_pkg)
+        cls.delete_vnf_package(cls.new_pkg)
 
     def setUp(self):
         super(VnfLcmTerraformTest, self).setUp()
 
-    def instantiate_vnf_instance(self, inst_id, req_body):
-        path = "/vnflcm/v2/vnf_instances/{}/instantiate".format(inst_id)
-        return self.tacker_client.do_request(
-            path, "POST", body=req_body, version="2.0.0")
-
-    def terminate_vnf_instance(self, inst_id, req_body):
-        path = "/vnflcm/v2/vnf_instances/{}/terminate".format(inst_id)
-        return self.tacker_client.do_request(
-            path, "POST", body=req_body, version="2.0.0")
-
     def test_basic_lcms(self):
         self._get_basic_lcms_procedure()
 
@@ -65,8 +64,10 @@ class VnfLcmTerraformTest(base_v2.BaseVnfLcmTerraformV2Test):
           - 1. Create VNF instance
           - 2. Instantiate VNF
           - 3. Show VNF instance
-          - 4. Terminate VNF
-          - 5. Delete a VNF instance
+          - 4. Change Current VNF Package
+          - 5. Show VNF instance
+          - 6. Terminate VNF
+          - 7. Delete a VNF instance
         """
 
         # 1. Create a new VNF instance resource
@@ -117,6 +118,11 @@ class VnfLcmTerraformTest(base_v2.BaseVnfLcmTerraformV2Test):
         self.check_resp_headers_in_get(resp)
         self.check_resp_body(body, expected_inst_attrs)
 
+        vnfc_resource_infos = body['instantiatedVnfInfo']['vnfcResourceInfo']
+        before_resource_ids = {vnfc_info['computeResource']['resourceId']
+                               for vnfc_info in vnfc_resource_infos}
+        self.assertEqual(1, len(before_resource_ids))
+
         # check instantiationState of VNF
         self.assertEqual(fields.VnfInstanceState.INSTANTIATED,
                          body['instantiationState'])
@@ -125,7 +131,47 @@ class VnfLcmTerraformTest(base_v2.BaseVnfLcmTerraformV2Test):
         self.assertEqual(fields.VnfOperationalStateType.STARTED,
                          body['instantiatedVnfInfo']['vnfState'])
 
-        # 4. Terminate VNF instance
+        # 4. Change Current VNF Package
+        change_vnfpkg_req = tf_paramgen.change_vnfpkg_req(self.new_vnfd_id)
+        resp, body = self.change_vnfpkg(inst_id, change_vnfpkg_req)
+        self.assertEqual(202, resp.status_code)
+        self.check_resp_headers_in_operation_task(resp)
+
+        lcmocc_id = os.path.basename(resp.headers['Location'])
+        self.wait_lcmocc_complete(lcmocc_id)
+
+        # wait a bit because there is a bit time lag between lcmocc DB
+        # update and change_vnfpkg completion.
+        time.sleep(WAIT_LCMOCC_UPDATE_TIME)
+
+        # check usageState of VNF Package
+        self.check_package_usage(self.basic_pkg, 'NOT_IN_USE')
+
+        # check usageState of VNF Package
+        self.check_package_usage(self.new_pkg, 'IN_USE')
+
+        # 5. Show VNF instance
+        additional_inst_attrs = [
+            'vimConnectionInfo',
+            'instantiatedVnfInfo'
+        ]
+        expected_inst_attrs.extend(additional_inst_attrs)
+        resp, body = self.show_vnf_instance(inst_id)
+        self.assertEqual(200, resp.status_code)
+        self.check_resp_headers_in_get(resp)
+        self.check_resp_body(body, expected_inst_attrs)
+
+        vnfc_resource_infos = body['instantiatedVnfInfo']['vnfcResourceInfo']
+        after_resource_ids = {vnfc_info['computeResource']['resourceId']
+                              for vnfc_info in vnfc_resource_infos}
+        self.assertEqual(1, len(after_resource_ids))
+        # In other infraDriver, computeResource.resourceId is
+        # "assertNotEqual" before and after ChangeCurrentVnfPkg.
+        # However, the current Terraform InfraDriver specification
+        # sets the same value.
+        self.assertEqual(before_resource_ids, after_resource_ids)
+
+        # 6. Terminate VNF instance
         terminate_req = tf_paramgen.terminate_req()
         resp, body = self.terminate_vnf_instance(inst_id, terminate_req)
         self.assertEqual(202, resp.status_code)
@@ -140,7 +186,144 @@ class VnfLcmTerraformTest(base_v2.BaseVnfLcmTerraformV2Test):
         self.assertEqual(fields.VnfInstanceState.NOT_INSTANTIATED,
                          body['instantiationState'])
 
-        # 5. Delete a VNF instance
+        # 7. Delete a VNF instance
+        resp, body = self.delete_vnf_instance(inst_id)
+        self.assertEqual(204, resp.status_code)
+        self.check_resp_headers_in_delete(resp)
+
+        # check deletion of VNF instance
+        resp, body = self.show_vnf_instance(inst_id)
+        self.assertEqual(404, resp.status_code)
+        self.check_package_usage(self.basic_pkg, state='NOT_IN_USE')
+
+    def test_change_vnfpkg_rollback(self):
+        """Test basic LCM operations
+
+        * About LCM operations:
+          This test includes the following operations.
+          - 1. Create VNF instance
+          - 2. Instantiate VNF
+          - 3. Show VNF instance
+          - 4. Change Current VNF Package => FAILED_TEMP
+          - 5. Rollback Change Current VNF Package
+          - 6. Show VNF instance
+          - 7. Terminate VNF
+          - 8. Delete a VNF instance
+        """
+
+        # 1. Create a new VNF instance resource
+        # NOTE: extensions and vnfConfigurableProperties are omitted
+        # because they are commented out in etsi_nfv_sol001.
+        expected_inst_attrs = [
+            'id',
+            'vnfInstanceName',
+            'vnfInstanceDescription',
+            'vnfdId',
+            'vnfProvider',
+            'vnfProductName',
+            'vnfSoftwareVersion',
+            'vnfdVersion',
+            # 'vnfConfigurableProperties', # omitted
+            # 'vimConnectionInfo', # omitted
+            'instantiationState',
+            # 'instantiatedVnfInfo', # omitted
+            'metadata',
+            # 'extensions', # omitted
+            '_links'
+        ]
+        create_req = tf_paramgen.create_req_by_vnfd_id(self.basic_vnfd_id)
+        resp, body = self.create_vnf_instance(create_req)
+        self.assertEqual(201, resp.status_code)
+        self.check_resp_headers_in_create(resp)
+        self.check_resp_body(body, expected_inst_attrs)
+        inst_id = body['id']
+        self.check_package_usage(self.basic_pkg, state='IN_USE')
+
+        # 2. Instantiate VNF
+        instantiate_req = tf_paramgen.instantiate_req()
+        resp, body = self.instantiate_vnf_instance(inst_id, instantiate_req)
+        self.assertEqual(202, resp.status_code)
+        self.check_resp_headers_in_operation_task(resp)
+
+        lcmocc_id = os.path.basename(resp.headers['Location'])
+        self.wait_lcmocc_complete(lcmocc_id)
+
+        # 3. Show VNF instance
+        additional_inst_attrs = [
+            'vimConnectionInfo',
+            'instantiatedVnfInfo'
+        ]
+        expected_inst_attrs.extend(additional_inst_attrs)
+        resp, body = self.show_vnf_instance(inst_id)
+        self.assertEqual(200, resp.status_code)
+        self.check_resp_headers_in_get(resp)
+        self.check_resp_body(body, expected_inst_attrs)
+
+        vnfc_resource_infos = body['instantiatedVnfInfo']['vnfcResourceInfo']
+        before_resource_ids = {vnfc_info['computeResource']['resourceId']
+                               for vnfc_info in vnfc_resource_infos}
+        self.assertEqual(1, len(before_resource_ids))
+
+        # check instantiationState of VNF
+        self.assertEqual(fields.VnfInstanceState.INSTANTIATED,
+                         body['instantiationState'])
+
+        # check vnfState of VNF
+        self.assertEqual(fields.VnfOperationalStateType.STARTED,
+                         body['instantiatedVnfInfo']['vnfState'])
+
+        # 4. Change Current VNF Package => FAILED_TEMP
+        change_vnfpkg_req = (
+            tf_paramgen.change_vnfpkg_fail_req(self.new_vnfd_id))
+        resp, body = self.change_vnfpkg(inst_id, change_vnfpkg_req)
+        self.assertEqual(202, resp.status_code)
+        self.check_resp_headers_in_operation_task(resp)
+
+        lcmocc_id = os.path.basename(resp.headers['Location'])
+        self.wait_lcmocc_failed_temp(lcmocc_id)
+
+        # 5. Rollback Change Current VNF Package
+        resp, body = self.rollback_lcmocc(lcmocc_id)
+        self.assertEqual(202, resp.status_code)
+        self.check_resp_headers_in_delete(resp)
+        self.wait_lcmocc_rolled_back(lcmocc_id)
+
+        # check usageState of VNF Package
+        self.check_package_usage(self.new_pkg, 'NOT_IN_USE')
+
+        # 6. Show VNF instance
+        additional_inst_attrs = [
+            'vimConnectionInfo',
+            'instantiatedVnfInfo'
+        ]
+        expected_inst_attrs.extend(additional_inst_attrs)
+        resp, body = self.show_vnf_instance(inst_id)
+        self.assertEqual(200, resp.status_code)
+        self.check_resp_headers_in_get(resp)
+        self.check_resp_body(body, expected_inst_attrs)
+
+        vnfc_resource_infos = body['instantiatedVnfInfo']['vnfcResourceInfo']
+        after_resource_ids = {vnfc_info['computeResource']['resourceId']
+                              for vnfc_info in vnfc_resource_infos}
+        self.assertEqual(1, len(after_resource_ids))
+        self.assertEqual(before_resource_ids, after_resource_ids)
+
+        # 7. Terminate VNF instance
+        terminate_req = tf_paramgen.terminate_req()
+        resp, body = self.terminate_vnf_instance(inst_id, terminate_req)
+        self.assertEqual(202, resp.status_code)
+        self.check_resp_headers_in_operation_task(resp)
+
+        lcmocc_id = os.path.basename(resp.headers['Location'])
+        self.wait_lcmocc_complete(lcmocc_id)
+
+        # check instantiationState of VNF
+        resp, body = self.show_vnf_instance(inst_id)
+        self.assertEqual(200, resp.status_code)
+        self.assertEqual(fields.VnfInstanceState.NOT_INSTANTIATED,
+                         body['instantiationState'])
+
+        # 8. Delete a VNF instance
         resp, body = self.delete_vnf_instance(inst_id)
         self.assertEqual(204, resp.status_code)
         self.check_resp_headers_in_delete(resp)

tacker/tests/unit/sol_refactored/infra_drivers/terraform/test_terraform.py

@@ -32,7 +32,7 @@ SAMPLE_FLAVOUR_ID = "simple"
 
 _vim_connection_info_example = {
     "vimId": "terraform_provider_aws_v4_tokyo",
-    "vimType": "ETSINFV.TERRAFORM.V_1",
+    "vimType": "TERRAFORM.V1",
     "interfaceInfo": {
         "providerType": "aws",
         "providerVersion": "4.0"
@@ -56,6 +56,17 @@ _instantiate_req_example = {
     }
 }
 
+# ChangeCurrentVnfPkgRequest example
+_change_vnfpkg_req_example = {
+    "vnfdId": SAMPLE_VNFD_ID,
+    "additionalParams": {
+        "upgrade_type": "RollingUpdate",
+        "vdu_params": [{
+            "vdu_id": "VDU1"
+        }]
+    }
+}
+
 
 class TestTerraform(base.BaseTestCase):
     def setUp(self):
@@ -114,10 +125,195 @@ class TestTerraform(base.BaseTestCase):
             f"/var/lib/tacker/terraform/{inst.id}",
             req.additionalParams.get('tf_var_path'))
 
+    @mock.patch.object(terraform.Terraform, '_terminate')
+    def test_terminate(self, mock_terminate):
+        '''Verifies terminate is called once'''
+
+        req_inst = objects.InstantiateVnfRequest.from_dict(
+            _instantiate_req_example)
+
+        req = objects.TerminateVnfRequest(
+            terminationType='GRACEFUL')
+
+        inst = objects.VnfInstanceV2(
+            # Required fields
+            id=uuidutils.generate_uuid(),
+            vnfdId=SAMPLE_VNFD_ID,
+            vnfProvider='provider',
+            vnfProductName='product name',
+            vnfSoftwareVersion='software version',
+            vnfdVersion='vnfd version',
+            instantiationState='INSTANTIATED',
+            vimConnectionInfo=req_inst.vimConnectionInfo,
+            instantiatedVnfInfo=objects.VnfInstanceV2_InstantiatedVnfInfo(
+                metadata={
+                    "tf_var_path": "None"
+                }
+            )
+        )
+
+        grant_req = objects.GrantRequestV1(
+            operation=fields.LcmOperationType.TERMINATE
+        )
+
+        grant = objects.GrantV1()
+
+        # Execute
+        self.driver.terminate(req, inst, grant_req, grant, self.vnfd_2)
+        # Verify _instantiate is called once
+        mock_terminate.assert_called_once_with(
+            req_inst.vimConnectionInfo["vim1"],
+            f"/var/lib/tacker/terraform/{inst.id}",
+            inst.instantiatedVnfInfo.metadata['tf_var_path'])
+
+    @mock.patch.object(terraform.Terraform, '_terminate')
+    def test_instantiate_rollback(self, mock_instantiate_rollback):
+        '''Verifies instantiate_rollback is called once'''
+
+        req = objects.InstantiateVnfRequest.from_dict(_instantiate_req_example)
+
+        inst = objects.VnfInstanceV2(
+            # Required fields
+            id=uuidutils.generate_uuid(),
+            vnfdId=SAMPLE_VNFD_ID,
+            vnfProvider='provider',
+            vnfProductName='product name',
+            vnfSoftwareVersion='software version',
+            vnfdVersion='vnfd version',
+            instantiationState='INSTANTIATED',
+            vimConnectionInfo=req.vimConnectionInfo,
+            instantiatedVnfInfo=objects.VnfInstanceV2_InstantiatedVnfInfo(
+                metadata={
+                    "tf_var_path": "None"
+                }
+            )
+        )
+
+        grant_req = objects.GrantRequestV1(
+            operation=fields.LcmOperationType.INSTANTIATE
+        )
+
+        grant = objects.GrantV1()
+
+        # Execute
+        self.driver.instantiate_rollback(req, inst, grant_req,
+                                         grant, self.vnfd_2)
+        # Verify _terminate is called once
+        mock_instantiate_rollback.assert_called_once_with(
+            req.vimConnectionInfo["vim1"],
+            f"/var/lib/tacker/terraform/{inst.id}",
+            inst.instantiatedVnfInfo.metadata['tf_var_path'])
+
+    @mock.patch.object(terraform.Terraform, '_get_tf_vnfpkg')
+    @mock.patch.object(terraform.Terraform, '_generate_provider_tf')
+    @mock.patch.object(terraform.Terraform, '_make_instantiated_vnf_info')
+    @mock.patch.object(terraform.Terraform, '_change_vnfpkg_rolling_update')
+    def test_change_vnfpkg(self, mock_change_vnfpkg,
+                           mock_make_instantiated_vnf_info,
+                           mock_generate_provider_tf,
+                           mock_tf_files):
+        '''Verifies change_vnfpkg is called once'''
+
+        req_inst = objects.InstantiateVnfRequest.from_dict(
+            _instantiate_req_example)
+
+        req = objects.ChangeCurrentVnfPkgRequest.from_dict(
+            _change_vnfpkg_req_example)
+
+        inst = objects.VnfInstanceV2(
+            # Required fields
+            id=uuidutils.generate_uuid(),
+            vnfdId=SAMPLE_VNFD_ID,
+            vnfProvider='provider',
+            vnfProductName='product name',
+            vnfSoftwareVersion='software version',
+            vnfdVersion='vnfd version',
+            instantiationState='INSTANTIATED',
+            vimConnectionInfo=req_inst.vimConnectionInfo
+        )
+
+        grant_req = objects.GrantRequestV1(
+            operation=fields.LcmOperationType.INSTANTIATE,
+            vnfdId=SAMPLE_VNFD_ID
+        )
+
+        grant = objects.GrantV1()
+
+        # Set the desired return value for _get_tf_vnfpkg
+        mock_tf_files.return_value = f"/var/lib/tacker/terraform/{inst.id}"
+
+        # Execute
+        self.driver.change_vnfpkg(req, inst, grant_req,
+                                  grant, self.vnfd_2)
+
+        mock_change_vnfpkg.assert_called_once_with(
+            req_inst.vimConnectionInfo["vim1"],
+            f"/var/lib/tacker/terraform/{inst.id}",
+            inst.vnfdId,
+            req.additionalParams.get('tf_dir_path'),
+            req.additionalParams.get('tf_var_path'))
+
+    @mock.patch.object(terraform.Terraform, '_get_tf_vnfpkg')
+    @mock.patch.object(terraform.Terraform, '_generate_provider_tf')
+    @mock.patch.object(terraform.Terraform, '_make_instantiated_vnf_info')
+    @mock.patch.object(terraform.Terraform, '_change_vnfpkg_rolling_update')
+    def test_change_vnfpkg_rollback(self, mock_change_vnfpkg,
+                           mock_make_instantiated_vnf_info,
+                           mock_generate_provider_tf,
+                           mock_tf_files):
+        '''Verifies change_vnfpkg_rollback is called once'''
+
+        req_inst = objects.InstantiateVnfRequest.from_dict(
+            _instantiate_req_example)
+
+        req = objects.ChangeCurrentVnfPkgRequest.from_dict(
+            _change_vnfpkg_req_example)
+
+        inst = objects.VnfInstanceV2(
+            # Required fields
+            id=uuidutils.generate_uuid(),
+            vnfdId=SAMPLE_VNFD_ID,
+            vnfProvider='provider',
+            vnfProductName='product name',
+            vnfSoftwareVersion='software version',
+            vnfdVersion='vnfd version',
+            instantiationState='INSTANTIATED',
+            vimConnectionInfo=req_inst.vimConnectionInfo,
+            instantiatedVnfInfo=objects.VnfInstanceV2_InstantiatedVnfInfo(
+                metadata={
+                    "tf_dir_path": "Files/terraform",
+                    "tf_var_path": "Files/terraform/variables.tf"
+                }
+            )
+        )
+
+        grant_req = objects.GrantRequestV1(
+            operation=fields.LcmOperationType.INSTANTIATE,
+            vnfdId=SAMPLE_VNFD_ID
+        )
+
+        grant = objects.GrantV1()
+
+        # Set the desired return value for _get_tf_vnfpkg
+        mock_tf_files.return_value = f"/var/lib/tacker/terraform/{inst.id}"
+
+        # Execute
+        self.driver.change_vnfpkg_rollback(req, inst, grant_req,
+                                           grant, self.vnfd_2)
+
+        mock_change_vnfpkg.assert_called_once_with(
+            req_inst.vimConnectionInfo["vim1"],
+            f"/var/lib/tacker/terraform/{inst.id}",
+            inst.vnfdId,
+            inst.instantiatedVnfInfo.metadata['tf_dir_path'],
+            inst.instantiatedVnfInfo.metadata['tf_var_path'])
+
     def test_make_instantiated_vnf_info(self):
         '''Verifies instantiated vnf info is correct'''
 
         req = objects.InstantiateVnfRequest.from_dict(_instantiate_req_example)
+        tf_dir_path = req.additionalParams.get('tf_dir_path')
+        tf_var_path = req.additionalParams.get('tf_var_path')
 
         inst = objects.VnfInstanceV2(
             # Required fields
@@ -174,7 +370,11 @@ class TestTerraform(base.BaseTestCase):
                     "vnfcResourceInfoId": "vdu2",
                     "vnfcState": "STARTED"
                 }
-            ]
+            ],
+            "metadata": {
+                "tf_dir_path": "Files/terraform",
+                "tf_var_path": "Files/terraform/variables.tf"
+            }
         }
 
         # Create a temporary directory
@@ -215,12 +415,10 @@ class TestTerraform(base.BaseTestCase):
                 json.dump(tfstate_content, tfstate_file)
 
             # Execute the test with the temporary tfstate_file
-            self.driver._make_instantiated_vnf_info(req, inst,
-                                                    grant_req, grant,
-                                                    self.vnfd_2, temp_dir,
-                                                    req.additionalParams.get(
-                                                        'tf_var_path')
-                                                    )
+            self.driver._make_instantiated_vnf_info(req, inst, grant_req,
+                                                    grant, self.vnfd_2,
+                                                    temp_dir, tf_dir_path,
+                                                    tf_var_path)
 
             # check
             result = inst.to_dict()["instantiatedVnfInfo"]
@@ -234,3 +432,7 @@ class TestTerraform(base.BaseTestCase):
             # order of vnfcInfo is same as vnfcResourceInfo
             self.assertIn("vnfcInfo", result)
             self.assertEqual(expected["vnfcInfo"], result["vnfcInfo"])
+
+            # check instantiatedVnfInfo.metadata
+            self.assertIn("metadata", result)
+            self.assertEqual(expected["metadata"], result["metadata"])