# -*- coding: utf-8 -*-
#
# Licensed to the Apache Software Foundation (ASF) under one
# or more contributor license agreements.  See the NOTICE file
# distributed with this work for additional information
# regarding copyright ownership.  The ASF licenses this file
# to you 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.

# pylint: disable=too-many-lines

import ast
import unittest
from copy import deepcopy

import httplib2
from googleapiclient.errors import HttpError

from airflow import AirflowException
from airflow.gcp.operators.compute import (
    GceInstanceGroupManagerUpdateTemplateOperator, GceInstanceStartOperator, GceInstanceStopOperator,
    GceInstanceTemplateCopyOperator, GceSetMachineTypeOperator,
)
from airflow.models import DAG, TaskInstance
from airflow.utils import timezone
from tests.compat import mock

EMPTY_CONTENT = b''

GCP_PROJECT_ID = 'project-id'
GCE_ZONE = 'zone'
RESOURCE_ID = 'resource-id'
GCE_SHORT_MACHINE_TYPE_NAME = 'n1-machine-type'
SET_MACHINE_TYPE_BODY = {
    'machineType': 'zones/{}/machineTypes/{}'.format(GCE_ZONE, GCE_SHORT_MACHINE_TYPE_NAME)
}

DEFAULT_DATE = timezone.datetime(2017, 1, 1)


class TestGceInstanceStart(unittest.TestCase):
    @mock.patch('airflow.gcp.operators.compute.ComputeEngineHook')
    def test_instance_start(self, mock_hook):
        mock_hook.return_value.start_instance.return_value = True
        op = GceInstanceStartOperator(
            project_id=GCP_PROJECT_ID,
            zone=GCE_ZONE,
            resource_id=RESOURCE_ID,
            task_id='id'
        )
        result = op.execute(None)
        mock_hook.assert_called_once_with(api_version='v1',
                                          gcp_conn_id='google_cloud_default')
        mock_hook.return_value.start_instance.assert_called_once_with(
            zone=GCE_ZONE, resource_id=RESOURCE_ID, project_id=GCP_PROJECT_ID
        )
        self.assertTrue(result)

    # Setting all of the operator's input parameters as template dag_ids
    # (could be anything else) just to test if the templating works for all fields
    @mock.patch('airflow.gcp.operators.compute.ComputeEngineHook')
    def test_instance_start_with_templates(self, _):
        dag_id = 'test_dag_id'
        args = {
            'start_date': DEFAULT_DATE
        }
        self.dag = DAG(dag_id, default_args=args)  # pylint: disable=attribute-defined-outside-init
        op = GceInstanceStartOperator(
            project_id='{{ dag.dag_id }}',
            zone='{{ dag.dag_id }}',
            resource_id='{{ dag.dag_id }}',
            gcp_conn_id='{{ dag.dag_id }}',
            api_version='{{ dag.dag_id }}',
            task_id='id',
            dag=self.dag
        )
        ti = TaskInstance(op, DEFAULT_DATE)
        ti.render_templates()
        self.assertEqual(dag_id, getattr(op, 'project_id'))
        self.assertEqual(dag_id, getattr(op, 'zone'))
        self.assertEqual(dag_id, getattr(op, 'resource_id'))
        self.assertEqual(dag_id, getattr(op, 'gcp_conn_id'))
        self.assertEqual(dag_id, getattr(op, 'api_version'))

    @mock.patch('airflow.gcp.operators.compute.ComputeEngineHook')
    def test_start_should_throw_ex_when_missing_project_id(self, mock_hook):
        with self.assertRaises(AirflowException) as cm:
            op = GceInstanceStartOperator(
                project_id="",
                zone=GCE_ZONE,
                resource_id=RESOURCE_ID,
                task_id='id'
            )
            op.execute(None)
        err = cm.exception
        self.assertIn("The required parameter 'project_id' is missing", str(err))
        mock_hook.assert_not_called()

    @mock.patch('airflow.gcp.operators.compute.ComputeEngineHook')
    def test_start_should_not_throw_ex_when_project_id_none(self, _):
        op = GceInstanceStartOperator(
            zone=GCE_ZONE,
            resource_id=RESOURCE_ID,
            task_id='id'
        )
        op.execute(None)

    @mock.patch('airflow.gcp.operators.compute.ComputeEngineHook')
    def test_start_should_throw_ex_when_missing_zone(self, mock_hook):
        with self.assertRaises(AirflowException) as cm:
            op = GceInstanceStartOperator(
                project_id=GCP_PROJECT_ID,
                zone="",
                resource_id=RESOURCE_ID,
                task_id='id'
            )
            op.execute(None)
        err = cm.exception
        self.assertIn("The required parameter 'zone' is missing", str(err))
        mock_hook.assert_not_called()

    @mock.patch('airflow.gcp.operators.compute.ComputeEngineHook')
    def test_start_should_throw_ex_when_missing_resource_id(self, mock_hook):
        with self.assertRaises(AirflowException) as cm:
            op = GceInstanceStartOperator(
                project_id=GCP_PROJECT_ID,
                zone=GCE_ZONE,
                resource_id="",
                task_id='id'
            )
            op.execute(None)
        err = cm.exception
        self.assertIn("The required parameter 'resource_id' is missing", str(err))
        mock_hook.assert_not_called()


