fix: update min version of cubicweb and fix test

cubicweb_rqlcontroller/__pkginfo__.py

@@ -14,7 +14,7 @@ description = 'restfull rql edition capabilities'
 web = 'http://www.cubicweb.org/project/%s' % distname
 
 __depends__ = {
-    'cubicweb': '>= 3.27.3, < 3.32',
+    'cubicweb': '>= 3.32.2',
     'six': None,
 }
 __recommends__ = {'cubicweb-signedrequest': None}

test/unittest_rqlcontroller.py

@@ -18,38 +18,32 @@ import json
 from six.moves.urllib import parse as urlparse
 import requests
 
-from cubicweb.devtools.httptest import CubicWebServerTC
+from cubicweb.pyramid.test import PyramidCWTest
 
 
-class RqlIOTC(CubicWebServerTC):
+class RqlIOTC(PyramidCWTest):
+    settings = {"cubicweb.bwcompat": True}
+
+    password = 'gingkow'
 
     def setup_database(self):
         with self.admin_access.client_cnx() as cnx:
             self.create_user(cnx, u'toto', password=u'toto')
             cnx.commit()
 
-    def connectedRQLIOSession(self, login='admin', password='gingkow'):
-        rsession = requests.Session()
-        res = rsession.get('%s?__login=%s&__password=%s' % (
-                           self.config['base-url'], login, password))
-        self.assertEqual(res.status_code, 200)
-        return rsession
-
-    def assertRQLPostOK(self, rsession, queries, code=200, version='1.0'):
-        url = urlparse.urljoin(self.config['base-url'], 'rqlio/%s' % version)
-        res_ok = rsession.post(url,
-                               data=json.dumps(queries),
-                               headers={'Content-Type': 'application/json'})
-        self.assertEqual(res_ok.status_code, code)
-        return res_ok
-
-    def assertRQLPostKO(self, rsession, queries, reason, code=500):
-        url = urlparse.urljoin(self.config['base-url'], 'rqlio/1.0')
-        res_ko = rsession.post(url,
-                               data=json.dumps(queries),
-                               headers={'Content-Type': 'application/json'})
-        self.assertEqual(res_ko.status_code, code)
-        self.assertIn(reason, res_ko.json()[u'reason'])
+    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'},
+                                do_not_grab_the_crsf_token=True)
+
+    def assertRQLPostKO(self, queries, reason, code=500):
+        res_ko = self.webapp.post('/rqlio/1.0',
+                                  params=json.dumps(queries),
+                                  headers={'Content-Type': 'application/json'},
+                                  status=code,
+                                  do_not_grab_the_crsf_token=True)
+        self.assertIn(reason, res_ko.json[u'reason'])
         return res_ko
 
     def test_queries(self):
@@ -60,28 +54,29 @@ class RqlIOTC(CubicWebServerTC):
                     {'u': '__r0', 'g': '__r1'})]
 
         # as an anonymous user
-        rsession = requests.Session()
         reason = (u'You are not allowed to perform add operation on relation'
                   ' CWUser in_group CWGroup')
         # should really be 403 if it wasn't for cubicweb brokenness
-        self.assertRQLPostKO(rsession, queries, reason, code=500)
+        self.assertRQLPostKO(queries, reason, code=500)
 
         # as a standard user
-        rsession = self.connectedRQLIOSession('toto', 'toto')
+        self.webapp.login(user='toto', password='toto')
         reason = (u'You are not allowed to perform add operation on relation'
                   ' CWUser in_group CWGroup')
         # should really be 403 if it wasn't for cubicweb brokenness
-        self.assertRQLPostKO(rsession, queries, reason, code=500)
+        self.assertRQLPostKO(queries, reason, code=500)
 
+        # logout
+        self.webapp.reset()
         # now, as an admin
-        rsession = self.connectedRQLIOSession()
-        res = self.assertRQLPostOK(rsession, queries)
+        self.webapp.login(user='admin', password=self.password)
+        res = self.assertRQLPostOK(queries)
 
         with self.admin_access.client_cnx() as cnx:
             rset = cnx.execute('String N WHERE U in_group G, U login "Babar", '
                                'G name N')
             self.assertEqual('pachyderms', rset.rows[0][0])
