fix: bringing back CubicWebServerTC and porting it to pyramid

It was removed in 3.32.0 but it was used by some cubes.
We also replace the wsgiref simpler server by webtest.StopableWSGIServer which
simpler to setup and doesn't try to infer missing content-types.

--HG--
branch : 3.32

cubicweb/devtools/httptest.py

@@ -19,8 +19,13 @@
 http server
 """
 
+import http.client
 import random
 import socket
+from urllib.parse import urlparse
+from webtest.http import StopableWSGIServer
+
+from cubicweb.pyramid.test import _BasePyramidCWTest
 
 
 def get_available_port(ports_scan):
@@ -43,7 +48,7 @@ def get_available_port(ports_scan):
         try:
             s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
             s.connect(("localhost", port))
-        except OSError as err:
+        except socket.error as err:
             if err.args[0] in (111, 106):
                 return port
         finally:
@@ -51,3 +56,96 @@ def get_available_port(ports_scan):
     raise RuntimeError(
         "get_available_port([ports_range]) cannot find an available port"
     )
+
+
+class _CubicWebServerTC(_BasePyramidCWTest):
+    """Base class for running a test web server."""
+
+    ports_range = range(7000, 8000)
+
+    def start_server(self):
+        raise NotImplementedError
+
+    def stop_server(self, timeout=15):
+        """Stop the webserver, waiting for the thread to return"""
+        raise NotImplementedError
+
+    def web_login(self, user=None, passwd=None):
+        """Log the current http session for the provided credential
+
+        If no user is provided, admin connection are used.
+        """
+        if user is None:
+            user = self.admlogin
+            passwd = self.admpassword
+        if passwd is None:
+            passwd = user
+        response = self.web_get("login?__login=%s&__password=%s" % (user, passwd))
+        assert response.status == http.client.SEE_OTHER, response.status
+        self._ident_cookie = response.getheader("Set-Cookie")
+        assert self._ident_cookie
+        return True
+
+    def web_logout(self, user="admin", pwd=None):
+        """Log out current http user"""
+        if self._ident_cookie is not None:
+            self.web_get("logout")
+        self._ident_cookie = None
+
+    def web_request(self, path="", method="GET", body=None, headers=None):
+        """Return an http.client.HTTPResponse object for the specified path
+
+        Use available credential if available.
+        """
+        if headers is None:
+            headers = {}
+        if self._ident_cookie is not None:
+            assert "Cookie" not in headers
+            headers["Cookie"] = self._ident_cookie
+        self._web_test_cnx.request(method, "/" + path, headers=headers, body=body)
+        response = self._web_test_cnx.getresponse()
+        response.body = response.read()  # to chain request
+        response.read = lambda: response.body
+        return response
+
+    def web_get(self, path="", body=None, headers=None):
+        return self.web_request(path=path, body=body, headers=headers)
+
+    def setUp(self):
+        super().setUp()
+        port = get_available_port(self.ports_range)
+        self.config.global_set_option("port", port)  # force rewrite here
+        self.config.global_set_option("base-url", "http://127.0.0.1:%d/" % port)
+        # call load_configuration again to let the config reset its datadir_url
+        self.config.load_configuration()
+        self.start_server()
+
+    def tearDown(self):
+        self.stop_server()
+        super().tearDown()
+
+
+class CubicWebServerTC(_CubicWebServerTC):
+    def start_server(self):
+
+        app = self._generate_pyramid_config().make_wsgi_app()
+
+        self.httpd = StopableWSGIServer.create(app, port=self.config["port"])
+        self.httpd.wait()
+
+        parseurl = urlparse(self.config["base-url"])
+        assert parseurl.port == self.config["port"], (
+            self.config["base-url"],
+            self.config["port"],
+        )
+
+        self._web_test_cnx = http.client.HTTPConnection(
+            parseurl.hostname, parseurl.port
+        )
+        self._ident_cookie = None
+
+    def stop_server(self, timeout=15):
+        if self._web_test_cnx is None:
+            self.web_logout()
+            self._web_test_cnx.close()
+        self.httpd.shutdown()

cubicweb/devtools/test/unittest_httptest.py

+# copyright 2010 LOGILAB S.A. (Paris, FRANCE), all rights reserved.
+# contact https://www.logilab.fr/ -- mailto:contact@logilab.fr
+#
+# This file is part of CubicWeb.
+#
+# CubicWeb 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.
+#
+# CubicWeb 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.
+#
+# You should have received a copy of the GNU Lesser General Public License along
+# with CubicWeb.  If not, see <https://www.gnu.org/licenses/>.
+"""unittest for cubicweb.devtools.httptest module"""
+
+import http.client
+
+from logilab.common.testlib import Tags
+from cubicweb.devtools.httptest import CubicWebServerTC
+
+
+class WsgiCWAnonTC(CubicWebServerTC):
+    settings = {"cubicweb.bwcompat": True}
+
+    def test_response(self):
+        try:
+            response = self.web_get()
+        except http.client.NotConnected as ex:
+            self.fail("Can't connection to test server: %s" % ex)
+
+    def test_response_anon(self):
+        response = self.web_get()
+        self.assertEqual(response.status, http.client.OK)
+
+    def test_base_url(self):
+        if self.config["base-url"] not in self.web_get().read().decode("ascii"):
+            self.fail("no mention of base url in retrieved page")
+
+
+class WsgiCWIdentTC(CubicWebServerTC):
+    settings = {"cubicweb.bwcompat": True}
+    test_db_id = "httptest-cwident"
+    anonymous_allowed = False
+    tags = CubicWebServerTC.tags | Tags(("auth",))
+
+    def test_response_denied(self):
+        response = self.web_get()
+        self.assertEqual(response.status, http.client.FORBIDDEN)
+
+    def test_login(self):
+        response = self.web_get()
+        if response.status != http.client.FORBIDDEN:
+            raise Exception(
+                'Already authenticated, "test_response_denied" must have failed'
+            )
+        # login
+        self.web_login(self.admlogin, self.admpassword)
+        response = self.web_get()
+        self.assertEqual(response.status, http.client.OK, response.body)
+        # logout
+        self.web_logout()
+        response = self.web_get()
+        self.assertEqual(response.status, http.client.FORBIDDEN, response.body)
+
+
+if __name__ == "__main__":
+    from logilab.common.testlib import unittest_main
+
+    unittest_main()

cubicweb/pyramid/test/__init__.py

@@ -112,17 +112,38 @@ class TestApp(webtest.TestApp):
         return response
 
 
-class PyramidCWTest(CubicWebTestTC):
+class _BasePyramidCWTest(CubicWebTestTC):
     settings = {}
 
     @classmethod
     def init_config(cls, config):
-        super(PyramidCWTest, cls).init_config(config)
+        super().init_config(config)
         config.global_set_option("anonymous-user", "anon")
 
+    def _generate_pyramid_config(self):
+        settings = {
+            "cubicweb.bwcompat": False,
+            "cubicweb.session.secret": "test",
+        }
+        settings.update(self.settings)
+        pyramid_config = Configurator(settings=settings)
+
+        pyramid_config.registry["cubicweb.repository"] = self.repo
+        pyramid_config.include("cubicweb.pyramid")
+
+        self.includeme(pyramid_config)
+        self.pyr_registry = pyramid_config.registry
+
+        return pyramid_config
+
+    def includeme(self, config):
+        config.registry.settings["pyramid.csrf_trusted_origins"] = ACCEPTED_ORIGINS
+
+
+class PyramidCWTest(_BasePyramidCWTest):
     def setUp(self):
         # Skip CubicWebTestTC setUp
-        super(CubicWebTestTC, self).setUp()
+        super().setUp()
         settings = {
             "cubicweb.bwcompat": False,
             "cubicweb.session.secret": "test",
@@ -141,6 +162,3 @@ class PyramidCWTest(CubicWebTestTC):
             admin_login=self.admlogin,
             admin_password=self.admpassword,
         )
-
-    def includeme(self, config):
-        config.registry.settings["pyramid.csrf_trusted_origins"] = ACCEPTED_ORIGINS

requirements/test-server.txt

 psycopg2-binary
 ldap3<3,>2
+webtest