# Copyright (C) 2020 NTT DATA
# 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 copy
from io import StringIO
import os
import sys
from unittest import mock

import ddt
from oslo_utils.fixture import uuidsentinel

from tackerclient import client as root_client
from tackerclient.common import exceptions
from tackerclient.osc import utils as tacker_osc_utils
from tackerclient.osc.v1.vnflcm import vnflcm
from tackerclient.tests.unit.osc import base
from tackerclient.tests.unit.osc.v1.fixture_data import client
from tackerclient.tests.unit.osc.v1 import vnflcm_fakes
from tackerclient.tests.unit.test_cli10 import MyResp
from tackerclient.v1_0 import client as proxy_client


class TestVnfLcm(base.FixturedTestCase):
    client_fixture_class = client.ClientFixture

    def setUp(self):
        super(TestVnfLcm, self).setUp()
        self.url = client.TACKER_URL
        self.header = {'content-type': 'application/json'}
        self.app = mock.Mock()
        self.app_args = mock.Mock()
        self.client_manager = self.cs
        self.app.client_manager.tackerclient = self.client_manager


def _get_columns_vnflcm(action='create'):
    columns = ['ID', 'Instantiation State', 'VNF Instance Description',
               'VNF Instance Name', 'VNF Product Name', 'VNF Provider',
               'VNF Software Version', 'VNFD ID', 'VNFD Version', 'Links',
               'VNF Configurable Properties', 'VNF Package ID']
    if action == 'show':
        columns.extend(['Instantiated Vnf Info', 'VIM Connection Info'])
    if action == 'list':
        columns = [ele for ele in columns if ele not in
                   ['VNFD Version', 'VNF Instance Description',
                    'VNF Configurable Properties', 'VNF Package ID']]
        columns.remove('Links')
    return columns


@ddt.ddt
class TestCreateVnfLcm(TestVnfLcm):

    def setUp(self):
        super(TestCreateVnfLcm, self).setUp()
        self.create_vnf_lcm = vnflcm.CreateVnfLcm(
            self.app, self.app_args, cmd_name='vnflcm create')

    def test_create_no_args(self):
        self.assertRaises(base.ParserException, self.check_parser,
                          self.create_vnf_lcm, [], [])

    @ddt.data({"optional_arguments": True, "instantiate": True},
              {"optional_arguments": False, "instantiate": False})
    @ddt.unpack
    def test_take_action(self, optional_arguments, instantiate):
        arglist = [uuidsentinel.vnf_package_vnfd_id]
        verifylist = [('vnfd_id', uuidsentinel.vnf_package_vnfd_id)]

        if optional_arguments:
            arglist.extend(['--name', 'test',
                            '--description', 'test'])
            verifylist.extend([('name', 'test'),
                               ('description', 'test')])

        # command param
        if instantiate:
            param_file = ("./tackerclient/osc/v1/vnflcm/samples/"
                          "instantiate_vnf_instance_param_sample.json")

            arglist.extend(['--I', param_file])
            verifylist.append(('I', param_file))

        parsed_args = self.check_parser(self.create_vnf_lcm, arglist,
                                        verifylist)

        json = vnflcm_fakes.vnf_instance_response()
        self.requests_mock.register_uri(
            'POST', os.path.join(self.url, 'vnflcm/v1/vnf_instances'),
            json=json, headers=self.header)

        if instantiate:
            self.requests_mock.register_uri(
                'POST', os.path.join(self.url, 'vnflcm/v1/vnf_instances',
                                     json['id'], 'instantiate'),
                json={}, headers=self.header)

        sys.stdout = buffer = StringIO()
        actual_columns, data = (self.create_vnf_lcm.take_action(parsed_args))

        headers, attributes = vnflcm._get_columns(json)

        expected_message = (
            'VNF Instance ' + json['id'] + ' is created and instantiation '
                                           'request has been accepted.')
        if instantiate:
            self.assertEqual(expected_message, buffer.getvalue().strip())

        self.assertCountEqual(_get_columns_vnflcm(),
                              actual_columns)
        self.assertListItemsEqual(vnflcm_fakes.get_vnflcm_data(
            json, columns=attributes), data)


class TestShowVnfLcm(TestVnfLcm):

    def setUp(self):
        super(TestShowVnfLcm, self).setUp()
        self.show_vnf_lcm = vnflcm.ShowVnfLcm(
            self.app, self.app_args, cmd_name='vnflcm show')

    def test_take_action(self):
        vnf_instance = vnflcm_fakes.vnf_instance_response(
            instantiation_state='INSTANTIATED')

        arglist = [vnf_instance['id']]
        verifylist = [('vnf_instance', vnf_instance['id'])]

        # command param
        parsed_args = self.check_parser(self.show_vnf_lcm, arglist,
                                        verifylist)

        self.requests_mock.register_uri(
            'GET', os.path.join(self.url, 'vnflcm/v1/vnf_instances',
                                vnf_instance['id']),
            json=vnf_instance, headers=self.header)

        columns, data = (self.show_vnf_lcm.take_action(parsed_args))
        self.assertCountEqual(_get_columns_vnflcm(action='show'),
                              columns)
        headers, attributes = vnflcm._get_columns(vnf_instance, action='show')
        self.assertListItemsEqual(
            vnflcm_fakes.get_vnflcm_data(vnf_instance, columns=attributes),
            data)


