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 datetime import date, datetime |
21 from datetime import date, datetime, timedelta, tzinfo |
22 |
22 |
23 from logilab.common.testlib import TestCase, unittest_main |
23 from logilab.common.testlib import TestCase, unittest_main |
24 from rql import BadRQLQuery, RQLSyntaxError |
24 from rql import BadRQLQuery, RQLSyntaxError |
25 |
25 |
26 from cubicweb import QueryError, Unauthorized, Binary |
26 from cubicweb import QueryError, Unauthorized, Binary |
29 from cubicweb.server.sources.native import make_schema |
29 from cubicweb.server.sources.native import make_schema |
30 from cubicweb.devtools import get_test_db_handler, TestServerConfiguration |
30 from cubicweb.devtools import get_test_db_handler, TestServerConfiguration |
31 |
31 |
32 from cubicweb.devtools.repotest import tuplify, BaseQuerierTC |
32 from cubicweb.devtools.repotest import tuplify, BaseQuerierTC |
33 from unittest_session import Variable |
33 from unittest_session import Variable |
|
34 |
|
35 class FixedOffset(tzinfo): |
|
36 def __init__(self, hours=0): |
|
37 self.hours = hours |
|
38 def utcoffset(self, dt): |
|
39 return timedelta(hours=self.hours) |
|
40 def dst(self, dt): |
|
41 return timedelta(0) |
34 |
42 |
35 |
43 |
36 # register priority/severity sorting registered procedure |
44 # register priority/severity sorting registered procedure |
37 from rql.utils import register_function, FunctionDescr |
45 from rql.utils import register_function, FunctionDescr |
38 |
46 |
1213 {'pwd': Binary(passwd)}) |
1221 {'pwd': Binary(passwd)}) |
1214 self.assertEqual(len(rset.rows), 1) |
1222 self.assertEqual(len(rset.rows), 1) |
1215 self.assertEqual(rset.description, [('CWUser',)]) |
1223 self.assertEqual(rset.description, [('CWUser',)]) |
1216 |
1224 |
1217 def test_update_upassword(self): |
1225 def test_update_upassword(self): |
1218 cursor = self.pool['system'] |
|
1219 rset = self.execute("INSERT CWUser X: X login 'bob', X upassword %(pwd)s", {'pwd': 'toto'}) |
1226 rset = self.execute("INSERT CWUser X: X login 'bob', X upassword %(pwd)s", {'pwd': 'toto'}) |
1220 self.assertEqual(rset.description[0][0], 'CWUser') |
1227 self.assertEqual(rset.description[0][0], 'CWUser') |
1221 rset = self.execute("SET X upassword %(pwd)s WHERE X is CWUser, X login 'bob'", |
1228 rset = self.execute("SET X upassword %(pwd)s WHERE X is CWUser, X login 'bob'", |
1222 {'pwd': 'tutu'}) |
1229 {'pwd': 'tutu'}) |
|
1230 cursor = self.pool['system'] |
1223 cursor.execute("SELECT %supassword from %sCWUser WHERE %slogin='bob'" |
1231 cursor.execute("SELECT %supassword from %sCWUser WHERE %slogin='bob'" |
1224 % (SQL_PREFIX, SQL_PREFIX, SQL_PREFIX)) |
1232 % (SQL_PREFIX, SQL_PREFIX, SQL_PREFIX)) |
1225 passwd = str(cursor.fetchone()[0]) |
1233 passwd = str(cursor.fetchone()[0]) |
1226 self.assertEqual(passwd, crypt_password('tutu', passwd[:2])) |
1234 self.assertEqual(passwd, crypt_password('tutu', passwd[:2])) |
1227 rset = self.execute("Any X WHERE X is CWUser, X login 'bob', X upassword %(pwd)s", |
1235 rset = self.execute("Any X WHERE X is CWUser, X login 'bob', X upassword %(pwd)s", |
1228 {'pwd': Binary(passwd)}) |
1236 {'pwd': Binary(passwd)}) |
1229 self.assertEqual(len(rset.rows), 1) |
1237 self.assertEqual(len(rset.rows), 1) |
1230 self.assertEqual(rset.description, [('CWUser',)]) |
1238 self.assertEqual(rset.description, [('CWUser',)]) |
1231 |
1239 |
1232 # non regression tests #################################################### |
1240 # ZT datetime tests ######################################################## |
|
1241 |
|
1242 def test_tz_datetime(self): |
|
1243 self.execute("INSERT Personne X: X nom 'bob', X tzdatenaiss %(date)s", |
|
1244 {'date': datetime(1977, 6, 7, 2, 0, tzinfo=FixedOffset(1))}) |
|
1245 datenaiss = self.execute("Any XD WHERE X nom 'bob', X tzdatenaiss XD")[0][0] |
|
1246 self.assertEqual(datenaiss.utctimetuple()[:5], (1977, 6, 7, 1, 0)) |
|
1247 |
|
1248 # non regression tests ##################################################### |
1233 |
1249 |
1234 def test_nonregr_1(self): |
1250 def test_nonregr_1(self): |
1235 teid = self.execute("INSERT Tag X: X name 'tag'")[0][0] |
1251 teid = self.execute("INSERT Tag X: X name 'tag'")[0][0] |
1236 self.execute("SET X tags Y WHERE X name 'tag', Y is State, Y name 'activated'") |
1252 self.execute("SET X tags Y WHERE X name 'tag', Y is State, Y name 'activated'") |
1237 rset = self.execute('Any X WHERE T tags X') |
1253 rset = self.execute('Any X WHERE T tags X') |