# Copyright (c) 2013 Mirantis Inc.
#
# 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
import datetime

from sqlalchemy import exc as sa_exc
import testtools

from sahara import context
from sahara import exceptions as ex
import sahara.tests.unit.conductor.base as test_base
from sahara.tests.unit.conductor.manager import test_clusters
from sahara.utils import edp


SAMPLE_DATA_SOURCE = {
    "tenant_id": "test_tenant",
    "name": "ngt_test",
    "description": "test_desc",
    "type": "Cassandra",
    "url": "localhost:1080",
    "credentials": {
        "user": "test",
        "password": "123"
    }
}

SAMPLE_JOB = {
    "tenant_id": "test_tenant",
    "name": "job_test",
    "description": "test_desc",
    "type": edp.JOB_TYPE_PIG,
    "mains": []
}

SAMPLE_JOB_EXECUTION = {
    "tenant_id": "tenant_id",
    "return_code": "1",
    "job_id": "undefined",
    "input_id": "undefined",
    "output_id": "undefined",
    "start_time": datetime.datetime.now(),
    "cluster_id": None
}

SAMPLE_CONF_JOB_EXECUTION = {
    "tenant_id": "tenant_id",
    "progress": "0.1",
    "return_code": "1",
    "job_id": "undefined",
    "input_id": "undefined",
    "output_id": "undefined",
    "cluster_id": None,
    "job_configs": {
        "conf2": "value_je",
        "conf3": "value_je"
    }
}

BINARY_DATA = "vU}\x97\x1c\xdf\xa686\x08\xf2\tf\x0b\xb1}"

SAMPLE_JOB_BINARY_INTERNAL = {
    "tenant_id": "test_tenant",
    "name": "job_test",
    "data": BINARY_DATA
}


SAMPLE_JOB_BINARY = {
    "tenant_id": "test_tenant",
    "name": "job_binary_test",
    "description": "test_dec",
    "url": "internal-db://test_binary",
}