class TestListVnfLcm(TestVnfLcm):

    def setUp(self):
        super(TestListVnfLcm, self).setUp()
        self.list_vnf_instance = vnflcm.ListVnfLcm(
            self.app, self.app_args, cmd_name='vnflcm list')

    def test_take_action(self):
        vnf_instances = vnflcm_fakes.create_vnf_instances(count=3)
        parsed_args = self.check_parser(self.list_vnf_instance, [], [])
        self.requests_mock.register_uri(
            'GET', os.path.join(self.url, 'vnflcm/v1/vnf_instances'),
            json=vnf_instances, headers=self.header)
        actual_columns, data = self.list_vnf_instance.take_action(parsed_args)

        headers, columns = tacker_osc_utils.get_column_definitions(
            vnflcm._attr_map, long_listing=True)

        expected_data = []
        for vnf_instance_obj in vnf_instances:
            expected_data.append(vnflcm_fakes.get_vnflcm_data(
                vnf_instance_obj, columns=columns, list_action=True))

        self.assertCountEqual(_get_columns_vnflcm(action='list'),
                              actual_columns)
        self.assertCountEqual(expected_data, list(data))

    def test_take_action_with_pagination(self):
        vnf_instances = vnflcm_fakes.create_vnf_instances(count=3)
        next_links_num = 3
        parsed_args = self.check_parser(self.list_vnf_instance, [], [])
        path = os.path.join(self.url, 'vnflcm/v1/vnf_instances')

        links = [0] * next_links_num
        link_headers = [0] * next_links_num

        for i in range(next_links_num):
            links[i] = (
                '{base_url}?nextpage_opaque_marker={vnf_instance_id}'.format(
                    base_url=path,
                    vnf_instance_id=vnf_instances[i]['id']))
            link_headers[i] = copy.deepcopy(self.header)
            link_headers[i]['Link'] = '<{link_url}>; rel="next"'.format(
                link_url=links[i])

        self.requests_mock.register_uri(
            'GET', path, json=[vnf_instances[0]], headers=link_headers[0])
        self.requests_mock.register_uri(
            'GET', links[0], json=[vnf_instances[1]],
            headers=link_headers[1])
        self.requests_mock.register_uri(
            'GET', links[1], json=[vnf_instances[2]],
            headers=link_headers[2])
        self.requests_mock.register_uri(
            'GET', links[2], json=[], headers=self.header)

        actual_columns, data = self.list_vnf_instance.take_action(parsed_args)

        headers, columns = tacker_osc_utils.get_column_definitions(
            vnflcm._attr_map, long_listing=True)

        expected_data = []
        for vnf_instance_obj in vnf_instances:
            expected_data.append(vnflcm_fakes.get_vnflcm_data(
                vnf_instance_obj, columns=columns, list_action=True))

        self.assertCountEqual(_get_columns_vnflcm(action='list'),
                              actual_columns)
        self.assertCountEqual(expected_data, list(data))


