unittest_rqlcontroller.py 7.02 KB
Newer Older
1
# copyright 2013-2022 LOGILAB S.A. (Paris, FRANCE), all rights reserved.
2
3
4
5
6
7
8
9
10
11
12
13
# contact http://www.logilab.fr -- mailto:contact@logilab.fr
#
# This program is free software: you can redistribute it and/or modify it under
# the terms of the GNU Lesser General Public License as published by the Free
# Software Foundation, either version 2.1 of the License, or (at your option)
# any later version.
#
# This program is distributed in the hope that it will be useful, but WITHOUT
# ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS
# FOR A PARTICULAR PURPOSE. See the GNU Lesser General Public License for more
# details.
#
Denis Laxalde's avatar
Denis Laxalde committed
14
15
# You should have received a copy of the GNU Lesser General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
16
17

import json
Denis Laxalde's avatar
Denis Laxalde committed
18

19
from cubicweb.pyramid.test import PyramidCWTest
20
21


22
23
24
class RqlIOTC(PyramidCWTest):
    settings = {"cubicweb.bwcompat": True}

Noé Gaumont's avatar
Noé Gaumont committed
25
    password = "gingkow"
26

Denis Laxalde's avatar
Denis Laxalde committed
27
    def setup_database(self):
28
        with self.admin_access.client_cnx() as cnx:
Noé Gaumont's avatar
Noé Gaumont committed
29
            self.create_user(cnx, u"toto", password=u"toto")
David Douard's avatar
David Douard committed
30
            cnx.commit()
31

Noé Gaumont's avatar
Noé Gaumont committed
32
33
34
35
36
37
    def assertRQLPostOK(self, queries, code=200, version="1.0"):
        return self.webapp.post(
            "/rqlio/%s" % version,
            params=json.dumps(queries),
            headers={"Content-Type": "application/json"},
        )
38
39

    def assertRQLPostKO(self, queries, reason, code=500):
Noé Gaumont's avatar
Noé Gaumont committed
40
41
42
43
44
45
46
        res_ko = self.webapp.post(
            "/rqlio/1.0",
            params=json.dumps(queries),
            headers={"Content-Type": "application/json"},
            status=code,
        )
        self.assertIn(reason, res_ko.json[u"reason"])
47
48
49
        return res_ko

    def test_queries(self):
Noé Gaumont's avatar
Noé Gaumont committed
50
51
52
53
54
55
56
57
58
59
60
        queries = [
            (
                "INSERT CWUser U: U login %(l)s, U upassword %(p)s",
                {"l": "Babar", "p": "cubicweb rulez & 42"},
            ),
            ('INSERT CWGroup G: G name "pachyderms"', {}),
            (
                "SET U in_group G WHERE U eid %(u)s, G eid %(g)s",
                {"u": "__r0", "g": "__r1"},
            ),
        ]
61
62

        # as an anonymous user
Noé Gaumont's avatar
Noé Gaumont committed
63
64
65
66
        reason = (
            u"You are not allowed to perform add operation on relation"
            " CWUser in_group CWGroup"
        )
67
        # should really be 403 if it wasn't for cubicweb brokenness
68
        self.assertRQLPostKO(queries, reason, code=500)
69
70

        # as a standard user
Noé Gaumont's avatar
Noé Gaumont committed
71
72
73
74
75
        self.webapp.login(user="toto", password="toto")
        reason = (
            u"You are not allowed to perform add operation on relation"
            " CWUser in_group CWGroup"
        )
76
        # should really be 403 if it wasn't for cubicweb brokenness
77
        self.assertRQLPostKO(queries, reason, code=500)
78

79
80
        # logout
        self.webapp.reset()
81
        # now, as an admin
Noé Gaumont's avatar
Noé Gaumont committed
82
        self.webapp.login(user="admin", password=self.password)
83
        res = self.assertRQLPostOK(queries)
84

David Douard's avatar
David Douard committed
85
        with self.admin_access.client_cnx() as cnx:
Noé Gaumont's avatar
Noé Gaumont committed
86
87
88
89
            rset = cnx.execute(
                'String N WHERE U in_group G, U login "Babar", ' "G name N"
            )
            self.assertEqual("pachyderms", rset.rows[0][0])
90
        output = [x for x, in res.json]
91
92
93
94
        self.assertEqual(1, len(output[0]))
        self.assertEqual(1, len(output[1]))
        self.assertEqual(2, len(output[2]))
        self.assertEqual([output[0][0], output[1][0]], output[2])
95

96
97
    def test_queries_multipart(self):
        queries = [
Noé Gaumont's avatar
Noé Gaumont committed
98
99
100
101
            (
                "INSERT CWUser U: U login %(l)s, U upassword %(p)s",
                {"l": "Babar", "p": "cubicweb rulez & 42"},
            ),
102
            ('INSERT CWGroup G: G name "pachyderms"', {}),
Noé Gaumont's avatar
Noé Gaumont committed
103
104
105
106
            (
                "SET U in_group G WHERE U eid %(u)s, G eid %(g)s",
                {"u": "__r0", "g": "__r1"},
            ),
107
        ]
