# copyright 2003-2014 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"""

from datetime import date

import os, os.path as osp

from contextlib import contextmanager

from logilab.common import tempattr

from yams.constraints import UniqueConstraint

from cubicweb.devtools import startpgcluster, stoppgcluster

from cubicweb.devtools.testlib import CubicWebTC

from cubicweb.server.sqlutils import SQL_PREFIX

from cubicweb.server.migractions import ServerMigrationHelper

import cubicweb.devtools

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

def setUpModule():

    startpgcluster(__file__)

migrschema = None

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

    CUBES_PATH = cubicweb.devtools.TestServerConfiguration.CUBES_PATH + [

        osp.join(HERE, 'data-migractions', 'cubes')]

class MigrationTC(CubicWebTC):

    appid = 'data-migractions'

    configcls = MigrationConfig

    tags = CubicWebTC.tags | Tags(('server', 'migration', 'migractions'))

    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

        config = self.config

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

        config._apphome = osp.join(HERE, config.appid)

        global migrschema

        migrschema = config.load_schema()

        config.appid = self.appid

        config._apphome = osp.join(HERE, self.appid)

    def setUp(self):

        self.configcls.cls_adjust_sys_path()

        super(MigrationTC, self).setUp()

    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)

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)

            #self.assertEqual([r.type for r in self.schema['Note'].ordered_relations()],

            #                  ['modification_date', 'creation_date', 'owned_by',

            #                   'eid', 'ecrit_par', 'inline1', 'date', 'type',

            #                   'whatever', 'date', 'in_basket'])

            # NB: commit instead of rollback make following test fail with py2.5

            #     this sounds like a pysqlite/2.5 bug (the same eid is affected to

            #     two different entities)

    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', 64))

            # 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, 'hop')

    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):

            wf = 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_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',

                                     ensure_workflowable=False)

            todo = wf.add_state(u'todo', initial=True)

            done = wf.add_state(u'done')

            wf.add_transition(u'redoit', done, todo)

            wf.add_transition(u'markasdone', todo, done)

            cnx.commit()

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

            mh.cmd_drop_entity_type('Folder2')

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

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

            # test automatic workflow deletion

            self.assertFalse(cnx.execute('Workflow X WHERE NOT X workflow_of ET'))

            self.assertFalse(cnx.execute('State X WHERE NOT X state_of WF'))

            self.assertFalse(cnx.execute('Transition X WHERE NOT X transition_of WF'))

    def test_rename_entity_type(self):

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

            entity = mh.create_entity('Old', name=u'old')

            self.repo.type_and_source_from_eid(entity.eid, entity._cw)

            mh.cmd_rename_entity_type('Old', 'New')

            mh.cmd_rename_attribute('New', 'name', 'new_name')

    def test_add_drop_relation_type(self):

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

            mh.cmd_add_entity_type('Folder2', auto=False)

            mh.cmd_add_relation_type('filed_under2')

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

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