class TestInstantiateVnfLcm(TestVnfLcm):

    def setUp(self):
        super(TestInstantiateVnfLcm, self).setUp()
        self.instantiate_vnf_lcm = vnflcm.InstantiateVnfLcm(
            self.app, self.app_args, cmd_name='vnflcm instantiate')

    def test_take_action(self):
        vnf_instance = vnflcm_fakes.vnf_instance_response()
        sample_param_file = ("./tackerclient/osc/v1/vnflcm/samples/"
                             "instantiate_vnf_instance_param_sample.json")

        arglist = [vnf_instance['id'], sample_param_file]
        verifylist = [('vnf_instance', vnf_instance['id']),
                      ('instantiation_request_file', sample_param_file)]

        # command param
        parsed_args = self.check_parser(self.instantiate_vnf_lcm, arglist,
                                        verifylist)

        url = os.path.join(self.url, 'vnflcm/v1/vnf_instances',
                           vnf_instance['id'], 'instantiate')
        self.requests_mock.register_uri(
            'POST', url, headers=self.header, json={})

        sys.stdout = buffer = StringIO()
        with mock.patch.object(proxy_client.ClientBase,
                               '_handle_fault_response') as m:
            self.instantiate_vnf_lcm.take_action(parsed_args)
            # check no fault response is received
            self.assertNotCalled(m)
            self.assertEqual(
                'Instantiate request for VNF Instance ' + vnf_instance['id'] +
                ' has been accepted.', buffer.getvalue().strip())

    def test_take_action_vnf_instance_not_found(self):
        vnf_instance = vnflcm_fakes.vnf_instance_response()
        sample_param_file = ("./tackerclient/osc/v1/vnflcm/samples/"
                             "instantiate_vnf_instance_param_sample.json")
        arglist = [vnf_instance['id'], sample_param_file]
        verifylist = [('vnf_instance', vnf_instance['id']),
                      ('instantiation_request_file', sample_param_file)]

        # command param
        parsed_args = self.check_parser(self.instantiate_vnf_lcm, arglist,
                                        verifylist)

        url = os.path.join(self.url, 'vnflcm/v1/vnf_instances',
                           vnf_instance['id'], 'instantiate')
        self.requests_mock.register_uri(
            'POST', url, headers=self.header, status_code=404, json={})

        self.assertRaises(exceptions.TackerClientException,
                          self.instantiate_vnf_lcm.take_action,
                          parsed_args)

    def test_take_action_param_file_not_exists(self):
        vnf_instance = vnflcm_fakes.vnf_instance_response()
        sample_param_file = "./not_exists.json"
        arglist = [vnf_instance['id'], sample_param_file]
        verifylist = [('vnf_instance', vnf_instance['id']),
                      ('instantiation_request_file', sample_param_file)]

        # command param
        parsed_args = self.check_parser(self.instantiate_vnf_lcm, arglist,
                                        verifylist)

        ex = self.assertRaises(exceptions.InvalidInput,
                               self.instantiate_vnf_lcm.take_action,
                               parsed_args)

        expected_msg = ("Invalid input: "
                        "User does not have read privileges to it")
        self.assertEqual(expected_msg, str(ex))

    @mock.patch("os.open")
    @mock.patch("os.access")
    def test_take_action_invalid_format_param_file(self, mock_open,
                                                   mock_access):
        vnf_instance = vnflcm_fakes.vnf_instance_response()
        sample_param_file = "./invalid_param_file.json"
        arglist = [vnf_instance['id'], sample_param_file]
        verifylist = [('vnf_instance', vnf_instance['id']),
                      ('instantiation_request_file', sample_param_file)]

        mock_open.return_value = "invalid_json_data"
        # command param
        parsed_args = self.check_parser(self.instantiate_vnf_lcm, arglist,
                                        verifylist)

        ex = self.assertRaises(exceptions.InvalidInput,
                               self.instantiate_vnf_lcm.take_action,
                               parsed_args)
        expected_msg = "Failed to load parameter file."
        self.assertIn(expected_msg, str(ex))


@ddt.ddt
class TestHealVnfLcm(TestVnfLcm):

    def setUp(self):
        super(TestHealVnfLcm, self).setUp()
        self.heal_vnf_lcm = vnflcm.HealVnfLcm(
            self.app, self.app_args, cmd_name='vnflcm heal')

    _heal_sample_param_file = ("./tackerclient/osc/v1/vnflcm/samples/"
                               "heal_vnf_instance_param_sample.json")

    @ddt.data((['--cause', 'test-cause', "--vnfc-instance",
                'vnfc-id-1', 'vnfc-id-2'],
               [('cause', 'test-cause'),
                ('vnfc_instance', ['vnfc-id-1', 'vnfc-id-2'])]),
              (['--cause', 'test-cause'],
               [('cause', 'test-cause')]),
              (["--vnfc-instance", 'vnfc-id-1', 'vnfc-id-2'],
               [('vnfc_instance', ['vnfc-id-1', 'vnfc-id-2'])]),
              (["--additional-param-file", _heal_sample_param_file],
               [('additional_param_file', _heal_sample_param_file)]),
              ([], []))
    @ddt.unpack
    def test_take_action(self, arglist, verifylist):
        vnf_instance = vnflcm_fakes.vnf_instance_response()
        arglist.insert(0, vnf_instance['id'])
        verifylist.extend([('vnf_instance', vnf_instance['id'])])

        # command param
        parsed_args = self.check_parser(self.heal_vnf_lcm, arglist,
                                        verifylist)

        url = os.path.join(self.url, 'vnflcm/v1/vnf_instances',
                           vnf_instance['id'], 'heal')
        self.requests_mock.register_uri(
            'POST', url, headers=self.header, json={})

        sys.stdout = buffer = StringIO()
        result_error = self.heal_vnf_lcm.take_action(parsed_args)

        self.assertIsNone(result_error)
        actual_message = buffer.getvalue().strip()

        expected_message = ("Heal request for VNF Instance %s has been "
                            "accepted.") % vnf_instance['id']
        self.assertIn(expected_message, actual_message)

    def test_take_action_vnf_instance_not_found(self):
        vnf_instance = vnflcm_fakes.vnf_instance_response()
        arglist = [vnf_instance['id']]
        verifylist = [('vnf_instance', vnf_instance['id'])]

        # command param
        parsed_args = self.check_parser(self.heal_vnf_lcm, arglist,
                                        verifylist)

        url = os.path.join(self.url, 'vnflcm/v1/vnf_instances',
                           vnf_instance['id'], 'heal')
        self.requests_mock.register_uri(
            'POST', url, headers=self.header, status_code=404, json={})

        self.assertRaises(exceptions.TackerClientException,
                          self.heal_vnf_lcm.take_action,
                          parsed_args)

    def test_take_action_param_file_not_exists(self):
        vnf_instance = vnflcm_fakes.vnf_instance_response()
        sample_param_file = "./not_exists.json"
        arglist = [vnf_instance['id'],
                   '--additional-param-file', sample_param_file]
        verifylist = [('vnf_instance', vnf_instance['id']),
                      ('additional_param_file', sample_param_file)]

        # command param
        parsed_args = self.check_parser(self.heal_vnf_lcm, arglist,
                                        verifylist)

        ex = self.assertRaises(exceptions.InvalidInput,
                               self.heal_vnf_lcm.take_action, parsed_args)

        expected_msg = ("Invalid input: "
                        "User does not have read privileges to it")
        self.assertEqual(expected_msg, str(ex))


