25 |
25 |
26 from six import PY2, integer_types, binary_type, text_type |
26 from six import PY2, integer_types, binary_type, text_type |
27 |
27 |
28 from logilab.common.testlib import TestCase, unittest_main |
28 from logilab.common.testlib import TestCase, unittest_main |
29 from rql import BadRQLQuery |
29 from rql import BadRQLQuery |
|
30 from rql.utils import register_function, FunctionDescr |
30 |
31 |
31 from cubicweb import QueryError, Unauthorized, Binary |
32 from cubicweb import QueryError, Unauthorized, Binary |
32 from cubicweb.server.sqlutils import SQL_PREFIX |
33 from cubicweb.server.sqlutils import SQL_CONNECT_HOOKS, SQL_PREFIX |
33 from cubicweb.server.utils import crypt_password |
34 from cubicweb.server.utils import crypt_password |
34 from cubicweb.server.querier import manual_build_descr, _make_description |
35 from cubicweb.server.querier import manual_build_descr, _make_description |
35 from cubicweb.devtools import get_test_db_handler, TestServerConfiguration |
36 from cubicweb.devtools import get_test_db_handler, TestServerConfiguration |
36 from cubicweb.devtools.testlib import CubicWebTC |
37 from cubicweb.devtools.testlib import CubicWebTC |
37 from cubicweb.devtools.repotest import tuplify, BaseQuerierTC |
38 from cubicweb.devtools.repotest import tuplify, BaseQuerierTC |
38 |
39 |
39 |
40 |
40 class FixedOffset(tzinfo): |
41 class FixedOffset(tzinfo): |
41 def __init__(self, hours=0): |
42 def __init__(self, hours=0): |
42 self.hours = hours |
43 self.hours = hours |
|
44 |
43 def utcoffset(self, dt): |
45 def utcoffset(self, dt): |
44 return timedelta(hours=self.hours) |
46 return timedelta(hours=self.hours) |
|
47 |
45 def dst(self, dt): |
48 def dst(self, dt): |
46 return timedelta(0) |
49 return timedelta(0) |
47 |
50 |
48 |
51 |
49 # register priority/severity sorting registered procedure |
52 # register priority/severity sorting registered procedure |
50 from rql.utils import register_function, FunctionDescr |
|
51 |
53 |
52 class group_sort_value(FunctionDescr): |
54 class group_sort_value(FunctionDescr): |
53 supported_backends = ('sqlite',) |
55 supported_backends = ('sqlite',) |
54 rtype = 'Int' |
56 rtype = 'Int' |
|
57 |
55 try: |
58 try: |
56 register_function(group_sort_value) |
59 register_function(group_sort_value) |
57 except AssertionError: |
60 except AssertionError: |
58 pass |
61 pass |
59 from cubicweb.server.sqlutils import SQL_CONNECT_HOOKS |
62 |
|
63 |
60 def init_sqlite_connexion(cnx): |
64 def init_sqlite_connexion(cnx): |
|
65 |
61 def group_sort_value(text): |
66 def group_sort_value(text): |
62 return {"managers": "3", "users": "2", "guests": "1", "owners": "0"}[text] |
67 return {"managers": "3", "users": "2", "guests": "1", "owners": "0"}[text] |
|
68 |
63 cnx.create_function("GROUP_SORT_VALUE", 1, group_sort_value) |
69 cnx.create_function("GROUP_SORT_VALUE", 1, group_sort_value) |
|
70 |
64 SQL_CONNECT_HOOKS['sqlite'].append(init_sqlite_connexion) |
71 SQL_CONNECT_HOOKS['sqlite'].append(init_sqlite_connexion) |
65 |
72 |
66 |
73 |
67 def setUpClass(cls, *args): |
74 def setUpClass(cls, *args): |
68 global repo, cnx |
75 global repo, cnx |
1490 rset = self.qexecute('Any lower(N) ORDERBY LOWER(N) WHERE X is Tag, X name N,' |
1497 rset = self.qexecute('Any lower(N) ORDERBY LOWER(N) WHERE X is Tag, X name N,' |
1491 'X owned_by U, U eid %(x)s', |
1498 'X owned_by U, U eid %(x)s', |
1492 {'x':self.session.user.eid}) |
1499 {'x':self.session.user.eid}) |
1493 self.assertEqual(rset.rows, [[u'\xe9name0']]) |
1500 self.assertEqual(rset.rows, [[u'\xe9name0']]) |
1494 |
1501 |
1495 |
|
1496 def test_nonregr_description(self): |
1502 def test_nonregr_description(self): |
1497 """check that a correct description is built in case where infered |
1503 """check that a correct description is built in case where infered |
1498 solutions may be "fusionned" into one by the querier while all solutions |
1504 solutions may be "fusionned" into one by the querier while all solutions |
1499 are needed to build the result's description |
1505 are needed to build the result's description |
1500 """ |
1506 """ |
1504 self.qexecute("SET X in_basket B WHERE X is Personne") |
1510 self.qexecute("SET X in_basket B WHERE X is Personne") |
1505 self.qexecute("SET X in_basket B WHERE X is Societe") |
1511 self.qexecute("SET X in_basket B WHERE X is Societe") |
1506 rset = self.qexecute('Any X WHERE X in_basket B, B eid %s' % beid) |
1512 rset = self.qexecute('Any X WHERE X in_basket B, B eid %s' % beid) |
1507 self.assertEqual(len(rset), 2) |
1513 self.assertEqual(len(rset), 2) |
1508 self.assertEqual(rset.description, [('Personne',), ('Societe',)]) |
1514 self.assertEqual(rset.description, [('Personne',), ('Societe',)]) |
1509 |
|
1510 |
1515 |
1511 def test_nonregr_cache_1(self): |
1516 def test_nonregr_cache_1(self): |
1512 peid = self.qexecute("INSERT Personne X: X nom 'bidule'")[0][0] |
1517 peid = self.qexecute("INSERT Personne X: X nom 'bidule'")[0][0] |
1513 beid = self.qexecute("INSERT Basket X: X name 'tag'")[0][0] |
1518 beid = self.qexecute("INSERT Basket X: X name 'tag'")[0][0] |
1514 self.qexecute("SET X in_basket Y WHERE X is Personne, Y eid %(y)s", |
1519 self.qexecute("SET X in_basket Y WHERE X is Personne, Y eid %(y)s", |
1559 'S eid %(x)s, S transition_of ET, O state_of ET', {'x': teid}) |
1564 'S eid %(x)s, S transition_of ET, O state_of ET', {'x': teid}) |
1560 self.assertEqual(len(rset), 2) |
1565 self.assertEqual(len(rset), 2) |
1561 rset = self.qexecute('Any O WHERE O is State, NOT S destination_state O, ' |
1566 rset = self.qexecute('Any O WHERE O is State, NOT S destination_state O, ' |
1562 'S eid %(x)s, S transition_of ET, O state_of ET', {'x': teid}) |
1567 'S eid %(x)s, S transition_of ET, O state_of ET', {'x': teid}) |
1563 self.assertEqual(len(rset), 1) |
1568 self.assertEqual(len(rset), 1) |
1564 |
|
1565 |
1569 |
1566 def test_nonregr_set_datetime(self): |
1570 def test_nonregr_set_datetime(self): |
1567 # huum, psycopg specific |
1571 # huum, psycopg specific |
1568 self.qexecute('SET X creation_date %(date)s WHERE X eid 1', {'date': date.today()}) |
1572 self.qexecute('SET X creation_date %(date)s WHERE X eid 1', {'date': date.today()}) |
1569 |
1573 |