# copyright 2003-2010 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 __future__ import with_statement

from copy import deepcopy

from datetime import date

from os.path import join

from logilab.common.testlib import TestCase, unittest_main, Tags, tag

from yams.constraints import UniqueConstraint

from cubicweb import ConfigurationError, ValidationError

from cubicweb.devtools.testlib import CubicWebTC

from cubicweb.schema import CubicWebSchemaLoader

from cubicweb.server.sqlutils import SQL_PREFIX

from cubicweb.server.migractions import *

migrschema = None

def tearDownModule(*args):

    global migrschema

    del migrschema

    if hasattr(MigrationCommandsTC, 'origschema'):

        del MigrationCommandsTC.origschema

class MigrationCommandsTC(CubicWebTC):

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

    @classmethod

    def _init_repo(cls):

        super(MigrationCommandsTC, cls)._init_repo()

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

        config = cls.config

        config._cubes = None

        cls.repo.fill_schema()

        cls.origschema = deepcopy(cls.repo.schema)

        # hack to read the schema from data/migrschema

        config.appid = join('data', 'migratedapp')

        config._apphome = cls.datapath('migratedapp')

        global migrschema

        migrschema = config.load_schema()

        config.appid = 'data'

        config._apphome = cls.datadir

        assert 'Folder' in migrschema

    def setUp(self):

        CubicWebTC.setUp(self)

        self.mh = ServerMigrationHelper(self.repo.config, migrschema,

                                        repo=self.repo, cnx=self.cnx,

                                        interactive=False)

        assert self.cnx is self.mh._cnx

        assert self.session is self.mh.session, (self.session.id, self.mh.session.id)

    def tearDown(self):

        CubicWebTC.tearDown(self)

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

    def test_add_attribute_int(self):

        self.assertFalse('whatever' in self.schema)

        self.request().create_entity('Note')

        self.commit()

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

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

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

        self.assertTrue('whatever' in self.schema)

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

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

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

        # test default value set on existing entities

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

        self.assertEqual(note.whatever, 2)

        # test default value set for next entities

        self.assertEqual(self.request().create_entity('Note').whatever, 2)

        # test attribute order

        orderdict2 = dict(self.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.iteritems():

            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)

        self.mh.rollback()

    def test_add_attribute_varchar(self):

        self.assertFalse('whatever' in self.schema)

        self.request().create_entity('Note')

        self.commit()

        self.assertFalse('shortpara' in self.schema)

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

        self.assertTrue('shortpara' in self.schema)

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

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

        # test created column is actually a varchar(64)

        notesql = self.mh.sqlexec("SELECT sql FROM sqlite_master WHERE type='table' and name='%sNote'" % SQL_PREFIX)[0][0]

        fields = dict(x.strip().split()[:2] for x in notesql.split('(', 1)[1].rsplit(')', 1)[0].split(','))

        self.assertEqual(fields['%sshortpara' % SQL_PREFIX], 'varchar(64)')

        req = self.request()

        # test default value set on existing entities

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

        # test default value set for next entities

        self.assertEqual(req.create_entity('Note').shortpara, 'hop')

        self.mh.rollback()

    def test_add_datetime_with_default_value_attribute(self):

        self.assertFalse('mydate' in self.schema)

        self.assertFalse('shortpara' in self.schema)

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

        self.assertTrue('mydate' in self.schema)

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

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

        testdate = date(2005, 12, 13)

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

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

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

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

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

        self.assertEqual(d2, testdate)

        self.mh.rollback()

    def test_drop_chosen_constraints_ctxmanager(self):

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

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

            # make sure the maxsize constraint is not dropped

            self.assertRaises(ValidationError,

                              self.mh.rqlexec,

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

            self.mh.rollback()

            # make sure the unique constraint is dropped

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

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

            self.mh.rqlexec('DELETE Note N')

        self.mh.rollback()

    def test_drop_required_ctxmanager(self):

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

                                             droprequired=True):

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

            self.mh.rqlexec('INSERT Note N')

        # make sure the required=True was restored

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

        self.mh.rollback()

    def test_rename_attribute(self):

        self.assertFalse('civility' in self.schema)

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

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

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

        self.assertFalse('sexe' in self.schema)

        self.assertTrue('civility' in self.schema)

        # test data has been backported

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

        self.assertEqual(c1, 'M')

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

        self.assertEqual(c2, None)

    def test_workflow_actions(self):

        wf = self.mh.cmd_add_workflow(u'foo', ('Personne', 'Email'),

                                      ensure_workflowable=False)

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

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

                                 etype)[0][0]

            self.assertEqual(s1, "foo")

            s1 = self.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):

        self.assertFalse('Folder2' in self.schema)

        self.assertFalse('filed_under2' in self.schema)

        self.mh.cmd_add_entity_type('Folder2')

        self.assertTrue('Folder2' in self.schema)

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

        self.assertTrue('filed_under2' in self.schema)

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

        self.schema.rebuild_infered_relations()

        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', 'is', 'is_instance_of',

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

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

                          ['filed_under2', 'identity'])

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

        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_drop_entity_type(self):

        self.mh.cmd_add_entity_type('Folder2')

        wf = self.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)

        self.commit()

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

        self.mh.cmd_drop_entity_type('Folder2')

        self.assertFalse('Folder2' in self.schema)

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

        # test automatic workflow deletion

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

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

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

    def test_add_drop_relation_type(self):

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

        self.mh.cmd_add_relation_type('filed_under2')

        self.schema.rebuild_infered_relations()

        self.assertTrue('filed_under2' in self.schema)

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

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

        self.mh.cmd_drop_relation_type('filed_under2')

        self.assertFalse('filed_under2' in self.schema)

    def test_add_relation_definition_nortype(self):

        self.mh.cmd_add_relation_definition('Personne', 'concerne2', 'Affaire')

        self.assertEqual(self.schema['concerne2'].subjects(),

                          ('Personne',))

        self.assertEqual(self.schema['concerne2'].objects(),

                          ('Affaire', ))

        self.assertEqual(self.schema['concerne2'].rdef('Personne', 'Affaire').cardinality,

                          '1*')

        self.mh.cmd_add_relation_definition('Personne', 'concerne2', 'Note')

        self.assertEqual(sorted(self.schema['concerne2'].objects()), ['Affaire', 'Note'])

        self.mh.create_entity('Personne', nom=u'tot')

        self.mh.create_entity('Affaire')

        self.mh.rqlexec('SET X concerne2 Y WHERE X is Personne, Y is Affaire')

        self.commit()

        self.mh.cmd_drop_relation_definition('Personne', 'concerne2', 'Affaire')

        self.assertTrue('concerne2' in self.schema)

        self.mh.cmd_drop_relation_definition('Personne', 'concerne2', 'Note')

        self.assertFalse('concerne2' in self.schema)

    def test_drop_relation_definition_existant_rtype(self):

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

                          ['Affaire', 'Personne'])

        self.assertEqual(sorted(str(e) for e in self.schema['concerne'].objects()),

                          ['Affaire', 'Division', 'Note', 'Societe', 'SubDivision'])

        self.mh.cmd_drop_relation_definition('Personne', 'concerne', 'Affaire')

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

                          ['Affaire'])

        self.assertEqual(sorted(str(e) for e in self.schema['concerne'].objects()),

                          ['Division', 'Note', 'Societe', 'SubDivision'])

        self.mh.cmd_add_relation_definition('Personne', 'concerne', 'Affaire')

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

                          ['Affaire', 'Personne'])

        self.assertEqual(sorted(str(e) for e in self.schema['concerne'].objects()),

                          ['Affaire', 'Division', 'Note', 'Societe', 'SubDivision'])

        # trick: overwrite self.maxeid to avoid deletion of just reintroduced types

        self.maxeid = self.execute('Any MAX(X)')[0][0]

    def test_drop_relation_definition_with_specialization(self):

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

                          ['Affaire', 'Personne'])

        self.assertEqual(sorted(str(e) for e in self.schema['concerne'].objects()),

                          ['Affaire', 'Division', 'Note', 'Societe', 'SubDivision'])

        self.mh.cmd_drop_relation_definition('Affaire', 'concerne', 'Societe')

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

                          ['Affaire', 'Personne'])

        self.assertEqual(sorted(str(e) for e in self.schema['concerne'].objects()),

                          ['Affaire', 'Division', 'Note', 'SubDivision'])

        self.schema.rebuild_infered_relations() # need to be explicitly called once everything is in place

        self.assertEqual(sorted(str(e) for e in self.schema['concerne'].objects()),

                          ['Affaire', 'Note'])

        self.mh.cmd_add_relation_definition('Affaire', 'concerne', 'Societe')

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

                          ['Affaire', 'Personne'])

        self.assertEqual(sorted(str(e) for e in self.schema['concerne'].objects()),

                          ['Affaire', 'Note', 'Societe'])

        self.schema.rebuild_infered_relations() # need to be explicitly called once everything is in place

        self.assertEqual(sorted(str(e) for e in self.schema['concerne'].objects()),

                          ['Affaire', 'Division', 'Note', 'Societe', 'SubDivision'])

        # trick: overwrite self.maxeid to avoid deletion of just reintroduced types

        self.maxeid = self.execute('Any MAX(X)')[0][0]

    def test_rename_relation(self):

        self.skipTest('implement me')

    def test_change_relation_props_non_final(self):

        rschema = self.schema['concerne']

        card = rschema.rdef('Affaire', 'Societe').cardinality

        self.assertEqual(card, '**')

        try:

            self.mh.cmd_change_relation_props('Affaire', 'concerne', 'Societe',

                                              cardinality='?*')

            card = rschema.rdef('Affaire', 'Societe').cardinality

            self.assertEqual(card, '?*')

        finally:

            self.mh.cmd_change_relation_props('Affaire', 'concerne', 'Societe',

                                              cardinality='**')

    def test_change_relation_props_final(self):