@ddt.ddt
class TestTerminateVnfLcm(TestVnfLcm):

    def setUp(self):
        super(TestTerminateVnfLcm, self).setUp()
        self.terminate_vnf_instance = vnflcm.TerminateVnfLcm(
            self.app, self.app_args, cmd_name='vnflcm terminate')

    @ddt.data({'termination_type': 'GRACEFUL', 'delete_vnf': True},
              {'termination_type': 'FORCEFUL', 'delete_vnf': False})
    @ddt.unpack
    def test_take_action(self, termination_type, delete_vnf):
        # argument 'delete_vnf' decides deletion of vnf instance post
        # termination.
        vnf_instance = vnflcm_fakes.vnf_instance_response()
        arglist = ['--termination-type', termination_type, vnf_instance['id']]

        verifylist = [('termination_type', termination_type),
                      ('vnf_instance', vnf_instance['id'])]

        if delete_vnf:
            arglist.extend(['--D'])
            verifylist.extend([('D', True)])

        if termination_type == 'GRACEFUL':
            arglist.extend(['--graceful-termination-timeout', '60'])
            verifylist.append(('graceful_termination_timeout', 60))
        parsed_args = self.check_parser(self.terminate_vnf_instance, arglist,
                                        verifylist)
        url = os.path.join(self.url, 'vnflcm/v1/vnf_instances',
                           vnf_instance['id'], 'terminate')

        with mock.patch.object(proxy_client.ClientBase,
                               '_handle_fault_response') as m:
            self.requests_mock.register_uri('POST', url, json={},
                                            headers=self.header)
            if delete_vnf:
                self.requests_mock.register_uri(
                    'GET', os.path.join(self.url, 'vnflcm/v1/vnf_instances',
                                        vnf_instance['id']),
                    json=vnf_instance, headers=self.header)
                self.requests_mock.register_uri(
                    'DELETE', os.path.join(
                        self.url, 'vnflcm/v1/vnf_instances',
                        vnf_instance['id']), json={}, headers=self.header)

            sys.stdout = buffer = StringIO()
            result = self.terminate_vnf_instance.take_action(parsed_args)
            actual_message = buffer.getvalue().strip()

            expected_message = ("Terminate request for VNF Instance '%s'"
                                " has been accepted.") % vnf_instance['id']
            self.assertIn(expected_message, actual_message)

            if delete_vnf:
                expected_message = ("VNF Instance '%s' is deleted successfully"
                                    % vnf_instance['id'])
                self.assertIn(expected_message, actual_message)

        self.assertIsNone(result)
        self.assertNotCalled(m)

    def test_take_action_terminate_and_delete_wait_failed(self):
        vnf_instance = vnflcm_fakes.vnf_instance_response()
        termination_type = 'GRACEFUL'
        arglist = ['--termination-type', termination_type, '--D',
                   '--graceful-termination-timeout', '5', vnf_instance['id']]

        verifylist = [('termination_type', termination_type), ('D', True),
                      ('graceful_termination_timeout', 5),
                      ('vnf_instance', vnf_instance['id'])]

        parsed_args = self.check_parser(self.terminate_vnf_instance, arglist,
                                        verifylist)
        url = os.path.join(self.url, 'vnflcm/v1/vnf_instances',
                           vnf_instance['id'], 'terminate')

        self.requests_mock.register_uri('POST', url, json={},
                                        headers=self.header)
        # set the instantiateState to "INSTANTIATED", so that the
        # _wait_until_vnf_is_terminated will fail
        vnf_instance['instantiationState'] = 'INSTANTIATED'

        self.requests_mock.register_uri(
            'GET', os.path.join(self.url, 'vnflcm/v1/vnf_instances',
                                vnf_instance['id']),
            json=vnf_instance, headers=self.header)

        sys.stdout = buffer = StringIO()
        with mock.patch.object(self.app.client_manager.tackerclient,
                               'delete_vnf_instance') as mock_delete:
            result = self.assertRaises(
                exceptions.CommandError,
                self.terminate_vnf_instance.take_action, parsed_args)

            actual_message = buffer.getvalue().strip()

            # Terminate vnf instance verification
            expected_message = ("Terminate request for VNF Instance '%s'"
                                " has been accepted.") % vnf_instance['id']
            self.assertIn(expected_message, actual_message)

            # Verify it fails to wait for termination before delete
            expected_message = ("Couldn't verify vnf instance is terminated "
                                "within '%(timeout)s' seconds. Unable to "
                                "delete vnf instance %(id)s"
                                % {'timeout': 15, 'id': vnf_instance['id']})

            self.assertIn(expected_message, str(result))
            self.assertNotCalled(mock_delete)

    def test_terminate_no_options(self):
        self.assertRaises(base.ParserException, self.check_parser,
                          self.terminate_vnf_instance, [], [])

    def test_take_action_vnf_instance_not_found(self):
        vnf_instance = vnflcm_fakes.vnf_instance_response()
        termination_type = 'GRACEFUL'
        arglist = ['--termination-type', termination_type, '--D',
                   '--graceful-termination-timeout', '5', vnf_instance['id']]

        verifylist = [('termination_type', termination_type), ('D', True),
                      ('graceful_termination_timeout', 5),
                      ('vnf_instance', vnf_instance['id'])]

        parsed_args = self.check_parser(self.terminate_vnf_instance, arglist,
                                        verifylist)

        url = os.path.join(self.url, 'vnflcm/v1/vnf_instances',
                           vnf_instance['id'], 'terminate')
        self.requests_mock.register_uri('POST', url, headers=self.header,
                                        status_code=404, json={})

        self.assertRaises(exceptions.TackerClientException,
                          self.terminate_vnf_instance.take_action,
                          parsed_args)