class DataSourceTest(test_base.ConductorManagerTestCase):
    def __init__(self, *args, **kwargs):
        super(DataSourceTest, self).__init__(
            checks=[
                lambda: SAMPLE_DATA_SOURCE
            ], *args, **kwargs)

    def test_crud_operation_create_list_delete(self):
        ctx = context.ctx()
        self.api.data_source_create(ctx, SAMPLE_DATA_SOURCE)

        lst = self.api.data_source_get_all(ctx)
        self.assertEqual(1, len(lst))

        ds_id = lst[0]['id']
        self.api.data_source_destroy(ctx, ds_id)

        lst = self.api.data_source_get_all(ctx)
        self.assertEqual(0, len(lst))

    def test_duplicate_data_source_create(self):
        ctx = context.ctx()
        self.api.data_source_create(ctx, SAMPLE_DATA_SOURCE)
        with testtools.ExpectedException(ex.DBDuplicateEntry):
            self.api.data_source_create(ctx, SAMPLE_DATA_SOURCE)

    def test_data_source_fields(self):
        ctx = context.ctx()
        ctx.tenant_id = SAMPLE_DATA_SOURCE['tenant_id']
        ds_db_obj_id = self.api.data_source_create(ctx,
                                                   SAMPLE_DATA_SOURCE)['id']

        ds_db_obj = self.api.data_source_get(ctx, ds_db_obj_id)
        self.assertIsInstance(ds_db_obj, dict)

        for key, val in SAMPLE_DATA_SOURCE.items():
            self.assertEqual(val, ds_db_obj.get(key),
                             "Key not found %s" % key)

    def test_data_source_delete(self):
        ctx = context.ctx()
        db_obj_ds = self.api.data_source_create(ctx, SAMPLE_DATA_SOURCE)
        _id = db_obj_ds['id']

        self.api.data_source_destroy(ctx, _id)

        with testtools.ExpectedException(ex.NotFoundException):
            self.api.data_source_destroy(ctx, _id)

    def test_data_source_search(self):
        ctx = context.ctx()
        ctx.tenant_id = SAMPLE_DATA_SOURCE['tenant_id']
        self.api.data_source_create(ctx, SAMPLE_DATA_SOURCE)

        lst = self.api.data_source_get_all(ctx)
        self.assertEqual(1, len(lst))

        kwargs = {'name': SAMPLE_DATA_SOURCE['name'],
                  'tenant_id': SAMPLE_DATA_SOURCE['tenant_id']}
        lst = self.api.data_source_get_all(ctx, **kwargs)
        self.assertEqual(1, len(lst))

        # Valid field but no matching value
        kwargs = {'name': SAMPLE_DATA_SOURCE['name']+"foo"}
        lst = self.api.data_source_get_all(ctx, **kwargs)
        self.assertEqual(0, len(lst))

        # Invalid field
        self.assertRaises(sa_exc.InvalidRequestError,
                          self.api.data_source_get_all,
                          ctx, **{'badfield': 'somevalue'})

    def test_data_source_count_in(self):
        ctx = context.ctx()
        ctx.tenant_id = SAMPLE_DATA_SOURCE['tenant_id']
        src = copy.copy(SAMPLE_DATA_SOURCE)
        self.api.data_source_create(ctx, src)

        cnt = self.api.data_source_count(ctx, name='ngt_test')
        self.assertEqual(1, cnt)

        cnt = self.api.data_source_count(ctx, name=('ngt_test',
                                                    'test2', 'test3'))
        self.assertEqual(1, cnt)

        cnt = self.api.data_source_count(ctx, name=('test1',
                                                    'test2', 'test3'))
        self.assertEqual(0, cnt)

        lst = self.api.data_source_get_all(ctx, name='ngt_test')
        myid = lst[0]['id']
        cnt = self.api.data_source_count(ctx,
                                         name=('ngt_test', 'test2', 'test3'),
                                         id=myid)
        self.assertEqual(1, cnt)

        cnt = self.api.data_source_count(ctx,
                                         name=('ngt_test', 'test2', 'test3'),
                                         id=(myid, '2'))
        self.assertEqual(1, cnt)

    def test_data_source_count_like(self):
        ctx = context.ctx()
        ctx.tenant_id = SAMPLE_DATA_SOURCE['tenant_id']
        src = copy.copy(SAMPLE_DATA_SOURCE)
        self.api.data_source_create(ctx, src)

        cnt = self.api.data_source_count(ctx, name='ngt_test')
        self.assertEqual(1, cnt)

        cnt = self.api.data_source_count(ctx, name='ngt%')
        self.assertEqual(1, cnt)

        cnt = self.api.data_source_count(ctx,
                                         name=('ngt_test',),
                                         url='localhost%')
        self.assertEqual(1, cnt)

        cnt = self.api.data_source_count(ctx,
                                         name=('ngt_test',),
                                         url='localhost')
        self.assertEqual(0, cnt)

    def test_data_source_update(self):
        ctx = context.ctx()
        orig = self.api.data_source_create(ctx, SAMPLE_DATA_SOURCE)
        update_json = {"name": "updatedName",
                       "url": "swift://updatedFakeUrl"}
        updated = self.api.data_source_update(ctx, orig["id"], update_json)
        self.assertEqual("updatedName", updated["name"])
        self.assertEqual("swift://updatedFakeUrl", updated["url"])

        self._create_job_execution_ref_data_source(ctx, updated["id"])
        update_json = {"name": "FailsToupdatedName",
                       "url": "swift://FailsupdatedFakeUrl"}
        with testtools.ExpectedException(ex.UpdateFailedException):
            self.api.data_source_update(ctx, updated["id"], update_json)

    def _create_job_execution_ref_data_source(self, context, ds_id):
        job = self.api.job_create(context, SAMPLE_JOB)
        ds_input = self.api.data_source_create(context, SAMPLE_DATA_SOURCE)
        SAMPLE_DATA_OUTPUT = copy.copy(SAMPLE_DATA_SOURCE)
        SAMPLE_DATA_OUTPUT['name'] = 'output'

        SAMPLE_JOB_EXECUTION['job_id'] = job['id']
        SAMPLE_JOB_EXECUTION['input_id'] = ds_input['id']
        SAMPLE_JOB_EXECUTION['output_id'] = ds_id
        SAMPLE_JOB_EXECUTION['data_source_urls'] = {ds_id: "fakeurl"}

        self.api.job_execution_create(context, SAMPLE_JOB_EXECUTION)

        lst = self.api.job_execution_get_all(context)
        job_ex_id = lst[0]['id']

        new_info = {"status": edp.JOB_STATUS_PENDING}
        self.api.job_execution_update(context, job_ex_id, {'info': new_info})