class TestGceInstanceStop(unittest.TestCase):
    @mock.patch('airflow.gcp.operators.compute.ComputeEngineHook')
    def test_instance_stop(self, mock_hook):
        op = GceInstanceStopOperator(
            project_id=GCP_PROJECT_ID,
            zone=GCE_ZONE,
            resource_id=RESOURCE_ID,
            task_id='id'
        )
        op.execute(None)
        mock_hook.assert_called_once_with(api_version='v1',
                                          gcp_conn_id='google_cloud_default')
        mock_hook.return_value.stop_instance.assert_called_once_with(
            zone=GCE_ZONE, resource_id=RESOURCE_ID, project_id=GCP_PROJECT_ID
        )

    # Setting all of the operator's input parameters as templated dag_ids
    # (could be anything else) just to test if the templating works for all fields
    @mock.patch('airflow.gcp.operators.compute.ComputeEngineHook')
    def test_instance_stop_with_templates(self, _):
        dag_id = 'test_dag_id'
        args = {
            'start_date': DEFAULT_DATE
        }
        self.dag = DAG(dag_id, default_args=args)  # pylint: disable=attribute-defined-outside-init
        op = GceInstanceStopOperator(
            project_id='{{ dag.dag_id }}',
            zone='{{ dag.dag_id }}',
            resource_id='{{ dag.dag_id }}',
            gcp_conn_id='{{ dag.dag_id }}',
            api_version='{{ dag.dag_id }}',
            task_id='id',
            dag=self.dag
        )
        ti = TaskInstance(op, DEFAULT_DATE)
        ti.render_templates()
        self.assertEqual(dag_id, getattr(op, 'project_id'))
        self.assertEqual(dag_id, getattr(op, 'zone'))
        self.assertEqual(dag_id, getattr(op, 'resource_id'))
        self.assertEqual(dag_id, getattr(op, 'gcp_conn_id'))
        self.assertEqual(dag_id, getattr(op, 'api_version'))

    @mock.patch('airflow.gcp.operators.compute.ComputeEngineHook')
    def test_stop_should_throw_ex_when_missing_project_id(self, mock_hook):
        with self.assertRaises(AirflowException) as cm:
            op = GceInstanceStopOperator(
                project_id="",
                zone=GCE_ZONE,
                resource_id=RESOURCE_ID,
                task_id='id'
            )
            op.execute(None)
        err = cm.exception
        self.assertIn("The required parameter 'project_id' is missing", str(err))
        mock_hook.assert_not_called()

    @mock.patch('airflow.gcp.operators.compute.ComputeEngineHook')
    def test_stop_should_not_throw_ex_when_project_id_none(self, mock_hook):
        op = GceInstanceStopOperator(
            zone=GCE_ZONE,
            resource_id=RESOURCE_ID,
            task_id='id'
        )
        op.execute(None)
        mock_hook.assert_called_once_with(api_version='v1',
                                          gcp_conn_id='google_cloud_default')
        mock_hook.return_value.stop_instance.assert_called_once_with(
            zone=GCE_ZONE, resource_id=RESOURCE_ID, project_id=None
        )

    @mock.patch('airflow.gcp.operators.compute.ComputeEngineHook')
    def test_stop_should_throw_ex_when_missing_zone(self, mock_hook):
        with self.assertRaises(AirflowException) as cm:
            op = GceInstanceStopOperator(
                project_id=GCP_PROJECT_ID,
                zone="",
                resource_id=RESOURCE_ID,
                task_id='id'
            )
            op.execute(None)
        err = cm.exception
        self.assertIn("The required parameter 'zone' is missing", str(err))
        mock_hook.assert_not_called()

    @mock.patch('airflow.gcp.operators.compute.ComputeEngineHook')
    def test_stop_should_throw_ex_when_missing_resource_id(self, mock_hook):
        with self.assertRaises(AirflowException) as cm:
            op = GceInstanceStopOperator(
                project_id=GCP_PROJECT_ID,
                zone=GCE_ZONE,
                resource_id="",
                task_id='id'
            )
            op.execute(None)
        err = cm.exception
        self.assertIn("The required parameter 'resource_id' is missing", str(err))
        mock_hook.assert_not_called()