class TestDeleteVnfLcm(TestVnfLcm):

    def setUp(self):
        super(TestDeleteVnfLcm, self).setUp()
        self.delete_vnf_instance = vnflcm.DeleteVnfLcm(
            self.app, self.app_args, cmd_name='vnflcm delete')

        # Vnf Instance to delete
        self.vnf_instances = vnflcm_fakes.create_vnf_instances(count=3)

    def _mock_request_url_for_delete(self, vnf_index):
        url = os.path.join(self.url, 'vnflcm/v1/vnf_instances',
                           self.vnf_instances[vnf_index]['id'])

        json = self.vnf_instances[vnf_index]

        self.requests_mock.register_uri('GET', url, json=json,
                                        headers=self.header)
        self.requests_mock.register_uri('DELETE', url,
                                        headers=self.header, json={})

    def test_delete_one_vnf_instance(self):
        arglist = [self.vnf_instances[0]['id']]
        verifylist = [('vnf_instances',
                       [self.vnf_instances[0]['id']])]

        parsed_args = self.check_parser(self.delete_vnf_instance, arglist,
                                        verifylist)

        self._mock_request_url_for_delete(0)
        sys.stdout = buffer = StringIO()
        result = self.delete_vnf_instance.take_action(parsed_args)
        self.assertIsNone(result)
        self.assertEqual(("Vnf instance '%s' is deleted successfully")
                         % self.vnf_instances[0]['id'],
                         buffer.getvalue().strip())

    def test_delete_multiple_vnf_instance(self):
        arglist = []
        for vnf_pkg in self.vnf_instances:
            arglist.append(vnf_pkg['id'])
        verifylist = [('vnf_instances', arglist)]
        parsed_args = self.check_parser(self.delete_vnf_instance, arglist,
                                        verifylist)
        for i in range(0, 3):
            self._mock_request_url_for_delete(i)
        sys.stdout = buffer = StringIO()
        result = self.delete_vnf_instance.take_action(parsed_args)
        self.assertIsNone(result)
        self.assertEqual('All specified vnf instances are deleted '
                         'successfully', buffer.getvalue().strip())

    def test_delete_multiple_vnf_instance_exception(self):
        arglist = [
            self.vnf_instances[0]['id'],
            'xxxx-yyyy-zzzz',
            self.vnf_instances[1]['id'],
        ]
        verifylist = [('vnf_instances', arglist)]
        parsed_args = self.check_parser(self.delete_vnf_instance,
                                        arglist, verifylist)

        self._mock_request_url_for_delete(0)

        url = os.path.join(self.url, 'vnflcm/v1/vnf_instances',
                           'xxxx-yyyy-zzzz')
        self.requests_mock.register_uri(
            'GET', url, exc=exceptions.ConnectionFailed)

        self._mock_request_url_for_delete(1)
        exception = self.assertRaises(exceptions.CommandError,
                                      self.delete_vnf_instance.take_action,
                                      parsed_args)

        self.assertEqual('Failed to delete 1 of 3 vnf instances.',
                         exception.message)