class JobExecutionTest(test_base.ConductorManagerTestCase):
    def test_crud_operation_create_list_delete_update(self):
        ctx = context.ctx()
        job = self.api.job_create(ctx, SAMPLE_JOB)
        ds_input = self.api.data_source_create(ctx, SAMPLE_DATA_SOURCE)
        SAMPLE_DATA_OUTPUT = copy.copy(SAMPLE_DATA_SOURCE)
        SAMPLE_DATA_OUTPUT['name'] = 'output'
        ds_output = self.api.data_source_create(ctx, SAMPLE_DATA_OUTPUT)

        SAMPLE_JOB_EXECUTION['job_id'] = job['id']
        SAMPLE_JOB_EXECUTION['input_id'] = ds_input['id']
        SAMPLE_JOB_EXECUTION['output_id'] = ds_output['id']

        self.api.job_execution_create(ctx, SAMPLE_JOB_EXECUTION)

        lst = self.api.job_execution_get_all(ctx)
        self.assertEqual(1, len(lst))

        count = self.api.job_execution_count(ctx)
        self.assertEqual(1, count)

        job_ex_id = lst[0]['id']

        self.assertIsNone(lst[0]['info'])
        new_info = {"status": edp.JOB_STATUS_PENDING}
        self.api.job_execution_update(ctx, job_ex_id, {'info': new_info})
        updated_job = self.api.job_execution_get(ctx, job_ex_id)
        self.assertEqual(new_info, updated_job['info'])
        self.assertEqual(SAMPLE_JOB_EXECUTION['start_time'],
                         updated_job['start_time'])

        self.api.job_execution_destroy(ctx, job_ex_id)

        with testtools.ExpectedException(ex.NotFoundException):
            self.api.job_execution_update(ctx, job_ex_id, {'info': new_info})

        with testtools.ExpectedException(ex.NotFoundException):
            self.api.job_execution_destroy(ctx, job_ex_id)

        lst = self.api.job_execution_get_all(ctx)
        self.assertEqual(0, len(lst))

    def test_crud_operation_on_configured_jobs(self):
        ctx = context.ctx()
        job = self.api.job_create(ctx, SAMPLE_JOB)
        ds_input = self.api.data_source_create(ctx, SAMPLE_DATA_SOURCE)
        SAMPLE_DATA_OUTPUT = copy.copy(SAMPLE_DATA_SOURCE)
        SAMPLE_DATA_OUTPUT['name'] = 'output'
        ds_output = self.api.data_source_create(ctx, SAMPLE_DATA_OUTPUT)

        SAMPLE_CONF_JOB_EXECUTION['job_id'] = job['id']
        SAMPLE_CONF_JOB_EXECUTION['input_id'] = ds_input['id']
        SAMPLE_CONF_JOB_EXECUTION['output_id'] = ds_output['id']

        self.api.job_execution_create(ctx, SAMPLE_CONF_JOB_EXECUTION)

        lst = self.api.job_execution_get_all(ctx)
        self.assertEqual(1, len(lst))

        job_ex = lst[0]
        configs = {
            'conf2': 'value_je',
            'conf3': 'value_je'
        }
        self.assertEqual(configs, job_ex['job_configs'])

    def test_deletion_constraints_on_data_and_jobs(self):
        ctx = context.ctx()
        job = self.api.job_create(ctx, SAMPLE_JOB)
        ds_input = self.api.data_source_create(ctx, SAMPLE_DATA_SOURCE)