[js] add a selector string escaping function
will help use id strings in jquery selector expressions, e.g:
"foo.[subject]:42" -> "foo\.\[subject\]\:42"
Related to #3154531.
# copyright 2003-2013 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"""fromdatetimeimportdatefromos.pathimportjoinfromlogilab.common.testlibimportTestCase,unittest_main,Tags,tagfromyams.constraintsimportUniqueConstraintfromcubicwebimportConfigurationError,ValidationErrorfromcubicweb.devtools.testlibimportCubicWebTCfromcubicweb.schemaimportCubicWebSchemaLoaderfromcubicweb.server.sqlutilsimportSQL_PREFIXfromcubicweb.server.migractionsimport*importcubicweb.devtoolsmigrschema=NonedeftearDownModule(*args):globalmigrschemadelmigrschemaifhasattr(MigrationCommandsTC,'origschema'):delMigrationCommandsTC.origschemaclassMigrationCommandsTC(CubicWebTC):configcls=cubicweb.devtools.TestServerConfigurationtags=CubicWebTC.tags|Tags(('server','migration','migractions'))@classmethoddef_init_repo(cls):super(MigrationCommandsTC,cls)._init_repo()# we have to read schema from the database to get eid for schema entitiescls.repo.set_schema(cls.repo.deserialize_schema(),resetvreg=False)# hack to read the schema from data/migrschemaconfig=cls.configconfig.appid=join('data','migratedapp')config._apphome=cls.datapath('migratedapp')globalmigrschemamigrschema=config.load_schema()config.appid='data'config._apphome=cls.datadirassert'Folder'inmigrschemadefsetUp(self):CubicWebTC.setUp(self)self.mh=ServerMigrationHelper(self.repo.config,migrschema,repo=self.repo,cnx=self.cnx,interactive=False)assertself.cnxisself.mh._cnxassertself.sessionisself.mh.session,(self.session.id,self.mh.session.id)deftearDown(self):CubicWebTC.tearDown(self)self.repo.vreg['etypes'].clear_caches()deftest_add_attribute_int(self):self.assertFalse('whatever'inself.schema)self.session.create_entity('Note')self.session.commit(free_cnxset=False)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'inself.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 entitiesnote=self.session.execute('Note X').get_entity(0,0)self.assertEqual(note.whatever,2)# test default value set for next entitiesself.assertEqual(self.session.create_entity('Note').whatever,2)# test attribute orderorderdict2=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').orderfork,vinorderdict.iteritems():ifv>=whateverorder:orderdict[k]=v+1orderdict['whatever']=whateverorderself.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()deftest_add_attribute_varchar(self):self.assertFalse('whatever'inself.schema)self.session.create_entity('Note')self.session.commit(free_cnxset=False)self.assertFalse('shortpara'inself.schema)self.mh.cmd_add_attribute('Note','shortpara')self.assertTrue('shortpara'inself.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]forxinnotesql.split('(',1)[1].rsplit(')',1)[0].split(','))self.assertEqual(fields['%sshortpara'%SQL_PREFIX],'varchar(64)')# test default value set on existing entitiesself.assertEqual(self.session.execute('Note X').get_entity(0,0).shortpara,'hop')# test default value set for next entitiesself.assertEqual(self.session.create_entity('Note').shortpara,'hop')self.mh.rollback()deftest_add_datetime_with_default_value_attribute(self):self.assertFalse('mydate'inself.schema)self.assertFalse('shortpara'inself.schema)self.mh.cmd_add_attribute('Note','mydate')self.assertTrue('mydate'inself.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()deftest_drop_chosen_constraints_ctxmanager(self):withself.mh.cmd_dropped_constraints('Note','unique_id',UniqueConstraint):self.mh.cmd_add_attribute('Note','unique_id')# make sure the maxsize constraint is not droppedself.assertRaises(ValidationError,self.mh.rqlexec,'INSERT Note N: N unique_id "xyz"')self.mh.rollback()# make sure the unique constraint is droppedself.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()deftest_drop_required_ctxmanager(self):withself.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 restoredself.assertRaises(ValidationError,self.mh.rqlexec,'INSERT Note N')self.mh.rollback()deftest_rename_attribute(self):self.assertFalse('civility'inself.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'inself.schema)self.assertTrue('civility'inself.schema)# test data has been backportedc1=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)deftest_workflow_actions(self):wf=self.mh.cmd_add_workflow(u'foo',('Personne','Email'),ensure_workflowable=False)foretypein('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")deftest_add_entity_type(self):self.assertFalse('Folder2'inself.schema)self.assertFalse('filed_under2'inself.schema)self.mh.cmd_add_entity_type('Folder2')self.assertTrue('Folder2'inself.schema)self.assertTrue('Old'inself.schema)self.assertTrue(self.session.execute('CWEType X WHERE X name "Folder2"'))self.assertTrue('filed_under2'inself.schema)self.assertTrue(self.session.execute('CWRType X WHERE X name "filed_under2"'))self.schema.rebuild_infered_relations()self.assertEqual(sorted(str(rs)forrsinself.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)forrsinself.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)foreinself.schema['filed_under2'].subjects()),sorted(str(e)foreinself.schema.entities()ifnote.finalande!='Old'))self.assertEqual(self.schema['filed_under2'].objects(),('Folder2',))eschema=self.schema.eschema('Folder2')forcstrineschema.rdef('name').constraints:self.assertTrue(hasattr(cstr,'eid'))deftest_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.session.commit(free_cnxset=False)eschema=self.schema.eschema('Folder2')self.mh.cmd_drop_entity_type('Folder2')self.assertFalse('Folder2'inself.schema)self.assertFalse(self.session.execute('CWEType X WHERE X name "Folder2"'))# test automatic workflow deletionself.assertFalse(self.session.execute('Workflow X WHERE NOT X workflow_of ET'))self.assertFalse(self.session.execute('State X WHERE NOT X state_of WF'))self.assertFalse(self.session.execute('Transition X WHERE NOT X transition_of WF'))deftest_rename_entity_type(self):entity=self.mh.create_entity('Old',name=u'old')self.repo.type_and_source_from_eid(entity.eid)self.mh.cmd_rename_entity_type('Old','New')self.mh.cmd_rename_attribute('New','name','new_name')deftest_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'inself.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)foreinself.schema['filed_under2'].subjects()),sorted(str(e)foreinself.schema.entities()ifnote.finalande!='Old'))self.assertEqual(self.schema['filed_under2'].objects(),('Folder2',))self.mh.cmd_drop_relation_type('filed_under2')self.assertFalse('filed_under2'inself.schema)deftest_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.session.commit(free_cnxset=False)self.mh.cmd_drop_relation_definition('Personne','concerne2','Affaire')self.assertTrue('concerne2'inself.schema)self.mh.cmd_drop_relation_definition('Personne','concerne2','Note')self.assertFalse('concerne2'inself.schema)deftest_drop_relation_definition_existant_rtype(self):self.assertEqual(sorted(str(e)foreinself.schema['concerne'].subjects()),['Affaire','Personne'])self.assertEqual(sorted(str(e)foreinself.schema['concerne'].objects()),['Affaire','Division','Note','Societe','SubDivision'])self.mh.cmd_drop_relation_definition('Personne','concerne','Affaire')self.assertEqual(sorted(str(e)foreinself.schema['concerne'].subjects()),['Affaire'])self.assertEqual(sorted(str(e)foreinself.schema['concerne'].objects()),['Division','Note','Societe','SubDivision'])self.mh.cmd_add_relation_definition('Personne','concerne','Affaire')self.assertEqual(sorted(str(e)foreinself.schema['concerne'].subjects()),['Affaire','Personne'])self.assertEqual(sorted(str(e)foreinself.schema['concerne'].objects()),['Affaire','Division','Note','Societe','SubDivision'])# trick: overwrite self.maxeid to avoid deletion of just reintroduced typesself.maxeid=self.session.execute('Any MAX(X)')[0][0]deftest_drop_relation_definition_with_specialization(self):self.assertEqual(sorted(str(e)foreinself.schema['concerne'].subjects()),['Affaire','Personne'])self.assertEqual(sorted(str(e)foreinself.schema['concerne'].objects()),['Affaire','Division','Note','Societe','SubDivision'])self.mh.cmd_drop_relation_definition('Affaire','concerne','Societe')self.assertEqual(sorted(str(e)foreinself.schema['concerne'].subjects()),['Affaire','Personne'])self.assertEqual(sorted(str(e)foreinself.schema['concerne'].objects()),['Affaire','Division','Note','SubDivision'])self.schema.rebuild_infered_relations()# need to be explicitly called once everything is in placeself.assertEqual(sorted(str(e)foreinself.schema['concerne'].objects()),['Affaire','Note'])self.mh.cmd_add_relation_definition('Affaire','concerne','Societe')self.assertEqual(sorted(str(e)foreinself.schema['concerne'].subjects()),['Affaire','Personne'])self.assertEqual(sorted(str(e)foreinself.schema['concerne'].objects()),['Affaire','Note','Societe'])self.schema.rebuild_infered_relations()# need to be explicitly called once everything is in placeself.assertEqual(sorted(str(e)foreinself.schema['concerne'].objects()),['Affaire','Division','Note','Societe','SubDivision'])# trick: overwrite self.maxeid to avoid deletion of just reintroduced typesself.maxeid=self.session.execute('Any MAX(X)')[0][0]deftest_rename_relation(self):self.skipTest('implement me')deftest_change_relation_props_non_final(self):rschema=self.schema['concerne']card=rschema.rdef('Affaire','Societe').cardinalityself.assertEqual(card,'**')try:self.mh.cmd_change_relation_props('Affaire','concerne','Societe',cardinality='?*')card=rschema.rdef('Affaire','Societe').cardinalityself.assertEqual(card,'?*')finally:self.mh.cmd_change_relation_props('Affaire','concerne','Societe',cardinality='**')deftest_change_relation_props_final(self):rschema=self.schema['adel']card=rschema.rdef('Personne','String').fulltextindexedself.assertEqual(card,False)try:self.mh.cmd_change_relation_props('Personne','adel','String',fulltextindexed=True)card=rschema.rdef('Personne','String').fulltextindexedself.assertEqual(card,True)finally:self.mh.cmd_change_relation_props('Personne','adel','String',fulltextindexed=False)@tag('longrun')deftest_sync_schema_props_perms(self):cursor=self.mh.sessioncursor.set_cnxset()nbrqlexpr_start=cursor.execute('Any COUNT(X) WHERE X is RQLExpression')[0][0]migrschema['titre'].rdefs[('Personne','String')].order=7migrschema['adel'].rdefs[('Personne','String')].order=6migrschema['ass'].rdefs[('Personne','String')].order=5migrschema['Personne'].description='blabla bla'migrschema['titre'].description='usually a title'migrschema['titre'].rdefs[('Personne','String')].description='title for this person'delete_concerne_rqlexpr=self._rrqlexpr_rset('delete','concerne')add_concerne_rqlexpr=self._rrqlexpr_rset('add','concerne')self.mh.cmd_sync_schema_props_perms(commit=False)self.assertEqual(cursor.execute('Any D WHERE X name "Personne", X description D')[0][0],'blabla bla')self.assertEqual(cursor.execute('Any D WHERE X name "titre", X description D')[0][0],'usually a title')self.assertEqual(cursor.execute('Any D WHERE X relation_type RT, RT name "titre",''X from_entity FE, FE name "Personne",''X description D')[0][0],'title for this person')rinorder=[nforn,incursor.execute('Any N ORDERBY O,N WHERE X is CWAttribute, X relation_type RT, RT name N,''X from_entity FE, FE name "Personne",''X ordernum O')]expected=[u'nom',u'prenom',u'sexe',u'promo',u'ass',u'adel',u'titre',u'web',u'tel',u'fax',u'datenaiss',u'test',u'tzdatenaiss',u'description',u'firstname',u'creation_date',u'cwuri',u'modification_date']self.assertEqual(expected,rinorder)# test permissions synchronization ##################################### new rql expr to add note entityeexpr=self._erqlexpr_entity('add','Note')self.assertEqual(eexpr.expression,'X ecrit_part PE, U in_group G, ''PE require_permission P, P name "add_note", P require_group G')self.assertEqual([et.nameforetineexpr.reverse_add_permission],['Note'])self.assertEqual(eexpr.reverse_read_permission,())self.assertEqual(eexpr.reverse_delete_permission,())self.assertEqual(eexpr.reverse_update_permission,())# no more rqlexpr to delete and add para attributeself.assertFalse(self._rrqlexpr_rset('add','para'))self.assertFalse(self._rrqlexpr_rset('delete','para'))# new rql expr to add ecrit_par relationrexpr=self._rrqlexpr_entity('add','ecrit_par')self.assertEqual(rexpr.expression,'O require_permission P, P name "add_note", ''U in_group G, P require_group G')self.assertEqual([rdef.rtype.nameforrdefinrexpr.reverse_add_permission],['ecrit_par'])self.assertEqual(rexpr.reverse_read_permission,())self.assertEqual(rexpr.reverse_delete_permission,())# no more rqlexpr to delete and add travaille relationself.assertFalse(self._rrqlexpr_rset('add','travaille'))self.assertFalse(self._rrqlexpr_rset('delete','travaille'))# no more rqlexpr to delete and update Societe entityself.assertFalse(self._erqlexpr_rset('update','Societe'))self.assertFalse(self._erqlexpr_rset('delete','Societe'))# no more rqlexpr to read Affaire entityself.assertFalse(self._erqlexpr_rset('read','Affaire'))# rqlexpr to update Affaire entity has been updatedeexpr=self._erqlexpr_entity('update','Affaire')self.assertEqual(eexpr.expression,'X concerne S, S owned_by U')# no change for rqlexpr to add and delete Affaire entityself.assertEqual(len(self._erqlexpr_rset('delete','Affaire')),1)self.assertEqual(len(self._erqlexpr_rset('add','Affaire')),1)# no change for rqlexpr to add and delete concerne relationself.assertEqual(len(self._rrqlexpr_rset('delete','concerne')),len(delete_concerne_rqlexpr))self.assertEqual(len(self._rrqlexpr_rset('add','concerne')),len(add_concerne_rqlexpr))# * migrschema involve:# * 7 rqlexprs deletion (2 in (Affaire read + Societe + travaille) + 1# in para attribute)# * 1 update (Affaire update)# * 2 new (Note add, ecrit_par add)# * 2 implicit new for attributes update_permission (Note.para, Personne.test)# remaining orphan rql expr which should be deleted at commit (composite relation)self.assertEqual(cursor.execute('Any COUNT(X) WHERE X is RQLExpression, ''NOT ET1 read_permission X, NOT ET2 add_permission X, ''NOT ET3 delete_permission X, NOT ET4 update_permission X')[0][0],7+1)# finallyself.assertEqual(cursor.execute('Any COUNT(X) WHERE X is RQLExpression')[0][0],nbrqlexpr_start+1+2+2)self.mh.commit()# unique_together testself.assertEqual(len(self.schema.eschema('Personne')._unique_together),1)self.assertCountEqual(self.schema.eschema('Personne')._unique_together[0],('nom','prenom','datenaiss'))rset=cursor.execute('Any C WHERE C is CWUniqueTogetherConstraint, C constraint_of ET, ET name "Personne"')self.assertEqual(len(rset),1)relations=[r.nameforrinrset.get_entity(0,0).relations]self.assertCountEqual(relations,('nom','prenom','datenaiss'))def_erqlexpr_rset(self,action,ertype):rql='RQLExpression X WHERE ET is CWEType, ET %s_permission X, ET name %%(name)s'%actionreturnself.mh.session.execute(rql,{'name':ertype})def_erqlexpr_entity(self,action,ertype):rset=self._erqlexpr_rset(action,ertype)self.assertEqual(len(rset),1)returnrset.get_entity(0,0)def_rrqlexpr_rset(self,action,ertype):rql='RQLExpression X WHERE RT is CWRType, RDEF %s_permission X, RT name %%(name)s, RDEF relation_type RT'%actionreturnself.mh.session.execute(rql,{'name':ertype})def_rrqlexpr_entity(self,action,ertype):rset=self._rrqlexpr_rset(action,ertype)self.assertEqual(len(rset),1)returnrset.get_entity(0,0)deftest_set_size_constraint(self):# existing previous valuetry:self.mh.cmd_set_size_constraint('CWEType','name',128)finally:self.mh.cmd_set_size_constraint('CWEType','name',64)# non existing previous valuetry:self.mh.cmd_set_size_constraint('CWEType','description',256)finally:self.mh.cmd_set_size_constraint('CWEType','description',None)@tag('longrun')deftest_add_remove_cube_and_deps(self):cubes=set(self.config.cubes())schema=self.repo.schemaself.assertEqual(sorted((str(s),str(o))fors,oinschema['see_also'].rdefs.iterkeys()),sorted([('EmailThread','EmailThread'),('Folder','Folder'),('Bookmark','Bookmark'),('Bookmark','Note'),('Note','Note'),('Note','Bookmark')]))try:try:self.mh.cmd_remove_cube('email',removedeps=True)# file was there because it's an email dependancy, should have been removedself.assertFalse('email'inself.config.cubes())self.assertFalse(self.config.cube_dir('email')inself.config.cubes_path())self.assertFalse('file'inself.config.cubes())self.assertFalse(self.config.cube_dir('file')inself.config.cubes_path())forertypein('Email','EmailThread','EmailPart','File','sender','in_thread','reply_to','data_format'):self.assertFalse(ertypeinschema,ertype)self.assertEqual(sorted(schema['see_also'].rdefs.iterkeys()),sorted([('Folder','Folder'),('Bookmark','Bookmark'),('Bookmark','Note'),('Note','Note'),('Note','Bookmark')]))self.assertEqual(sorted(schema['see_also'].subjects()),['Bookmark','Folder','Note'])self.assertEqual(sorted(schema['see_also'].objects()),['Bookmark','Folder','Note'])self.assertEqual(self.session.execute('Any X WHERE X pkey "system.version.email"').rowcount,0)self.assertEqual(self.session.execute('Any X WHERE X pkey "system.version.file"').rowcount,0)except:importtracebacktraceback.print_exc()raisefinally:self.mh.cmd_add_cube('email')self.assertTrue('email'inself.config.cubes())self.assertTrue(self.config.cube_dir('email')inself.config.cubes_path())self.assertTrue('file'inself.config.cubes())self.assertTrue(self.config.cube_dir('file')inself.config.cubes_path())forertypein('Email','EmailThread','EmailPart','File','sender','in_thread','reply_to','data_format'):self.assertTrue(ertypeinschema,ertype)self.assertEqual(sorted(schema['see_also'].rdefs.iterkeys()),sorted([('EmailThread','EmailThread'),('Folder','Folder'),('Bookmark','Bookmark'),('Bookmark','Note'),('Note','Note'),('Note','Bookmark')]))self.assertEqual(sorted(schema['see_also'].subjects()),['Bookmark','EmailThread','Folder','Note'])self.assertEqual(sorted(schema['see_also'].objects()),['Bookmark','EmailThread','Folder','Note'])fromcubes.email.__pkginfo__importversionasemail_versionfromcubes.file.__pkginfo__importversionasfile_versionself.assertEqual(self.session.execute('Any V WHERE X value V, X pkey "system.version.email"')[0][0],email_version)self.assertEqual(self.session.execute('Any V WHERE X value V, X pkey "system.version.file"')[0][0],file_version)# trick: overwrite self.maxeid to avoid deletion of just reintroduced# types (and their associated tables!)self.maxeid=self.session.execute('Any MAX(X)')[0][0]# why this commit is necessary is unclear to me (though without it# next test may fail complaining of missing tablesself.session.commit(free_cnxset=False)@tag('longrun')deftest_add_remove_cube_no_deps(self):cubes=set(self.config.cubes())schema=self.repo.schematry:try:self.mh.cmd_remove_cube('email')cubes.remove('email')self.assertFalse('email'inself.config.cubes())self.assertTrue('file'inself.config.cubes())forertypein('Email','EmailThread','EmailPart','sender','in_thread','reply_to'):self.assertFalse(ertypeinschema,ertype)except:importtracebacktraceback.print_exc()raisefinally:self.mh.cmd_add_cube('email')self.assertTrue('email'inself.config.cubes())# trick: overwrite self.maxeid to avoid deletion of just reintroduced# types (and their associated tables!)self.maxeid=self.session.execute('Any MAX(X)')[0][0]# why this commit is necessary is unclear to me (though without it# next test may fail complaining of missing tablesself.session.commit(free_cnxset=False)deftest_remove_dep_cube(self):withself.assertRaises(ConfigurationError)ascm:self.mh.cmd_remove_cube('file')self.assertEqual(str(cm.exception),"can't remove cube file, used as a dependency")@tag('longrun')deftest_introduce_base_class(self):self.mh.cmd_add_entity_type('Para')self.mh.repo.schema.rebuild_infered_relations()self.assertEqual(sorted(et.typeforetinself.schema['Para'].specialized_by()),['Note'])self.assertEqual(self.schema['Note'].specializes().type,'Para')self.mh.cmd_add_entity_type('Text')self.mh.repo.schema.rebuild_infered_relations()self.assertEqual(sorted(et.typeforetinself.schema['Para'].specialized_by()),['Note','Text'])self.assertEqual(self.schema['Text'].specializes().type,'Para')# test columns have been actually addedtext=self.session.execute('INSERT Text X: X para "hip", X summary "hop", X newattr "momo"').get_entity(0,0)note=self.session.execute('INSERT Note X: X para "hip", X shortpara "hop", X newattr "momo", X unique_id "x"').get_entity(0,0)aff=self.session.execute('INSERT Affaire X').get_entity(0,0)self.assertTrue(self.session.execute('SET X newnotinlined Y WHERE X eid %(x)s, Y eid %(y)s',{'x':text.eid,'y':aff.eid}))self.assertTrue(self.session.execute('SET X newnotinlined Y WHERE X eid %(x)s, Y eid %(y)s',{'x':note.eid,'y':aff.eid}))self.assertTrue(self.session.execute('SET X newinlined Y WHERE X eid %(x)s, Y eid %(y)s',{'x':text.eid,'y':aff.eid}))self.assertTrue(self.session.execute('SET X newinlined Y WHERE X eid %(x)s, Y eid %(y)s',{'x':note.eid,'y':aff.eid}))# XXX remove specializes by ourselves, else tearDown fails when removing# Para because of Note inheritance. This could be fixed by putting the# MemSchemaCWETypeDel(session, name) operation in the# after_delete_entity(CWEType) hook, since in that case the MemSchemaSpecializesDel# operation would be removed before, but I'm not sure this is a desired behaviour.## also we need more tests about introducing/removing base classes or# specialization relationship...self.session.data['rebuild-infered']=Truetry:self.session.execute('DELETE X specializes Y WHERE Y name "Para"')self.session.commit(free_cnxset=False)finally:self.session.data['rebuild-infered']=Falseself.assertEqual(sorted(et.typeforetinself.schema['Para'].specialized_by()),[])self.assertEqual(self.schema['Note'].specializes(),None)self.assertEqual(self.schema['Text'].specializes(),None)deftest_add_symmetric_relation_type(self):same_as_sql=self.mh.sqlexec("SELECT sql FROM sqlite_master WHERE type='table' ""and name='same_as_relation'")self.assertFalse(same_as_sql)self.mh.cmd_add_relation_type('same_as')same_as_sql=self.mh.sqlexec("SELECT sql FROM sqlite_master WHERE type='table' ""and name='same_as_relation'")self.assertTrue(same_as_sql)if__name__=='__main__':unittest_main()