class TestUpdateVnfLcm(TestVnfLcm):
    def setUp(self):
        super(TestUpdateVnfLcm, self).setUp()
        self.update_vnf_lcm = vnflcm.UpdateVnfLcm(
            self.app, self.app_args, cmd_name='vnflcm modify')

    def test_take_action(self):
        vnf_instance = vnflcm_fakes.vnf_instance_response()
        sample_param_file = ("./tackerclient/osc/v1/vnflcm/samples/"
                             "update_vnf_instance_param_sample.json")

        arglist = [vnf_instance['id'], '--I', sample_param_file]
        verifylist = [('vnf_instance', vnf_instance['id']),
                      ('I', sample_param_file)]

        # command param
        parsed_args = self.check_parser(
            self.update_vnf_lcm, arglist, verifylist)
        url = os.path.join(
            self.url,
            'vnflcm/v1/vnf_instances',
            vnf_instance['id'])

        self.requests_mock.register_uri(
            'PATCH', url, headers=self.header, json={})

        sys.stdout = buffer = StringIO()
        self.update_vnf_lcm.take_action(parsed_args)

        actual_message = buffer.getvalue().strip()

        expected_message = ('Update vnf:' + vnf_instance['id'])

        self.assertEqual(expected_message, actual_message)

    @mock.patch.object(proxy_client.ClientBase, 'deserialize')
    def test_take_action_check_content_type(self, mock_des):
        vnf_instance = vnflcm_fakes.vnf_instance_response()
        sample_param_file = ('./tackerclient/osc/v1/vnflcm/samples/'
                             'update_vnf_instance_param_sample.json')

        arglist = [vnf_instance['id'], '--I', sample_param_file]
        verifylist = [('vnf_instance', vnf_instance['id']),
                      ('I', sample_param_file)]
        mock_des.return_value = {}

        # command param
        parsed_args = self.check_parser(
            self.update_vnf_lcm, arglist, verifylist)
        url = os.path.join(self.url, 'vnflcm/v1/vnf_instances',
                           vnf_instance['id'])

        self.requests_mock.register_uri(
            'PATCH', url, headers=self.header, json={})

        sys.stdout = buffer = StringIO()

        with mock.patch.object(root_client.HTTPClient,
                               'do_request') as mock_req:
            headers = {'Content-Type': 'application/json'}
            mock_req.return_value = (MyResp(202, headers=headers), None)
            self.update_vnf_lcm.take_action(parsed_args)
            # check content_type
            mock_req.assert_called_once_with(
                f'/vnflcm/v1/vnf_instances/{vnf_instance["id"]}', 'PATCH',
                body=mock.ANY, headers=mock.ANY,
                content_type='application/merge-patch+json', accept='json')

        actual_message = buffer.getvalue().strip()

        expected_message = f'Update vnf:{vnf_instance["id"]}'

        self.assertEqual(expected_message, actual_message)

    def test_take_action_param_file_not_exists(self):
        vnf_instance = vnflcm_fakes.vnf_instance_response()
        sample_param_file = "./not_exists.json"
        arglist = [vnf_instance['id'], '--I', sample_param_file]
        verifylist = [('vnf_instance', vnf_instance['id']),
                      ('I', sample_param_file)]

        # command param
        parsed_args = self.check_parser(self.update_vnf_lcm, arglist,
                                        verifylist)

        self.assertRaises(exceptions.InvalidInput,
                          self.update_vnf_lcm.take_action, parsed_args)


