# copyright 2003 LOGILAB S.A. (Paris, FRANCE), all rights reserved.

# contact http://www.logilab.fr/ -- mailto:contact@logilab.fr

#

# This file is part of CubicWeb.

#

# CubicWeb is free software: you can redistribute it and/or modify it under the

# terms of the GNU Lesser General Public License as published by the Free

# Software Foundation, either version 2.1 of the License, or (at your option)

# any later version.

#

# CubicWeb is distributed in the hope that it will be useful, but WITHOUT

# ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS

# FOR A PARTICULAR PURPOSE.  See the GNU Lesser General Public License for more

# details.

#

# You should have received a copy of the GNU Lesser General Public License along

# with CubicWeb.  If not, see <http://www.gnu.org/licenses/>.

"""unit tests for module cubicweb.server.migractions"""

import os

import os.path as osp

import sys

from datetime import date

from contextlib import contextmanager

from logilab.common import tempattr

from yams.constraints import UniqueConstraint

from cubicweb import (ConfigurationError, ValidationError,

                      ExecutionError, Binary)

from cubicweb.devtools import startpgcluster, stoppgcluster

from cubicweb.schema import constraint_name_for

from cubicweb.server.sqlutils import SQL_PREFIX

from cubicweb.server.migractions import ServerMigrationHelper

from cubicweb.server.sources import storages

from cubicweb.server.schema2sql import build_index_name

import cubicweb.devtools

HERE = osp.dirname(osp.abspath(__file__))

migrschema = None

def setUpModule():

    startpgcluster(__file__)

def tearDownModule(*args):

    global migrschema

    del migrschema

    if hasattr(MigrationCommandsTC, 'origschema'):

        del MigrationCommandsTC.origschema

    if hasattr(MigrationCommandsComputedTC, 'origschema'):

        del MigrationCommandsComputedTC.origschema

    stoppgcluster(__file__)

class MigrationConfig(cubicweb.devtools.TestServerConfiguration):

    default_sources = cubicweb.devtools.DEFAULT_PSQL_SOURCES

class MigrationTC(CubicWebTC):

    appid = 'data-migractions'

    configcls = MigrationConfig

    def _init_repo(self):

        super(MigrationTC, self)._init_repo()

        # we have to read schema from the database to get eid for schema entities

        self.repo.set_schema(self.repo.deserialize_schema(), resetvreg=False)

        # hack to read the schema from data/migrschema

        @contextmanager

        def temp_app(config, appid, apphome):

            old = config.apphome, config.appid

            sys.path.remove(old[0])

            sys.path.insert(0, apphome)

            config._apphome, config.appid = apphome, appid

            try:

                yield config

            finally:

                sys.path.remove(apphome)

                sys.path.insert(0, old[0])

                config._apphome, config.appid = old

        appid = osp.join(self.appid, 'migratedapp')

        apphome = osp.join(HERE, appid)

        with temp_app(self.config, appid, apphome) as config:

            global migrschema

            migrschema = config.load_schema()

    def tearDown(self):

        super(MigrationTC, self).tearDown()

        self.repo.vreg['etypes'].clear_caches()

    @contextmanager

    def mh(self):

        with self.admin_access.repo_cnx() as cnx:

            yield cnx, ServerMigrationHelper(self.repo.config, migrschema,

                                             repo=self.repo, cnx=cnx,

                                             interactive=False)

    def table_sql(self, mh, tablename):

        result = mh.sqlexec(

            "SELECT table_name FROM information_schema.tables WHERE LOWER(table_name)=%(table)s",

            {'table': tablename.lower()})

        if result:

            return result[0][0]

        return None  # no such table

    def table_schema(self, mh, tablename):

        result = mh.sqlexec(

            "SELECT column_name, data_type, character_maximum_length "

            "FROM information_schema.columns "

            "WHERE LOWER(table_name) = %(table)s", {'table': tablename.lower()})

        assert result, 'no table %s' % tablename

        return dict((x[0], (x[1], x[2])) for x in result)

    def table_constraints(self, mh, tablename):

        result = mh.sqlexec(

            "SELECT DISTINCT constraint_name FROM information_schema.constraint_column_usage "

            "WHERE LOWER(table_name) = '%(table)s' AND constraint_name LIKE 'cstr%%'"

            % {'table': tablename.lower()})

        assert result, 'no table %s' % tablename

        return set(x[0] for x in result)

