#!/usr/bin/env python
# -*- coding: utf-8 -*-
# Licensed to Cloudera, Inc. under one
# or more contributor license agreements.  See the NOTICE file
# distributed with this work for additional information
# regarding copyright ownership.  Cloudera, Inc. 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.

from builtins import object
from datetime import datetime
import json

from nose.tools import assert_equal, assert_false, assert_true

from beeswax.models import SavedQuery
from beeswax.design import hql_query
from librdbms.design import SQLdesign
from oozie.models import Link, Workflow
from oozie.tests import add_node
from pig.models import create_or_update_script
from useradmin.models import get_default_user_group, User

from desktop.converters import DocumentConverter
from desktop.lib.django_test_util import make_logged_in_client
from desktop.lib.test_utils import grant_access
from desktop.models import Directory, Document, Document2, DocumentPermission, DocumentTag


class TestDocumentConverter(object):

  def setUp(self):
    self.client = make_logged_in_client(username="doc2", groupname="doc2", recreate=True, is_superuser=False)
    self.user = User.objects.get(username="doc2")
    grant_access("doc2", "doc2", "beeswax")
    grant_access("doc2", "doc2", "pig")
    grant_access("doc2", "doc2", "jobsub")

    # This creates the user directories for the new user
    response = self.client.get('/desktop/api2/doc/')
    data = json.loads(response.content)
    assert_equal('/', data['document']['path'], data)

    self.home_dir = Document2.objects.get_home_directory(user=self.user)


  def test_convert_hive_query(self):
    sql = 'SELECT * FROM sample_07'
    settings = [
        {'key': 'hive.exec.scratchdir', 'value': '/tmp/mydir'},
        {'key': 'hive.querylog.location', 'value': '/tmp/doc2'}
    ]
    file_resources = [{'type': 'jar', 'path': '/tmp/doc2/test.jar'}]
    functions = [{'name': 'myUpper', 'class_name': 'org.hue.udf.MyUpper'}]
    design = hql_query(sql, database='etl', settings=settings, file_resources=file_resources, functions=functions)

    query = SavedQuery.objects.create(
        type=SavedQuery.TYPES_MAPPING['hql'],
        owner=self.user,
        data=design.dumps(),
        name='Hive query',
        desc='Test Hive query'
    )
    doc = Document.objects.link(query, owner=query.owner, extra=query.type, name=query.name, description=query.desc)

    # Setting doc.last_modified to older date
    Document.objects.filter(id=doc.id).update(last_modified=datetime.strptime('2000-01-01T00:00:00Z', '%Y-%m-%dT%H:%M:%SZ'))
    doc = Document.objects.get(id=doc.id)

    query2 = SavedQuery.objects.create(
        type=SavedQuery.TYPES_MAPPING['hql'],
        owner=self.user,
        data=design.dumps(),
        name='Hive query h
istory',
        desc='Test Hive query history',
        is_auto=True
    )
    doch = Document.objects.link(query2, owner=query2.owner, extra=query2.type, name=query2.name, description=query2.desc)
    doch.add_to_history()

    try:
      # Test that corresponding doc2 is created after convert
      assert_equal(0, Document2.objects.filter(owner=self.user, type='query-hive').count())

      