@ddt.ddt
class TestScaleVnfLcm(TestVnfLcm):
    def setUp(self):
        super(TestScaleVnfLcm, self).setUp()
        self.scale_vnf_lcm = vnflcm.ScaleVnfLcm(
            self.app, self.app_args, cmd_name='vnflcm scale')

    @ddt.data('SCALE_IN', 'SCALE_OUT')
    def test_take_action(self, scale_type):
        vnf_instance = vnflcm_fakes.vnf_instance_response()
        sample_param_file = ("./tackerclient/osc/v1/vnflcm/samples/"
                             "scale_vnf_instance_param_sample.json")

        arglist = [vnf_instance['id'],
                   '--aspect-id', uuidsentinel.aspect_id,
                   '--number-of-steps', '1',
                   '--type', scale_type,
                   '--additional-param-file', sample_param_file]
        verifylist = [('vnf_instance', vnf_instance['id']),
                      ('aspect_id', uuidsentinel.aspect_id),
                      ('number_of_steps', 1),
                      ('type', scale_type),
                      ('additional_param_file', sample_param_file)]

        # command param
        parsed_args = self.check_parser(self.scale_vnf_lcm, arglist,
                                        verifylist)
        url = os.path.join(
            self.url,
            'vnflcm/v1/vnf_instances',
            vnf_instance['id'],
            'scale')

        self.requests_mock.register_uri(
            'POST', url, headers=self.header, json={})

        sys.stdout = buffer = StringIO()
        self.scale_vnf_lcm.take_action(parsed_args)

        actual_message = buffer.getvalue().strip()

        expected_message = ("Scale request for VNF Instance %s has been "
                            "accepted.") % vnf_instance['id']

        self.assertEqual(expected_message, actual_message)

    @ddt.data('SCALE_IN', 'SCALE_OUT')
    def test_take_action_no_param_file(self, scale_type):
        vnf_instance = vnflcm_fakes.vnf_instance_response()

        arglist = [vnf_instance['id'],
                   '--aspect-id', uuidsentinel.aspect_id,
                   '--number-of-steps', '1',
                   '--type', scale_type]
        verifylist = [('vnf_instance', vnf_instance['id']),
                      ('aspect_id', uuidsentinel.aspect_id),
                      ('number_of_steps', 1),
                      ('type', scale_type)]

        parsed_args = self.check_parser(self.scale_vnf_lcm, arglist,
                                        verifylist)

        url = os.path.join(self.url, 'vnflcm/v1/vnf_instances',
                           vnf_instance['id'], 'scale')

        self.requests_mock.register_uri(
            'POST', url, headers=self.header, json={})

        sys.stdout = buffer = StringIO()
        self.scale_vnf_lcm.take_action(parsed_args)

        actual_message = buffer.getvalue().strip()

        expected_message = ("Scale request for VNF Instance %s has been "
                            "accepted.") % vnf_instance['id']

        self.assertEqual(expected_message, actual_message)

    @ddt.data('SCALE_IN', 'SCALE_OUT')
    def test_take_action_param_file_not_exists(self, scale_type):
        vnf_instance = vnflcm_fakes.vnf_instance_response()
        sample_param_file = "./not_exists.json"
        arglist = [vnf_instance['id'],
                   '--aspect-id', uuidsentinel.aspect_id,
                   '--number-of-steps', '2',
                   '--type', scale_type,
                   '--additional-param-file', sample_param_file]
        verifylist = [('vnf_instance', vnf_instance['id']),
                      ('aspect_id', uuidsentinel.aspect_id),
                      ('number_of_steps', 2),
                      ('type', scale_type),
                      ('additional_param_file', sample_param_file)]

        # command param
        parsed_args = self.check_parser(self.scale_vnf_lcm, arglist,
                                        verifylist)

        ex = self.assertRaises(exceptions.InvalidInput,
                               self.scale_vnf_lcm.take_action, parsed_args)

        expected_msg = ("Invalid input: "
                        "User does not have read privileges to it")
        self.assertEqual(expected_msg, str(ex))

    @ddt.data('SCALE_IN', 'SCALE_OUT')
    def test_take_action_vnf_instance_not_found(self, scale_type):
        vnf_instance = vnflcm_fakes.vnf_instance_response()
        sample_param_file = ("./tackerclient/osc/v1/vnflcm/samples/"
                             "update_vnf_instance_param_sample.json")
        arglist = [vnf_instance['id'],
                   '--aspect-id', uuidsentinel.aspect_id,
                   '--number-of-steps', '3',
                   '--type', scale_type,
                   '--additional-param-file', sample_param_file]
        verifylist = [('vnf_instance', vnf_instance['id']),
                      ('aspect_id', uuidsentinel.aspect_id),
                      ('number_of_steps', 3),
                      ('type', scale_type),
                      ('additional_param_file', sample_param_file)]

        # command param
        parsed_args = self.check_parser(self.scale_vnf_lcm, arglist,
                                        verifylist)

        url = os.path.join(
            self.url,
            'vnflcm/v1/vnf_instances',
            vnf_instance['id'])
        self.requests_mock.register_uri(
            'POST', url, headers=self.header, status_code=404, json={})

        self.assertRaises(exceptions.TackerClientException,
                          self.scale_vnf_lcm.take_action,
                          parsed_args)


