from contextlib import contextmanager

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

from cubicweb.server.migractions import ServerMigrationHelper

    @contextmanager

    def mh(self):

        with self.admin_access.client_cnx() as cnx:

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

                                             repo=self.repo, cnx=cnx,

                                             interactive=False)

        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)

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

            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)

        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)

            notesql = 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)')

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

        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)

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

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

            # make sure the required=True was restored

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

            mh.rollback()

        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)

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

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

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

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

                              ['filed_under2', 'identity'])

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

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

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

        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

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

            mh.cmd_drop_relation_type('filed_under2')

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

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

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

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

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

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

            mh.create_entity('Affaire')

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

            cnx.commit()

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

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

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

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

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

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

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

            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 = cnx.execute('Any MAX(X)')[0][0]

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

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