server/test/unittest_querier.py
branchstable
changeset 4674 3d509dbb473a
parent 4467 0e73d299730a
child 4680 8a6bee838464
equal deleted inserted replaced
4673:6f8b925a29f4 4674:3d509dbb473a
    76         self.assertEquals(len(rqlst.solutions), 1)
    76         self.assertEquals(len(rqlst.solutions), 1)
    77 
    77 
    78     def test_preprocess_security(self):
    78     def test_preprocess_security(self):
    79         plan = self._prepare_plan('Any ETN,COUNT(X) GROUPBY ETN '
    79         plan = self._prepare_plan('Any ETN,COUNT(X) GROUPBY ETN '
    80                                   'WHERE X is ET, ET name ETN')
    80                                   'WHERE X is ET, ET name ETN')
    81         plan.session = self._user_session(('users',))[1]
    81         plan.session = self.user_groups_session('users')
    82         union = plan.rqlst
    82         union = plan.rqlst
    83         plan.preprocess(union)
    83         plan.preprocess(union)
    84         self.assertEquals(len(union.children), 1)
    84         self.assertEquals(len(union.children), 1)
    85         self.assertEquals(len(union.children[0].with_), 1)
    85         self.assertEquals(len(union.children[0].with_), 1)
    86         subq = union.children[0].with_[0].query
    86         subq = union.children[0].with_[0].query
   156                                        'ETN': 'String',
   156                                        'ETN': 'String',
   157                                        }])
   157                                        }])
   158 
   158 
   159     def test_preprocess_security_aggregat(self):
   159     def test_preprocess_security_aggregat(self):
   160         plan = self._prepare_plan('Any MAX(X)')
   160         plan = self._prepare_plan('Any MAX(X)')
   161         plan.session = self._user_session(('users',))[1]
   161         plan.session = self.user_groups_session('users')
   162         union = plan.rqlst
   162         union = plan.rqlst
   163         plan.preprocess(union)
   163         plan.preprocess(union)
   164         self.assertEquals(len(union.children), 1)
   164         self.assertEquals(len(union.children), 1)
   165         self.assertEquals(len(union.children[0].with_), 1)
   165         self.assertEquals(len(union.children[0].with_), 1)
   166         subq = union.children[0].with_[0].query
   166         subq = union.children[0].with_[0].query
   926         self.execute("DELETE X travaille Y WHERE X is Personne, Y nom 'Logilab'")
   926         self.execute("DELETE X travaille Y WHERE X is Personne, Y nom 'Logilab'")
   927         rset = self.execute('Personne X WHERE X travaille Y, Y nom "Logilab"')
   927         rset = self.execute('Personne X WHERE X travaille Y, Y nom "Logilab"')
   928         self.assertEqual(len(rset.rows), 0, rset.rows)
   928         self.assertEqual(len(rset.rows), 0, rset.rows)
   929 
   929 
   930     def test_delete_3(self):
   930     def test_delete_3(self):
   931         u, s = self._user_session(('users',))
   931         s = self.user_groups_session('users')
   932         peid, = self.o.execute(s, "INSERT Personne P: P nom 'toto'")[0]
   932         peid, = self.o.execute(s, "INSERT Personne P: P nom 'toto'")[0]
   933         seid, = self.o.execute(s, "INSERT Societe S: S nom 'logilab'")[0]
   933         seid, = self.o.execute(s, "INSERT Societe S: S nom 'logilab'")[0]
   934         self.o.execute(s, "SET P travaille S")
   934         self.o.execute(s, "SET P travaille S")
   935         rset = self.execute('Personne P WHERE P travaille S')
   935         rset = self.execute('Personne P WHERE P travaille S')
   936         self.assertEqual(len(rset.rows), 1)
   936         self.assertEqual(len(rset.rows), 1)
   957     def test_nonregr_delete_cache(self):
   957     def test_nonregr_delete_cache(self):
   958         """test that relations are properly cleaned when an entity is deleted
   958         """test that relations are properly cleaned when an entity is deleted
   959         (using cachekey on sql generation returned always the same query for an eid,
   959         (using cachekey on sql generation returned always the same query for an eid,
   960         whatever the relation)
   960         whatever the relation)
   961         """
   961         """
   962         u, s = self._user_session(('users',))
   962         s = self.user_groups_session('users')
   963         aeid, = self.o.execute(s, 'INSERT EmailAddress X: X address "toto@logilab.fr", X alias "hop"')[0]
   963         aeid, = self.o.execute(s, 'INSERT EmailAddress X: X address "toto@logilab.fr", X alias "hop"')[0]
   964         # XXX would be nice if the rql below was enough...
   964         # XXX would be nice if the rql below was enough...
   965         #'INSERT Email X: X messageid "<1234>", X subject "test", X sender Y, X recipients Y'
   965         #'INSERT Email X: X messageid "<1234>", X subject "test", X sender Y, X recipients Y'
   966         eeid, = self.o.execute(s, 'INSERT Email X: X messageid "<1234>", X subject "test", X sender Y, X recipients Y WHERE Y is EmailAddress')[0]
   966         eeid, = self.o.execute(s, 'INSERT Email X: X messageid "<1234>", X subject "test", X sender Y, X recipients Y WHERE Y is EmailAddress')[0]
   967         self.o.execute(s, "DELETE Email X")
   967         self.o.execute(s, "DELETE Email X")