converter = DocumentConverter(self.user)
      converter.convert()

      assert_equal(2, Document2.objects.filter(owner=self.user, type='query-hive').count())

      #
      # Query
      #
      doc2 = Document2.objects.get(owner=self.user, type='query-hive', is_history=False)

      # Verify Document2 attributes
      assert_equal(doc.name, doc2.data_dict['name'])
      assert_equal(doc.description, doc2.data_dict['description'])

      # Verify session type
      assert_equal('hive', doc2.data_dict['sessions'][0]['type'])

      # Verify snippet values
      assert_equal('ready', doc2.data_dict['snippets'][0]['status'])
      assert_equal(sql, doc2.data_dict['snippets'][0]['statement'])
      assert_equal(sql, doc2.data_dict['snippets'][0]['statement_raw'])
      assert_equal('etl', doc2.data_dict['snippets'][0]['database'])

      # Verify snippet properties
      assert_equal(settings, doc2.data_dict['snippets'][0]['properties']['settings'])
      assert_equal(file_resources, doc2.data_dict['snippets'][0]['properties']['files'])
      assert_equal(functions, doc2.data_dict['snippets'][0]['properties']['functions'])

      # Verify default properties
      assert_true(doc2.data_dict['isSaved'])
      assert_equal(doc.last_modified.strftime('%Y-%m-%dT%H:%M:%S'), doc2.last_modified.strftime('%Y-%m-%dT%H:%M:%S'))

      #
      # Query History
      #
      doc2 = Document2.objects.get(owner=self.user, type='query-hive', is_history=True)

      # Verify Document2 attributes
      assert_equal(doch.name, doc2.data_dict['name'])
      assert_equal(doch.description, doc2.data_dict['description'])
      assert_equal(doch.last_modified.strftime('%Y-%m-%dT%H:%M:%S'), doc2.last_modified.strftime('%Y-%m-%dT%H:%M:%S'))

      # Verify session type
      assert_false(doc2.data_dict['sessions'])

      # Verify snippet values
      assert_equal('ready', doc2.data_dict['snippets'][0]['status'])
      assert_equal(sql, doc2.data_dict['snippets'][0]['statement'])
      assert_equal(sql, doc2.data_dict['snippets'][0]['statement_raw'])
      assert_equal('etl', doc2.data_dict['snippets'][0]['database'])

      # Verify snippet properties
      assert_equal(settings, doc2.data_dict['snippets'][0]['properties']['settings'])
      assert_equal(file_resources, doc2.data_dict['snippets'][0]['properties']['files'])
      assert_equal(functions, doc2.data_dict['snippets'][0]['properties']['functions'])

      # Verify default properties
      assert_false(doc2.data_dict['isSaved'])


      #
      # Check that we don't re-import again
      #
      converter = DocumentConverter(self.user)
      converter.convert()

      assert_equal(2, Document2.objects.filter(owner=self.user, type='query-hive').count())
    finally:
      query.delete()
      query2.delete()


  def test_convert_hive_query_with_special_chars(self):
    sql = 'SELECT * FROM sample_07'
    settings = [
      {'key': 'hive.exec.scratchdir', 'value': '/tmp/mydir'},
      {'key': 'hive.querylog.location', 'value': '/tmp/doc2'}
    ]
    file_resources = [{'type': 'jar', 'path': '/tmp/doc2/test.jar'}]
    functions = [{'name': 'myUpper', 'class_name': 'org.hue.udf.MyUpper'}]
    design = hql_query(sql, database='etl', settings=settings, file_resources=file_resources, functions=functions)

    query = SavedQuery.objects.create(
      type=SavedQuery.TYPES_MAPPING['hql'],
      owner=self.user,
      data=design.dumps(),
      name='Test / Hive query',
      desc='Test Hive query'
    )
    doc = Document.objects.link(query, owner=query.owner, extra=query.type, name=query.name, description=query.desc)

    try:
      # Test that corresponding doc2 is created after convert
      assert_equal(0, Document2.objects.filter(owner=self.user, type='query-hive').count())

      converter = DocumentConverter(self.user)
      converter.convert()

      assert_equal(1, Document2.objects.filter(owner=self.user, type='query-hive').count())

      doc2 = Document2.objects.get(owner=self.user, type='query-hive', is_history=False)

      # Verify name is maintained
      assert_equal('Test / Hive query', doc2.name)

      # Verify Document2 path is stripped of invalid chars
      assert_equal('/Test%20/%20Hive%20query', doc2.path)
    finally:
      query.delete()


  def test_convert_impala_query(self):
    sql = 'SELECT * FROM sample_07'
    settings = [
        {'key': 'EXPLAIN_LEVEL', 'value': '2'},
        {'key': 'ABORT_ON_ERROR', 'value': '1'}
    ]
    design = hq