108
109
110
111
112
113
114
115
116
117
118

        # FIXME XXX: to change to self.webapp.login() once the issue is solved:
        # https://forge.extranet.logilab.fr/cubicweb/cubicweb/-/issues/370
        response = self.webapp.get("/login")
        csrf_token = response.forms[0].fields["csrf_token"][0].value

        res = self.webapp.post(
            "/login", {"__login": "admin", "__password": self.password}
        )
        self.assertEqual(res.status_int, 303)

Noé Gaumont's avatar
Noé Gaumont committed
119
120
121
122
        files = [("json", "loutre.json", json.dumps(queries).encode("utf-8"))]

        self.webapp.post(
            "/rqlio/1.0",
123
            {"csrf_token": csrf_token},
Noé Gaumont's avatar
Noé Gaumont committed
124
125
126
127
128
            upload_files=files,
            headers={
                "Accept": "application/json",
            },
        )
129

130
    def test_rewrite_args_errors(self):
Noé Gaumont's avatar
Noé Gaumont committed
131
        rql1 = "Any U WHERE U login %(l)s"
132
        rql2 = 'SET U in_group G WHERE G name "managers", U eid %(u)s'
Noé Gaumont's avatar
Noé Gaumont committed
133
        args2 = {"u": "__r0"}
134
        # setup test
Noé Gaumont's avatar
Noé Gaumont committed
135
        self.webapp.login(user="admin", password=self.password)
136
        # check ok
Noé Gaumont's avatar
Noé Gaumont committed
137
        queries_ok = [(rql1, {"l": "toto"}), (rql2, args2)]
138
        self.assertRQLPostOK(queries_ok)
139
        # check ko (1)
Noé Gaumont's avatar
Noé Gaumont committed
140
141
        queries_ko = [(rql1, {"l": "doesnotexist"}), (rql2, args2)]
        self.assertRQLPostKO(queries_ko, "__r0 references empty result set")
142
        # check ko (2)
Noé Gaumont's avatar
Noé Gaumont committed
143
144
        queries_ko = [("Any U WHERE U is CWUser", None), (rql2, args2)]
        self.assertRQLPostKO(queries_ko, "__r0 references multi lines result set")
145
        # check ko (3)
Noé Gaumont's avatar
Noé Gaumont committed
146
147
148
149
150
        queries_ko = [
            ("Any U,L WHERE U login L, U login %(l)s", {"l": "toto"}),
            (rql2, args2),
        ]
        self.assertRQLPostKO(queries_ko, "__r0 references multi column result set")
151

152
153
154
    def test_variables_in_rqlio_v2(self):
        ueid = self.admin_access._user.eid
        queries = [
Noé Gaumont's avatar
Noé Gaumont committed
155
156
157
            ("Any X WHERE X is CWUser, X eid %(x)s", {"x": ueid}),
            ("Any COUNT(X) WHERE X is CWUser", {}),
            ("Any X,COUNT(X) WHERE X is CWUser", {}),
158
        ]
Noé Gaumont's avatar
Noé Gaumont committed
159
160
        self.webapp.login(user="admin", password=self.password)
        res = self.assertRQLPostOK(queries, version="2.0")
161
        output = res.json
Noé Gaumont's avatar
Noé Gaumont committed
162
163
164
        self.assertEqual(["X"], output[0]["variables"])
        self.assertEqual(["COUNT(X)"], output[1]["variables"])
        self.assertEqual(["X", "COUNT(X)"], output[2]["variables"])
165
166
167
168

        with self.admin_access.client_cnx() as cnx:
            for query, result in zip(queries, output):
                rset = cnx.execute(query[0], query[1])
Noé Gaumont's avatar
Noé Gaumont committed
169
170
                self.assertEqual(rset.variables, result["variables"])
                self.assertEqual(rset.rows, result["rows"])
171

172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
    def test_variables_in_rqlio_get_entities(self):
        ueid = self.admin_access._user.eid
        queries = [
            ("Any X, L WHERE X is CWUser, X eid %(x)s, X login L", {"x": ueid}),
        ]
        self.webapp.login(user="admin", password=self.password)
        res = self.webapp.post(
            "/rqlio/get_entities",
            params=json.dumps([queries, 0]),
            headers={"Content-Type": "application/json"},
        )
        output = res.json

        with self.admin_access.client_cnx() as cnx:
            rset = cnx.execute(queries[0][0], queries[0][1])
            self.assertEqual(rset.one().login, output[0][0].get("login"))

Julien Cristau's avatar
Julien Cristau committed
189

Noé Gaumont's avatar
Noé Gaumont committed
190
if __name__ == "__main__":
Denis Laxalde's avatar
Denis Laxalde committed
191
    import unittest
Noé Gaumont's avatar
Noé Gaumont committed
192

Denis Laxalde's avatar
Denis Laxalde committed
193
    unittest.main()