class TestChangeExtConnVnfLcm(TestVnfLcm):

    def setUp(self):
        super(TestChangeExtConnVnfLcm, self).setUp()
        self.change_ext_conn_vnf_lcm = vnflcm.ChangeExtConnVnfLcm(
            self.app, self.app_args,
            cmd_name='vnflcm change-ext-conn')

    def test_take_action(self):
        vnf_instance = vnflcm_fakes.vnf_instance_response()
        sample_param_file = ("./tackerclient/osc/v1/vnflcm/samples/"
                             "change_ext_conn_vnf_instance_param_sample.json")

        arglist = [vnf_instance['id'], sample_param_file]
        verifylist = [('vnf_instance', vnf_instance['id']),
                      ('request_file', sample_param_file)]

        # command param
        parsed_args = self.check_parser(self.change_ext_conn_vnf_lcm,
                                        arglist,
                                        verifylist)

        url = os.path.join(self.url, 'vnflcm/v1/vnf_instances',
                           vnf_instance['id'], 'change_ext_conn')
        self.requests_mock.register_uri(
            'POST', url, headers=self.header, json={})

        sys.stdout = buffer = StringIO()
        with mock.patch.object(proxy_client.ClientBase,
                               '_handle_fault_response') as m:
            self.change_ext_conn_vnf_lcm.take_action(parsed_args)
            # check no fault response is received
            self.assertNotCalled(m)
            self.assertEqual(
                ('Change External VNF Connectivity for VNF Instance {0} '
                 'has been accepted.'.format(vnf_instance['id'])),
                buffer.getvalue().strip())

    def test_take_action_vnf_instance_not_found(self):
        vnf_instance = vnflcm_fakes.vnf_instance_response()
        sample_param_file = ("./tackerclient/osc/v1/vnflcm/samples/"
                             "change_ext_conn_vnf_instance_param_sample.json")
        arglist = [vnf_instance['id'], sample_param_file]
        verifylist = [('vnf_instance', vnf_instance['id']),
                      ('request_file', sample_param_file)]

        # command param
        parsed_args = self.check_parser(self.change_ext_conn_vnf_lcm,
                                        arglist,
                                        verifylist)

        url = os.path.join(self.url, 'vnflcm/v1/vnf_instances',
                           vnf_instance['id'], 'change_ext_conn')
        self.requests_mock.register_uri(
            'POST', url, headers=self.header, status_code=404, json={})

        self.assertRaises(exceptions.TackerClientException,
                          self.change_ext_conn_vnf_lcm.take_action,
                          parsed_args)

    def test_take_action_param_file_not_exists(self):
        vnf_instance = vnflcm_fakes.vnf_instance_response()
        sample_param_file = "./not_exists.json"
        arglist = [vnf_instance['id'], sample_param_file]
        verifylist = [('vnf_instance', vnf_instance['id']),
                      ('request_file', sample_param_file)]

        # command param
        parsed_args = self.check_parser(
            self.change_ext_conn_vnf_lcm,
            arglist,
            verifylist)

        ex = self.assertRaises(
            exceptions.InvalidInput,
            self.change_ext_conn_vnf_lcm.take_action,
            parsed_args)

        expected_msg = ("Invalid input: "
                        "User does not have read privileges to it")
        self.assertEqual(expected_msg, str(ex))

    @mock.patch("os.open")
    @mock.patch("os.access")
    def test_take_action_invalid_format_param_file(self, mock_open,
                                                   mock_access):
        vnf_instance = vnflcm_fakes.vnf_instance_response()
        sample_param_file = "./invalid_param_file.json"
        arglist = [vnf_instance['id'], sample_param_file]
        verifylist = [('vnf_instance', vnf_instance['id']),
                      ('request_file', sample_param_file)]

        mock_open.return_value = "invalid_json_data"
        # command param
        parsed_args = self.check_parser(self.change_ext_conn_vnf_lcm,
                                        arglist,
                                        verifylist)

        ex = self.assertRaises(
            exceptions.InvalidInput,
            self.change_ext_conn_vnf_lcm.take_action,
            parsed_args)

        expected_msg = "Failed to load parameter file."
        self.assertIn(expected_msg, str(ex))


class TestChangeVnfPkgVnfLcm(TestVnfLcm):

    def setUp(self):
        super(TestChangeVnfPkgVnfLcm, self).setUp()
        self.change_vnfpkg_vnf_lcm = vnflcm.ChangeVnfPkgVnfLcm(
            self.app, self.app_args,
            cmd_name='vnflcm change-vnfpkg')

    def test_take_action_with_v1_version(self):
        vnf_instance = vnflcm_fakes.vnf_instance_response()
        sample_param_file = ("./tackerclient/osc/v2/vnflcm/samples/"
                             "change_vnfpkg_vnf_instance_param_sample.json")
        arglist = [vnf_instance['id'], sample_param_file]
        verifylist = [('vnf_instance', vnf_instance['id']),
                      ('request_file', sample_param_file)]

        # command param
        parsed_args = self.check_parser(self.change_vnfpkg_vnf_lcm,
                                        arglist,
                                        verifylist)

        url = os.path.join(self.url, 'vnflcm/v1/vnf_instances',
                           vnf_instance['id'], 'change_vnfpkg')
        self.requests_mock.register_uri(
            'POST', url, headers=self.header, status_code=400, json={})

        ex = self.assertRaises(exceptions.UnsupportedCommandVersion,
                               self.change_vnfpkg_vnf_lcm.take_action,
                               parsed_args)

        expected_msg = "This command is not supported in version 1"
        self.assertEqual(expected_msg, str(ex))


class TestVnfLcmV1(base.FixturedTestCase):
    client_fixture_class = client.ClientFixture
    api_version = '1'

    def setUp(self):
        super(TestVnfLcmV1, self).setUp()

    def test_client_v2(self):
        self.assertEqual(self.cs.vnf_lcm_client.headers,
                         {'Version': '1.3.0'})
        self.assertEqual(self.cs.vnf_lcm_client.vnf_instances_path,
                         '/vnflcm/v1/vnf_instances')
        # check of other paths is omitted.