fromos.pathimportdirname,join,abspathfromlogilab.common.decoratorsimportcachedfrommx.DateTimeimportnowfromcubicweb.devtoolsimportTestServerConfiguration,init_test_databasefromcubicweb.devtools.apptestimportRepositoryBasedTCfromcubicweb.devtools.repotestimportdo_monkey_patch,undo_monkey_patchclassTwoSourcesConfiguration(TestServerConfiguration):sourcefile='sources_multi'classExternalSource1Configuration(TestServerConfiguration):sourcefile='sources_extern'classExternalSource2Configuration(TestServerConfiguration):sourcefile='sources_multi2'repo2,cnx2=init_test_database('sqlite',config=ExternalSource1Configuration('data'))cu=cnx2.cursor()ec1=cu.execute('INSERT Card X: X title "C3: An external card", X wikiid "aaa"')[0][0]cu.execute('INSERT Card X: X title "C4: Ze external card", X wikiid "zzz"')aff1=cu.execute('INSERT Affaire X: X ref "AFFREF", X in_state S WHERE S name "pitetre"')[0][0]cnx2.commit()MTIME=now()-0.1repo3,cnx3=init_test_database('sqlite',config=ExternalSource2Configuration('data'))# XXX, access existing connection, no pyro connectionfromcubicweb.server.sources.pyrorqlimportPyroRQLSourcePyroRQLSource.get_connection=lambdax:x.uri=='extern-multi'andcnx3orcnx2# necessary since the repository is closing its initial connections pool though# we want to keep cnx2 validfromcubicweb.dbapiimportConnectionConnection.close=lambdax:NoneclassTwoSourcesTC(RepositoryBasedTC):repo_config=TwoSourcesConfiguration('data')defsetUp(self):RepositoryBasedTC.setUp(self)# trigger discoveryself.execute('Card X')self.execute('Affaire X')self.execute('State X')self.commit()# don't delete external entities!self.maxeid=self.session.system_sql('SELECT MAX(eid) FROM entities').fetchone()[0]# add some entitiesself.ic1=self.execute('INSERT Card X: X title "C1: An internal card", X wikiid "aaai"')[0][0]self.ic2=self.execute('INSERT Card X: X title "C2: Ze internal card", X wikiid "zzzi"')[0][0]self.commit()do_monkey_patch()deftearDown(self):RepositoryBasedTC.tearDown(self)undo_monkey_patch()deftest_eid_comp(self):rset=self.execute('Card X WHERE X eid > 1')self.assertEquals(len(rset),4)rset=self.execute('Any X,T WHERE X title T, X eid > 1')self.assertEquals(len(rset),4)deftest_metainformation(self):rset=self.execute('Card X ORDERBY T WHERE X title T')# 2 added to the system source, 2 added to the external sourceself.assertEquals(len(rset),4)# since they are orderd by eid, we know the 3 first one is coming from the system source# and the others from external sourceself.assertEquals(rset.get_entity(0,0).metainformation(),{'source':{'adapter':'native','uri':'system'},'type':u'Card','extid':None})externent=rset.get_entity(3,0)metainf=externent.metainformation()self.assertEquals(metainf['source'],{'adapter':'pyrorql','uri':'extern'})self.assertEquals(metainf['type'],'Card')self.assert_(metainf['extid'])etype=self.execute('Any ETN WHERE X is ET, ET name ETN, X eid %(x)s',{'x':externent.eid},'x')[0][0]self.assertEquals(etype,'Card')deftest_order_limit_offset(self):rsetbase=self.execute('Any W,X ORDERBY W,X WHERE X wikiid W')self.assertEquals(len(rsetbase),4)self.assertEquals(sorted(rsetbase.rows),rsetbase.rows)rset=self.execute('Any W,X ORDERBY W,X LIMIT 2 OFFSET 2 WHERE X wikiid W')self.assertEquals(rset.rows,rsetbase.rows[2:4])deftest_has_text(self):self.repo.sources_by_uri['extern'].synchronize(MTIME)# in case fti_update has been run beforeself.failUnless(self.execute('Any X WHERE X has_text "affref"'))self.failUnless(self.execute('Affaire X WHERE X has_text "affref"'))deftest_anon_has_text(self):self.repo.sources_by_uri['extern'].synchronize(MTIME)# in case fti_update has been run beforeself.execute('INSERT Affaire X: X ref "no readable card"')[0][0]aff1=self.execute('INSERT Affaire X: X ref "card"')[0][0]# grant read accessself.execute('SET X owned_by U WHERE X eid %(x)s, U login "anon"',{'x':aff1},'x')self.commit()cnx=self.login('anon')cu=cnx.cursor()rset=cu.execute('Any X WHERE X has_text "card"')self.assertEquals(len(rset),5,zip(rset.rows,rset.description))deftest_synchronization(self):cu=cnx2.cursor()cu.execute('SET X ref "BLAH" WHERE X eid %(x)s',{'x':aff1},'x')aff2=cu.execute('INSERT Affaire X: X ref "AFFREUX", X in_state S WHERE S name "pitetre"')[0][0]cnx2.commit()try:# force syncself.repo.sources_by_uri['extern'].synchronize(MTIME)self.failUnless(self.execute('Any X WHERE X has_text "blah"'))self.failUnless(self.execute('Any X WHERE X has_text "affreux"'))cu.execute('DELETE Affaire X WHERE X eid %(x)s',{'x':aff2})cnx2.commit()self.repo.sources_by_uri['extern'].synchronize(MTIME)rset=self.execute('Any X WHERE X has_text "affreux"')self.failIf(rset)finally:# restore statecu.execute('SET X ref "AFFREF" WHERE X eid %(x)s',{'x':aff1},'x')cnx2.commit()deftest_simplifiable_var(self):affeid=self.execute('Affaire X WHERE X ref "AFFREF"')[0][0]rset=self.execute('Any X,AA,AB WHERE E eid %(x)s, E in_state X, X name AA, X modification_date AB',{'x':affeid},'x')self.assertEquals(len(rset),1)self.assertEquals(rset[0][1],"pitetre")deftest_simplifiable_var_2(self):affeid=self.execute('Affaire X WHERE X ref "AFFREF"')[0][0]rset=self.execute('Any E WHERE E eid %(x)s, E in_state S, NOT S name "moved"',{'x':affeid,'u':self.session.user.eid},'x')self.assertEquals(len(rset),1)deftest_sort_func(self):self.execute('Affaire X ORDERBY DUMB_SORT(RF) WHERE X ref RF')deftest_sort_func_ambigous(self):self.execute('Any X ORDERBY DUMB_SORT(RF) WHERE X title RF')deftest_in_eid(self):iec1=self.repo.extid2eid(self.repo.sources_by_uri['extern'],ec1,'Card',self.session)rset=self.execute('Any X WHERE X eid IN (%s, %s)'%(iec1,self.ic1))self.assertEquals(sorted(r[0]forrinrset.rows),sorted([iec1,self.ic1]))deftest_greater_eid(self):rset=self.execute('Any X WHERE X eid > %s'%self.maxeid)self.assertEquals(len(rset.rows),2)# self.ic1 and self.ic2ec2=cu.execute('INSERT Card X: X title "glup"')[0][0]cnx2.commit()# 'X eid > something' should not trigger discoveryrset=self.execute('Any X WHERE X eid > %s'%self.maxeid)self.assertEquals(len(rset.rows),2)# trigger discovery using another querycrset=self.execute('Card X WHERE X title "glup"')self.assertEquals(len(crset.rows),1)rset=self.execute('Any X WHERE X eid > %s'%self.maxeid)self.assertEquals(len(rset.rows),3)rset=self.execute('Any MAX(X)')self.assertEquals(len(rset.rows),1)self.assertEquals(rset.rows[0][0],crset[0][0])deftest_attr_unification_1(self):n1=self.execute('INSERT Note X: X type "AFFREF"')[0][0]n2=self.execute('INSERT Note X: X type "AFFREU"')[0][0]rset=self.execute('Any X,Y WHERE X is Note, Y is Affaire, X type T, Y ref T')self.assertEquals(len(rset),1,rset.rows)deftest_attr_unification_2(self):ec2=cu.execute('INSERT Card X: X title "AFFREF"')[0][0]cnx2.commit()try:c1=self.execute('INSERT Card C: C title "AFFREF"')[0][0]rset=self.execute('Any X,Y WHERE X is Card, Y is Affaire, X title T, Y ref T')self.assertEquals(len(rset),2,rset.rows)finally:cu.execute('DELETE Card X WHERE X eid %(x)s',{'x':ec2},'x')cnx2.commit()deftest_attr_unification_neq_1(self):# XXX completeself.execute('Any X,Y WHERE X is Note, Y is Affaire, X creation_date D, Y creation_date > D')deftest_attr_unification_neq_2(self):# XXX completeself.execute('Any X,Y WHERE X is Card, Y is Affaire, X creation_date D, Y creation_date > D')deftest_union(self):afeids=self.execute('Affaire X')ueids=self.execute('EUser X')rset=self.execute('(Any X WHERE X is Affaire) UNION (Any X WHERE X is EUser)')self.assertEquals(sorted(r[0]forrinrset.rows),sorted(r[0]forrinafeids+ueids))deftest_subquery1(self):rsetbase=self.execute('Any W,X WITH W,X BEING (Any W,X ORDERBY W,X WHERE X wikiid W)')self.assertEquals(len(rsetbase),4)self.assertEquals(sorted(rsetbase.rows),rsetbase.rows)rset=self.execute('Any W,X LIMIT 2 OFFSET 2 WITH W,X BEING (Any W,X ORDERBY W,X WHERE X wikiid W)')self.assertEquals(rset.rows,rsetbase.rows[2:4])rset=self.execute('Any W,X ORDERBY W,X LIMIT 2 OFFSET 2 WITH W,X BEING (Any W,X WHERE X wikiid W)')self.assertEquals(rset.rows,rsetbase.rows[2:4])rset=self.execute('Any W,X WITH W,X BEING (Any W,X ORDERBY W,X LIMIT 2 OFFSET 2 WHERE X wikiid W)')self.assertEquals(rset.rows,rsetbase.rows[2:4])deftest_subquery2(self):affeid=self.execute('Affaire X WHERE X ref "AFFREF"')[0][0]rset=self.execute('Any X,AA,AB WITH X,AA,AB BEING (Any X,AA,AB WHERE E eid %(x)s, E in_state X, X name AA, X modification_date AB)',{'x':affeid})self.assertEquals(len(rset),1)self.assertEquals(rset[0][1],"pitetre")deftest_not_relation(self):states=set(tuple(x)forxinself.execute('Any S,SN WHERE S is State, S name SN'))userstate=self.session.user.in_state[0]states.remove((userstate.eid,userstate.name))notstates=set(tuple(x)forxinself.execute('Any S,SN WHERE S is State, S name SN, NOT X in_state S, X eid %(x)s',{'x':self.session.user.eid},'x'))self.assertEquals(notstates,states)aff1=self.execute('Any X WHERE X is Affaire, X ref "AFFREF"')[0][0]aff1stateeid,aff1statename=self.execute('Any S,SN WHERE X eid %(x)s, X in_state S, S name SN',{'x':aff1},'x')[0]self.assertEquals(aff1statename,'pitetre')states.add((userstate.eid,userstate.name))states.remove((aff1stateeid,aff1statename))notstates=set(tuple(x)forxinself.execute('Any S,SN WHERE S is State, S name SN, NOT X in_state S, X eid %(x)s',{'x':aff1},'x'))self.set_debug(False)self.assertSetEquals(notstates,states)deftest_nonregr1(self):ueid=self.session.user.eidaffaire=self.execute('Affaire X WHERE X ref "AFFREF"').get_entity(0,0)self.execute('Any U WHERE U in_group G, (G name IN ("managers", "logilab") OR (X require_permission P?, P name "bla", P require_group G)), X eid %(x)s, U eid %(u)s',{'x':affaire.eid,'u':ueid})deftest_nonregr2(self):treid=self.session.user.latest_trinfo().eidrset=self.execute('Any X ORDERBY D DESC WHERE E eid %(x)s, E wf_info_for X, X modification_date D',{'x':treid})self.assertEquals(len(rset),1)self.assertEquals(rset.rows[0],[self.session.user.eid])deftest_nonregr3(self):self.execute('DELETE Card X WHERE X eid %(x)s, NOT X multisource_inlined_rel Y',{'x':self.ic1})if__name__=='__main__':fromlogilab.common.testlibimportunittest_mainunittest_main()