1 # -*- coding: iso-8859-1 -*- |
1 # -*- coding: iso-8859-1 -*- |
2 # copyright 2003-2011 LOGILAB S.A. (Paris, FRANCE), all rights reserved. |
2 # copyright 2003-2012 LOGILAB S.A. (Paris, FRANCE), all rights reserved. |
3 # contact http://www.logilab.fr/ -- mailto:contact@logilab.fr |
3 # contact http://www.logilab.fr/ -- mailto:contact@logilab.fr |
4 # |
4 # |
5 # This file is part of CubicWeb. |
5 # This file is part of CubicWeb. |
6 # |
6 # |
7 # CubicWeb is free software: you can redistribute it and/or modify it under the |
7 # CubicWeb is free software: you can redistribute it and/or modify it under the |
16 # |
16 # |
17 # You should have received a copy of the GNU Lesser General Public License along |
17 # You should have received a copy of the GNU Lesser General Public License along |
18 # with CubicWeb. If not, see <http://www.gnu.org/licenses/>. |
18 # with CubicWeb. If not, see <http://www.gnu.org/licenses/>. |
19 """unit tests for modules cubicweb.server.querier and cubicweb.server.ssplanner |
19 """unit tests for modules cubicweb.server.querier and cubicweb.server.ssplanner |
20 """ |
20 """ |
|
21 from __future__ import with_statement |
|
22 |
21 from datetime import date, datetime, timedelta, tzinfo |
23 from datetime import date, datetime, timedelta, tzinfo |
22 |
24 |
23 from logilab.common.testlib import TestCase, unittest_main |
25 from logilab.common.testlib import TestCase, unittest_main |
24 from rql import BadRQLQuery, RQLSyntaxError |
26 from rql import BadRQLQuery, RQLSyntaxError |
25 |
27 |
26 from cubicweb import QueryError, Unauthorized, Binary |
28 from cubicweb import QueryError, Unauthorized, Binary |
27 from cubicweb.server.sqlutils import SQL_PREFIX |
29 from cubicweb.server.sqlutils import SQL_PREFIX |
28 from cubicweb.server.utils import crypt_password |
30 from cubicweb.server.utils import crypt_password |
29 from cubicweb.server.sources.native import make_schema |
31 from cubicweb.server.sources.native import make_schema |
30 from cubicweb.devtools import get_test_db_handler, TestServerConfiguration |
32 from cubicweb.devtools import get_test_db_handler, TestServerConfiguration |
31 |
33 from cubicweb.devtools.testlib import CubicWebTC |
32 from cubicweb.devtools.repotest import tuplify, BaseQuerierTC |
34 from cubicweb.devtools.repotest import tuplify, BaseQuerierTC |
33 from unittest_session import Variable |
35 from unittest_session import Variable |
34 |
36 |
35 class FixedOffset(tzinfo): |
37 class FixedOffset(tzinfo): |
36 def __init__(self, hours=0): |
38 def __init__(self, hours=0): |
68 self.assertEqual(make_schema((Variable('A'), Variable('B')), solution, |
70 self.assertEqual(make_schema((Variable('A'), Variable('B')), solution, |
69 'table0', TYPEMAP), |
71 'table0', TYPEMAP), |
70 ('C0 text,C1 integer', {'A': 'table0.C0', 'B': 'table0.C1'})) |
72 ('C0 text,C1 integer', {'A': 'table0.C0', 'B': 'table0.C1'})) |
71 |
73 |
72 |
74 |
73 def setUpModule(*args): |
75 def setUpClass(cls, *args): |
74 global repo, cnx |
76 global repo, cnx |
75 config = TestServerConfiguration(apphome=UtilsTC.datadir) |
77 config = TestServerConfiguration(apphome=UtilsTC.datadir) |
76 handler = get_test_db_handler(config) |
78 handler = get_test_db_handler(config) |
77 handler.build_db_cache() |
79 handler.build_db_cache() |
78 repo, cnx = handler.get_repo_and_cnx() |
80 repo, cnx = handler.get_repo_and_cnx() |
79 |
81 cls.repo = repo |
80 def tearDownModule(*args): |
82 |
|
83 def tearDownClass(cls, *args): |
81 global repo, cnx |
84 global repo, cnx |
82 cnx.close() |
85 cnx.close() |
83 repo.shutdown() |
86 repo.shutdown() |
84 del repo, cnx |
87 del repo, cnx |
85 |
88 |
86 |
89 |
87 class UtilsTC(BaseQuerierTC): |
90 class UtilsTC(BaseQuerierTC): |
88 def setUp(self): |
91 setUpClass = classmethod(setUpClass) |
89 self.__class__.repo = repo |
92 tearDownClass = classmethod(tearDownClass) |
90 super(UtilsTC, self).setUp() |
|
91 |
93 |
92 def get_max_eid(self): |
94 def get_max_eid(self): |
93 # no need for cleanup here |
95 # no need for cleanup here |
94 return None |
96 return None |
95 def cleanup(self): |
97 def cleanup(self): |
240 rset = self.execute('Any %(x)s', {'x': u'str'}) |
242 rset = self.execute('Any %(x)s', {'x': u'str'}) |
241 self.assertEqual(rset.description[0][0], 'String') |
243 self.assertEqual(rset.description[0][0], 'String') |
242 |
244 |
243 |
245 |
244 class QuerierTC(BaseQuerierTC): |
246 class QuerierTC(BaseQuerierTC): |
245 def setUp(self): |
247 setUpClass = classmethod(setUpClass) |
246 self.__class__.repo = repo |
248 tearDownClass = classmethod(tearDownClass) |
247 super(QuerierTC, self).setUp() |
|
248 |
249 |
249 def test_encoding_pb(self): |
250 def test_encoding_pb(self): |
250 self.assertRaises(RQLSyntaxError, self.execute, |
251 self.assertRaises(RQLSyntaxError, self.execute, |
251 'Any X WHERE X is CWRType, X name "öwned_by"') |
252 'Any X WHERE X is CWRType, X name "öwned_by"') |
252 |
253 |
1257 self.execute, "Any P WHERE X is CWUser, X login 'bob', X upassword P") |
1258 self.execute, "Any P WHERE X is CWUser, X login 'bob', X upassword P") |
1258 cursor = self.cnxset['system'] |
1259 cursor = self.cnxset['system'] |
1259 cursor.execute("SELECT %supassword from %sCWUser WHERE %slogin='bob'" |
1260 cursor.execute("SELECT %supassword from %sCWUser WHERE %slogin='bob'" |
1260 % (SQL_PREFIX, SQL_PREFIX, SQL_PREFIX)) |
1261 % (SQL_PREFIX, SQL_PREFIX, SQL_PREFIX)) |
1261 passwd = str(cursor.fetchone()[0]) |
1262 passwd = str(cursor.fetchone()[0]) |
1262 self.assertEqual(passwd, crypt_password('toto', passwd[:2])) |
1263 self.assertEqual(passwd, crypt_password('toto', passwd)) |
1263 rset = self.execute("Any X WHERE X is CWUser, X login 'bob', X upassword %(pwd)s", |
1264 rset = self.execute("Any X WHERE X is CWUser, X login 'bob', X upassword %(pwd)s", |
1264 {'pwd': Binary(passwd)}) |
1265 {'pwd': Binary(passwd)}) |
1265 self.assertEqual(len(rset.rows), 1) |
1266 self.assertEqual(len(rset.rows), 1) |
1266 self.assertEqual(rset.description, [('CWUser',)]) |
1267 self.assertEqual(rset.description, [('CWUser',)]) |
1267 |
1268 |
1272 {'pwd': 'tutu'}) |
1273 {'pwd': 'tutu'}) |
1273 cursor = self.cnxset['system'] |
1274 cursor = self.cnxset['system'] |
1274 cursor.execute("SELECT %supassword from %sCWUser WHERE %slogin='bob'" |
1275 cursor.execute("SELECT %supassword from %sCWUser WHERE %slogin='bob'" |
1275 % (SQL_PREFIX, SQL_PREFIX, SQL_PREFIX)) |
1276 % (SQL_PREFIX, SQL_PREFIX, SQL_PREFIX)) |
1276 passwd = str(cursor.fetchone()[0]) |
1277 passwd = str(cursor.fetchone()[0]) |
1277 self.assertEqual(passwd, crypt_password('tutu', passwd[:2])) |
1278 self.assertEqual(passwd, crypt_password('tutu', passwd)) |
1278 rset = self.execute("Any X WHERE X is CWUser, X login 'bob', X upassword %(pwd)s", |
1279 rset = self.execute("Any X WHERE X is CWUser, X login 'bob', X upassword %(pwd)s", |
1279 {'pwd': Binary(passwd)}) |
1280 {'pwd': Binary(passwd)}) |
1280 self.assertEqual(len(rset.rows), 1) |
1281 self.assertEqual(len(rset.rows), 1) |
1281 self.assertEqual(rset.description, [('CWUser',)]) |
1282 self.assertEqual(rset.description, [('CWUser',)]) |
1282 |
1283 |
1499 def test_nonregr_sql_cache(self): |
1500 def test_nonregr_sql_cache(self): |
1500 # different SQL generated when 'name' is None or not (IS NULL). |
1501 # different SQL generated when 'name' is None or not (IS NULL). |
1501 self.assertFalse(self.execute('Any X WHERE X is CWEType, X name %(name)s', {'name': None})) |
1502 self.assertFalse(self.execute('Any X WHERE X is CWEType, X name %(name)s', {'name': None})) |
1502 self.assertTrue(self.execute('Any X WHERE X is CWEType, X name %(name)s', {'name': 'CWEType'})) |
1503 self.assertTrue(self.execute('Any X WHERE X is CWEType, X name %(name)s', {'name': 'CWEType'})) |
1503 |
1504 |
|
1505 |
|
1506 class NonRegressionTC(CubicWebTC): |
|
1507 |
|
1508 def test_has_text_security_cache_bug(self): |
|
1509 req = self.request() |
|
1510 self.create_user(req, 'user', ('users',)) |
|
1511 aff1 = req.create_entity('Societe', nom=u'aff1') |
|
1512 aff2 = req.create_entity('Societe', nom=u'aff2') |
|
1513 self.commit() |
|
1514 with self.login('user', password='user'): |
|
1515 res = self.execute('Any X WHERE X has_text %(text)s', {'text': 'aff1'}) |
|
1516 self.assertEqual(res.rows, [[aff1.eid]]) |
|
1517 res = self.execute('Any X WHERE X has_text %(text)s', {'text': 'aff2'}) |
|
1518 self.assertEqual(res.rows, [[aff2.eid]]) |
|
1519 |
1504 if __name__ == '__main__': |
1520 if __name__ == '__main__': |
1505 unittest_main() |
1521 unittest_main() |