class TestGceInstanceSetMachineType(unittest.TestCase):
    @mock.patch('airflow.gcp.operators.compute.ComputeEngineHook')
    def test_set_machine_type(self, mock_hook):
        mock_hook.return_value.set_machine_type.return_value = True
        op = GceSetMachineTypeOperator(
            project_id=GCP_PROJECT_ID,
            zone=GCE_ZONE,
            resource_id=RESOURCE_ID,
            body=SET_MACHINE_TYPE_BODY,
            task_id='id'
        )
        op.execute(None)
        mock_hook.assert_called_once_with(api_version='v1',
                                          gcp_conn_id='google_cloud_default')
        mock_hook.return_value.set_machine_type.assert_called_once_with(
            zone=GCE_ZONE,
            resource_id=RESOURCE_ID,
            body=SET_MACHINE_TYPE_BODY,
            project_id=GCP_PROJECT_ID
        )

    # Setting all of the operator's input parameters as templated dag_ids
    # (could be anything else) just to test if the templating works for all fields
    @mock.patch('airflow.gcp.operators.compute.ComputeEngineHook')
    def test_set_machine_type_with_templates(self, _):
        dag_id = 'test_dag_id'
        args = {
            'start_date': DEFAULT_DATE
        }
        self.dag = DAG(dag_id, default_args=args)  # pylint: disable=attribute-defined-outside-init
        op = GceSetMachineTypeOperator(
            project_id='{{ dag.dag_id }}',
            zone='{{ dag.dag_id }}',
            resource_id='{{ dag.dag_id }}',
            body={},
            gcp_conn_id='{{ dag.dag_id }}',
            api_version='{{ dag.dag_id }}',
            task_id='id',
            dag=self.dag
        )
        ti = TaskInstance(op, DEFAULT_DATE)
        ti.render_templates()
        self.assertEqual(dag_id, getattr(op, 'project_id'))
        self.assertEqual(dag_id, getattr(op, 'zone'))
        self.assertEqual(dag_id, getattr(op, 'resource_id'))
        self.assertEqual(dag_id, getattr(op, 'gcp_conn_id'))
        self.assertEqual(dag_id, getattr(op, 'api_version'))

    @mock.patch('airflow.gcp.operators.compute.ComputeEngineHook')
    def test_set_machine_type_should_throw_ex_when_missing_project_id(self, mock_hook):
        with self.assertRaises(AirflowException) as cm:
            op = GceSetMachineTypeOperator(
                project_id="",
                zone=GCE_ZONE,
                resource_id=RESOURCE_ID,
                body=SET_MACHINE_TYPE_BODY,
                task_id='id'
            )
            op.execute(None)
        err = cm.exception
        self.assertIn("The required parameter 'project_id' is missing", str(err))
        mock_hook.assert_not_called()

    @mock.patch('airflow.gcp.operators.compute.ComputeEngineHook')
    def test_set_machine_type_should_not_throw_ex_when_project_id_none(self, mock_hook):
        op = GceSetMachineTypeOperator(
            zone=GCE_ZONE,
            resource_id=RESOURCE_ID,
            body=SET_MACHINE_TYPE_BODY,
            task_id='id'
        )
        op.execute(None)
        mock_hook.assert_called_once_with(api_version='v1',
                                          gcp_conn_id='google_cloud_default')
        mock_hook.return_value.set_machine_type.assert_called_once_with(
            zone=GCE_ZONE,
            resource_id=RESOURCE_ID,
            body=SET_MACHINE_TYPE_BODY,
            project_id=None
        )

    @mock.patch('airflow.gcp.operators.compute.ComputeEngineHook')
    def test_set_machine_type_should_throw_ex_when_missing_zone(self, mock_hook):
        with self.assertRaises(AirflowException) as cm:
            op = GceSetMachineTypeOperator(
                project_id=GCP_PROJECT_ID,
                zone="",
                resource_id=RESOURCE_ID,
                body=SET_MACHINE_TYPE_BODY,
                task_id='id'
            )
            op.execute(None)
        err = cm.exception
        self.assertIn("The required parameter 'zone' is missing", str(err))
        mock_hook.assert_not_called()

    @mock.patch('airflow.gcp.operators.compute.ComputeEngineHook')
    def test_set_machine_type_should_throw_ex_when_missing_resource_id(self, mock_hook):
        with self.assertRaises(AirflowException) as cm:
            op = GceSetMachineTypeOperator(
                project_id=GCP_PROJECT_ID,
                zone=GCE_ZONE,
                resource_id="",
                body=SET_MACHINE_TYPE_BODY,
                task_id='id'
            )
            op.execute(None)
        err = cm.exception
        self.assertIn("The required parameter 'resource_id' is missing", str(err))
        mock_hook.assert_not_called()

    @mock.patch('airflow.gcp.operators.compute.ComputeEngineHook')
    def test_set_machine_type_should_throw_ex_when_missing_machine_type(self, mock_hook):
        with self.assertRaises(AirflowException) as cm:
            op = GceSetMachineTypeOperator(
                project_id=GCP_PROJECT_ID,
                zone=GCE_ZONE,
                resource_id=RESOURCE_ID,
                body={},