server/test/unittest_repository.py
changeset 2968 0e3460341023
parent 2773 b2530e3e0afb
parent 2923 b97a0f8dd4dc
child 3445 32492e9e734b
equal deleted inserted replaced
2902:dd9f2dd02f85 2968:0e3460341023
    87         repo.close(cnxid)
    87         repo.close(cnxid)
    88 
    88 
    89     def test_login_upassword_accent(self):
    89     def test_login_upassword_accent(self):
    90         repo = self.repo
    90         repo = self.repo
    91         cnxid = repo.connect(self.admlogin, self.admpassword)
    91         cnxid = repo.connect(self.admlogin, self.admpassword)
    92         repo.execute(cnxid, 'INSERT CWUser X: X login %(login)s, X upassword %(passwd)s, X in_state S, X in_group G WHERE S name "activated", G name "users"',
    92         repo.execute(cnxid, 'INSERT CWUser X: X login %(login)s, X upassword %(passwd)s, X in_group G WHERE G name "users"',
    93                      {'login': u"barnabé", 'passwd': u"héhéhé".encode('UTF8')})
    93                      {'login': u"barnabé", 'passwd': u"héhéhé".encode('UTF8')})
    94         repo.commit(cnxid)
    94         repo.commit(cnxid)
    95         repo.close(cnxid)
    95         repo.close(cnxid)
    96         self.assert_(repo.connect(u"barnabé", u"héhéhé".encode('UTF8')))
    96         self.assert_(repo.connect(u"barnabé", u"héhéhé".encode('UTF8')))
    97 
    97 
    98     def test_invalid_entity_rollback(self):
    98     def test_invalid_entity_rollback(self):
    99         repo = self.repo
    99         repo = self.repo
   100         cnxid = repo.connect(self.admlogin, self.admpassword)
   100         cnxid = repo.connect(self.admlogin, self.admpassword)
   101         # no group
   101         # no group
   102         repo.execute(cnxid, 'INSERT CWUser X: X login %(login)s, X upassword %(passwd)s, X in_state S WHERE S name "activated"',
   102         repo.execute(cnxid, 'INSERT CWUser X: X login %(login)s, X upassword %(passwd)s',
   103                      {'login': u"tutetute", 'passwd': 'tutetute'})
   103                      {'login': u"tutetute", 'passwd': 'tutetute'})
   104         self.assertRaises(ValidationError, repo.commit, cnxid)
   104         self.assertRaises(ValidationError, repo.commit, cnxid)
   105         rset = repo.execute(cnxid, 'CWUser X WHERE X login "tutetute"')
   105         rset = repo.execute(cnxid, 'CWUser X WHERE X login "tutetute"')
   106         self.assertEquals(rset.rowcount, 0)
   106         self.assertEquals(rset.rowcount, 0)
   107 
   107 
   175 
   175 
   176     def test_transaction_base3(self):
   176     def test_transaction_base3(self):
   177         repo = self.repo
   177         repo = self.repo
   178         cnxid = repo.connect(self.admlogin, self.admpassword)
   178         cnxid = repo.connect(self.admlogin, self.admpassword)
   179         # rollback state change which trigger TrInfo insertion
   179         # rollback state change which trigger TrInfo insertion
   180         ueid = repo._get_session(cnxid).user.eid
   180         user = repo._get_session(cnxid).user
   181         rset = repo.execute(cnxid, 'TrInfo T WHERE T wf_info_for X, X eid %(x)s', {'x': ueid})
   181         user.fire_transition('deactivate')
       
   182         rset = repo.execute(cnxid, 'TrInfo T WHERE T wf_info_for X, X eid %(x)s', {'x': user.eid})
   182         self.assertEquals(len(rset), 1)
   183         self.assertEquals(len(rset), 1)
   183         repo.execute(cnxid, 'SET X in_state S WHERE X eid %(x)s, S name "deactivated"',
       
   184                      {'x': ueid}, 'x')
       
   185         rset = repo.execute(cnxid, 'TrInfo T WHERE T wf_info_for X, X eid %(x)s', {'x': ueid})
       
   186         self.assertEquals(len(rset), 2)
       
   187         repo.rollback(cnxid)
   184         repo.rollback(cnxid)
   188         rset = repo.execute(cnxid, 'TrInfo T WHERE T wf_info_for X, X eid %(x)s', {'x': ueid})
   185         rset = repo.execute(cnxid, 'TrInfo T WHERE T wf_info_for X, X eid %(x)s', {'x': user.eid})
   189         self.assertEquals(len(rset), 1)
   186         self.assertEquals(len(rset), 0)
   190 
   187 
   191     def test_transaction_interleaved(self):
   188     def test_transaction_interleaved(self):
   192         self.skip('implement me')
   189         self.skip('implement me')
   193 
   190 
   194     def test_close_wait_processing_request(self):
   191     def test_close_wait_processing_request(self):
   327 #             self.create_user('toto')
   324 #             self.create_user('toto')
   328 #         finally:
   325 #         finally:
   329 #             self.set_debug(False)
   326 #             self.set_debug(False)
   330 #         print 'test time: %.3f (time) %.3f (cpu)' % ((time() - t), clock() - c)
   327 #         print 'test time: %.3f (time) %.3f (cpu)' % ((time() - t), clock() - c)
   331 
   328 
       
   329     def test_delete_if_singlecard1(self):
       
   330         note = self.add_entity('Affaire')
       
   331         p1 = self.add_entity('Personne', nom=u'toto')
       
   332         self.execute('SET A todo_by P WHERE A eid %(x)s, P eid %(p)s',
       
   333                      {'x': note.eid, 'p': p1.eid})
       
   334         rset = self.execute('Any P WHERE A todo_by P, A eid %(x)s',
       
   335                             {'x': note.eid})
       
   336         self.assertEquals(len(rset), 1)
       
   337         p2 = self.add_entity('Personne', nom=u'tutu')
       
   338         self.execute('SET A todo_by P WHERE A eid %(x)s, P eid %(p)s',
       
   339                      {'x': note.eid, 'p': p2.eid})
       
   340         rset = self.execute('Any P WHERE A todo_by P, A eid %(x)s',
       
   341                             {'x': note.eid})
       
   342         self.assertEquals(len(rset), 1)
       
   343         self.assertEquals(rset.rows[0][0], p2.eid)
       
   344 
   332 
   345 
   333 class DataHelpersTC(CubicWebTC):
   346 class DataHelpersTC(CubicWebTC):
   334 
   347 
   335     def test_create_eid(self):
   348     def test_create_eid(self):
   336         self.session.set_pool()
   349         self.session.set_pool()
   469         self.execute('SET N ecrit_par Y WHERE N type "T", Y nom "toto"')
   482         self.execute('SET N ecrit_par Y WHERE N type "T", Y nom "toto"')
   470         self.assertEquals(self.called, [(eidn, 'ecrit_par', eidp,)])
   483         self.assertEquals(self.called, [(eidn, 'ecrit_par', eidp,)])
   471 
   484 
   472     def test_after_add_inline(self):
   485     def test_after_add_inline(self):
   473         """make sure after_<event>_relation hooks are deferred"""
   486         """make sure after_<event>_relation hooks are deferred"""
       
   487         p1 = self.add_entity('Personne', nom=u'toto')
   474         self.hm.register_hook(self._after_relation_hook,
   488         self.hm.register_hook(self._after_relation_hook,
   475                              'after_add_relation', 'in_state')
   489                              'after_add_relation', 'ecrit_par')
   476         eidp = self.execute('INSERT CWUser X: X login "toto", X upassword "tutu", X in_state S WHERE S name "activated"')[0][0]
   490         eidn = self.execute('INSERT Note N: N ecrit_par P WHERE P nom "toto"')[0][0]
   477         eids = self.execute('State X WHERE X name "activated"')[0][0]
   491         self.assertEquals(self.called, [(eidn, 'ecrit_par', p1.eid,)])
   478         self.assertEquals(self.called, [(eidp, 'in_state', eids,)])
       
   479 
   492 
   480     def test_before_delete_inline_relation(self):
   493     def test_before_delete_inline_relation(self):
   481         """make sure before_<event>_relation hooks are called directly"""
   494         """make sure before_<event>_relation hooks are called directly"""
   482         self.hm.register_hook(self._before_relation_hook,
   495         self.hm.register_hook(self._before_relation_hook,
   483                              'before_delete_relation', 'ecrit_par')
   496                              'before_delete_relation', 'ecrit_par')