-        output = [x for x, in res.json()]
+        output = [x for x, in res.json]
         self.assertEqual(1, len(output[0]))
         self.assertEqual(1, len(output[1]))
         self.assertEqual(2, len(output[2]))
@@ -95,42 +90,39 @@ class RqlIOTC(CubicWebServerTC):
             ('SET U in_group G WHERE U eid %(u)s, G eid %(g)s',
              {'u': '__r0', 'g': '__r1'}),
         ]
-        rsession = self.connectedRQLIOSession()
-        url = urlparse.urljoin(self.config['base-url'], 'rqlio/1.0')
-        files = {
-            'json': ('json', json.dumps(queries), 'application/json'),
-        }
-        response = rsession.post(
-            url, files=files,
-            headers={
-                'Accept': 'application/json',
-            },
-        )
-        self.assertEqual(response.status_code, 200, response.text)
+        self.webapp.login(user='admin', password=self.password)
+        files = [('json', 'loutre.json', json.dumps(queries).encode('utf-8'))]
+
+        self.webapp.post('/rqlio/1.0', upload_files=files,
+                         headers={
+                             'Accept': 'application/json',
+                         },
+                         do_not_grab_the_crsf_token=True
+                         )
 
     def test_rewrite_args_errors(self):
         rql1 = 'Any U WHERE U login %(l)s'
         rql2 = 'SET U in_group G WHERE G name "managers", U eid %(u)s'
         args2 = {'u': '__r0'}
         # setup test
-        rsession = self.connectedRQLIOSession()
+        self.webapp.login(user='admin', password=self.password)
         # check ok
         queries_ok = [(rql1, {'l': 'toto'}), (rql2, args2)]
-        self.assertRQLPostOK(rsession, queries_ok)
+        self.assertRQLPostOK(queries_ok)
         # check ko (1)
         queries_ko = [(rql1, {'l': 'doesnotexist'}),
                       (rql2, args2)]
-        self.assertRQLPostKO(rsession, queries_ko,
+        self.assertRQLPostKO(queries_ko,
                              "__r0 references empty result set")
         # check ko (2)
         queries_ko = [('Any U WHERE U is CWUser', None),
                       (rql2, args2)]
-        self.assertRQLPostKO(rsession, queries_ko,
+        self.assertRQLPostKO(queries_ko,
                              "__r0 references multi lines result set")
         # check ko (3)
         queries_ko = [('Any U,L WHERE U login L, U login %(l)s',
                        {'l': 'toto'}), (rql2, args2)]
-        self.assertRQLPostKO(rsession, queries_ko,
+        self.assertRQLPostKO(queries_ko,
                              "__r0 references multi column result set")
 
     def test_variables_in_rqlio_v2(self):
@@ -140,9 +132,9 @@ class RqlIOTC(CubicWebServerTC):
             ('Any COUNT(X) WHERE X is CWUser', {}),
             ('Any X,COUNT(X) WHERE X is CWUser', {})
         ]
-        rsession = self.connectedRQLIOSession()
-        res = self.assertRQLPostOK(rsession, queries, version='2.0')
-        output = res.json()
+        self.webapp.login(user='admin', password=self.password)
+        res = self.assertRQLPostOK(queries, version='2.0')
+        output = res.json
         self.assertEqual(['X'], output[0]['variables'])
         self.assertEqual(['COUNT(X)'], output[1]['variables'])
         self.assertEqual(['X', 'COUNT(X)'], output[2]['variables'])

tox.ini

@@ -3,6 +3,7 @@ envlist = py3,flake8
 
 [testenv]
 deps =
+  webtest
   pytest
   requests>=1.2
 commands =
@@ -37,4 +38,3 @@ commands =
   python3 setup.py sdist bdist_wheel
   twine check dist/*
   twine upload --skip-existing dist/*
-