class MigrationCommandsTC(MigrationTC):

    def _init_repo(self):

        super(MigrationCommandsTC, self)._init_repo()

        assert 'Folder' in migrschema

    def test_add_attribute_bool(self):

        with self.mh() as (cnx, mh):

            self.assertNotIn('yesno', self.schema)

            cnx.create_entity('Note')

            cnx.commit()

            mh.cmd_add_attribute('Note', 'yesno')

            self.assertIn('yesno', self.schema)

            self.assertEqual(self.schema['yesno'].subjects(), ('Note',))

            self.assertEqual(self.schema['yesno'].objects(), ('Boolean',))

            self.assertEqual(self.schema['Note'].default('yesno'), False)

            # test default value set on existing entities

            note = cnx.execute('Note X').get_entity(0, 0)

            self.assertEqual(note.yesno, False)

            # test default value set for next entities

            self.assertEqual(cnx.create_entity('Note').yesno, False)

    def test_add_attribute_int(self):

        with self.mh() as (cnx, mh):

            self.assertNotIn('whatever', self.schema)

            cnx.create_entity('Note')

            cnx.commit()

            orderdict = dict(mh.rqlexec('Any RTN, O WHERE X name "Note", RDEF from_entity X, '

                                        'RDEF relation_type RT, RDEF ordernum O, RT name RTN'))

            mh.cmd_add_attribute('Note', 'whatever')

            self.assertIn('whatever', self.schema)

            self.assertEqual(self.schema['whatever'].subjects(), ('Note',))

            self.assertEqual(self.schema['whatever'].objects(), ('Int',))

            self.assertEqual(self.schema['Note'].default('whatever'), 0)

            # test default value set on existing entities

            note = cnx.execute('Note X').get_entity(0, 0)

            self.assertIsInstance(note.whatever, int)

            self.assertEqual(note.whatever, 0)

            # test default value set for next entities

            self.assertEqual(cnx.create_entity('Note').whatever, 0)

            # test attribute order

            orderdict2 = dict(mh.rqlexec('Any RTN, O WHERE X name "Note", RDEF from_entity X, '

                                         'RDEF relation_type RT, RDEF ordernum O, RT name RTN'))

            whateverorder = migrschema['whatever'].rdef('Note', 'Int').order

            for k, v in orderdict.items():

                if v >= whateverorder:

                    orderdict[k] = v + 1

            orderdict['whatever'] = whateverorder

            self.assertDictEqual(orderdict, orderdict2)

    def test_add_attribute_varchar(self):

        with self.mh() as (cnx, mh):

            self.assertNotIn('whatever', self.schema)

            cnx.create_entity('Note')

            cnx.commit()

            self.assertNotIn('shortpara', self.schema)

            mh.cmd_add_attribute('Note', 'shortpara')

            self.assertIn('shortpara', self.schema)

            self.assertEqual(self.schema['shortpara'].subjects(), ('Note', ))

            self.assertEqual(self.schema['shortpara'].objects(), ('String', ))

            # test created column is actually a varchar(64)

            fields = self.table_schema(mh, '%sNote' % SQL_PREFIX)

            self.assertEqual(fields['%sshortpara' % SQL_PREFIX], ('character varying', 11))

            # test default value set on existing entities

            self.assertEqual(cnx.execute('Note X').get_entity(0, 0).shortpara, 'hop')

            # test default value set for next entities

            self.assertEqual(cnx.create_entity('Note', shortpara=u'hop hop').shortpara, u'hop hop')

            # serialized constraint added

            constraints = self.table_constraints(mh, 'cw_Personne')

            self.assertEqual(len(constraints), 1, constraints)

    def test_add_attribute_unique(self):

        with self.mh() as (cnx, mh):

            self.assertNotIn('unique_id', self.schema)

            mh.cmd_add_attribute('Note', 'unique_id')

            # test unique index creation

            dbh = self.repo.system_source.dbhelper

            self.assertTrue(dbh.index_exists(cnx.cnxset.cu, 'cw_Note', 'cw_unique_id', unique=True))

    def test_add_datetime_with_default_value_attribute(self):

        with self.mh() as (cnx, mh):

            self.assertNotIn('mydate', self.schema)

            self.assertNotIn('oldstyledefaultdate', self.schema)

            self.assertNotIn('newstyledefaultdate', self.schema)

            mh.cmd_add_attribute('Note', 'mydate')

            mh.cmd_add_attribute('Note', 'oldstyledefaultdate')

            mh.cmd_add_attribute('Note', 'newstyledefaultdate')

            self.assertIn('mydate', self.schema)

            self.assertIn('oldstyledefaultdate', self.schema)

            self.assertIn('newstyledefaultdate', self.schema)

            self.assertEqual(self.schema['mydate'].subjects(), ('Note', ))

            self.assertEqual(self.schema['mydate'].objects(), ('Date', ))

            testdate = date(2005, 12, 13)

            eid1 = mh.rqlexec('INSERT Note N')[0][0]

            eid2 = mh.rqlexec('INSERT Note N: N mydate %(mydate)s', {'mydate': testdate})[0][0]

            d1 = mh.rqlexec('Any D WHERE X eid %(x)s, X mydate D', {'x': eid1})[0][0]

            d2 = mh.rqlexec('Any D WHERE X eid %(x)s, X mydate D', {'x': eid2})[0][0]

            d3 = mh.rqlexec('Any D WHERE X eid %(x)s, X oldstyledefaultdate D', {'x': eid1})[0][0]

            d4 = mh.rqlexec('Any D WHERE X eid %(x)s, X newstyledefaultdate D', {'x': eid1})[0][0]

            self.assertEqual(d1, date.today())

            self.assertEqual(d2, testdate)

            myfavoritedate = date(2013, 1, 1)

            self.assertEqual(d3, myfavoritedate)

            self.assertEqual(d4, myfavoritedate)

    def test_drop_chosen_constraints_ctxmanager(self):

        with self.mh() as (cnx, mh):

            with mh.cmd_dropped_constraints('Note', 'unique_id', UniqueConstraint):

                mh.cmd_add_attribute('Note', 'unique_id')

                # make sure the maxsize constraint is not dropped

                self.assertRaises(ValidationError,

                                  mh.rqlexec,

                                  'INSERT Note N: N unique_id "xyz"')

                mh.rollback()

                # make sure the unique constraint is dropped

                mh.rqlexec('INSERT Note N: N unique_id "x"')

                mh.rqlexec('INSERT Note N: N unique_id "x"')

                mh.rqlexec('DELETE Note N')

    def test_drop_required_ctxmanager(self):

        with self.mh() as (cnx, mh):

            with mh.cmd_dropped_constraints('Note', 'unique_id', cstrtype=None,

                                            droprequired=True):

                mh.cmd_add_attribute('Note', 'unique_id')

                mh.rqlexec('INSERT Note N')

                mh.rqlexec('SET N unique_id "x"')

            # make sure the required=True was restored

            self.assertRaises(ValidationError, mh.rqlexec, 'INSERT Note N')

            mh.rollback()

    def test_rename_attribute(self):

        with self.mh() as (cnx, mh):

            self.assertNotIn('civility', self.schema)

            eid1 = mh.rqlexec('INSERT Personne X: X nom "lui", X sexe "M"')[0][0]

            eid2 = mh.rqlexec('INSERT Personne X: X nom "l\'autre", X sexe NULL')[0][0]

            mh.cmd_rename_attribute('Personne', 'sexe', 'civility')

            self.assertNotIn('sexe', self.schema)

            self.assertIn('civility', self.schema)

            # test data has been backported

            c1 = mh.rqlexec('Any C WHERE X eid %s, X civility C' % eid1)[0][0]

            self.assertEqual(c1, 'M')

            c2 = mh.rqlexec('Any C WHERE X eid %s, X civility C' % eid2)[0][0]

            self.assertEqual(c2, None)

    def test_workflow_actions(self):

        with self.mh() as (cnx, mh):

            mh.cmd_add_workflow(u'foo', ('Personne', 'Email'),

                                ensure_workflowable=False)

            for etype in ('Personne', 'Email'):

                s1 = mh.rqlexec('Any N WHERE WF workflow_of ET, ET name "%s", WF name N' %

                                etype)[0][0]

                self.assertEqual(s1, "foo")

                s1 = mh.rqlexec('Any N WHERE ET default_workflow WF, ET name "%s", WF name N' %

                                etype)[0][0]

                self.assertEqual(s1, "foo")

    def test_add_entity_type(self):

        with self.mh() as (cnx, mh):

            self.assertNotIn('Folder2', self.schema)

            self.assertNotIn('filed_under2', self.schema)

            mh.cmd_add_entity_type('Folder2')

            self.assertIn('Folder2', self.schema)

            self.assertIn('Old', self.schema)

            self.assertTrue(cnx.execute('CWEType X WHERE X name "Folder2"'))

            self.assertIn('filed_under2', self.schema)

            self.assertTrue(cnx.execute('CWRType X WHERE X name "filed_under2"'))

            self.assertEqual(sorted(str(rs) for rs in self.schema['Folder2'].subject_relations()),

                             ['created_by', 'creation_date', 'cw_source', 'cwuri',

                              'description', 'description_format',

                              'eid',

                              'filed_under2', 'has_text',

                              'identity', 'in_basket', 'inlined_rel', 'is', 'is_instance_of',

                              'modification_date', 'name', 'owned_by'])

            self.assertCountEqual([str(rs) for rs in self.schema['Folder2'].object_relations()],

                                  ['filed_under2', 'identity', 'inlined_rel'])

            # Old will be missing as it has been renamed into 'New' in the migrated

            # schema while New hasn't been added here.

            self.assertEqual(sorted(str(e) for e in self.schema['filed_under2'].subjects()),

                             sorted(str(e) for e in self.schema.entities()

                                    if not e.final and e != 'Old'))

            self.assertEqual(self.schema['filed_under2'].objects(), ('Folder2',))

            eschema = self.schema.eschema('Folder2')

            for cstr in eschema.rdef('name').constraints:

                self.assertTrue(hasattr(cstr, 'eid'))

    def test_add_entity_type_with_constraint(self):

        with self.mh() as (cnx, mh):

            mh.cmd_add_entity_type('Activity')

            constraints = self.table_constraints(mh, 'cw_Activity')

            self.assertEqual(len(constraints), 2, constraints)

    def test_add_cube_with_custom_final_type(self):

        with self.mh() as (cnx, mh):

            try:

                mh.cmd_add_cube('fakecustomtype')

                self.assertIn('Numeric', self.schema)

                self.assertTrue(self.schema['Numeric'].final)

                rdef = self.schema['num'].rdefs[('Location', 'Numeric')]

                self.assertEqual(rdef.scale, 10)

                self.assertEqual(rdef.precision, 18)

                fields = self.table_schema(mh, '%sLocation' % SQL_PREFIX)

                self.assertEqual(fields['%snum' % SQL_PREFIX], ('numeric', None))  # XXX

            finally:

                mh.cmd_drop_cube('fakecustomtype')

                mh.drop_entity_type('Numeric')

    def test_add_drop_entity_type(self):

        with self.mh() as (cnx, mh):

            mh.cmd_add_entity_type('Folder2')

            wf = mh.cmd_add_workflow(u'folder2 wf', 'Folder2',