Use flask.test_client instead of webtest.test_app

ckan/tests/config/test_middleware.py

@@ -141,21 +141,12 @@ def test_view():
     return app
 
 
-@pytest.mark.skipif(six.PY3, reason=u"Do not test AskAppDispatcherMiddleware in Py3")
-def test_flask_core_route_is_served_by_flask(patched_app):
+def test_flask_core_route_is_served(patched_app):
     res = patched_app.get(u"/")
-    assert res.environ[u"ckan.app"] == u"flask_app"
-
+    assert res.status_code == 200
 
-@pytest.mark.skipif(six.PY3, reason=u"Do not test AskAppDispatcherMiddleware in Py3")
-def test_flask_core_and_pylons_core_route_is_served_by_flask(patched_app):
-    """
-    This should never happen in core, but just in case
-    """
     res = patched_app.get(u"/flask_core")
-
-    assert res.environ[u"ckan.app"] == u"flask_app"
-    assert res.body == u"This was served from Flask"
+    assert six.ensure_text(res.data) == u"This was served from Flask"
 
 
 @pytest.mark.ckan_config(u"ckan.plugins", u"test_routing_plugin")
@@ -235,19 +226,15 @@ def test_ask_around_pylons_extension_route_get_after_map(
             (True, u"pylons_app", u"extension"),
         ]
 
-    @pytest.mark.skipif(six.PY3, reason=u"Do not test AskAppDispatcherMiddleware in Py3")
     def test_flask_extension_route_is_served_by_flask(self, patched_app):
         res = patched_app.get(u"/simple_flask")
-        assert res.environ[u"ckan.app"] == u"flask_app"
+        assert res.status_code == 200
 
     @pytest.mark.skipif(six.PY3, reason=u"Do not test AskAppDispatcherMiddleware in Py3")
     def test_pylons_extension_route_is_served_by_pylons(self, patched_app):
-
         res = patched_app.get(u"/from_pylons_extension_before_map")
-
-        assert res.environ[u"ckan.app"] == u"pylons_app"
         assert (
-            res.body == u"Hello World, this is served from a Pylons extension"
+            res.data == u"Hello World, this is served from a Pylons extension"
         )
 
     @pytest.mark.usefixtures(u"clean_db", u"with_request_context")
@@ -260,7 +247,7 @@ def test_user_objects_in_g_normal_user(self, app):
         test_user_obj = model.User.by_name(username)
         app.get(
             u"/simple_flask",
-            extra_environ={u"REMOTE_USER": username.encode(u"ascii") if six.PY2 else username},
+            environ_overrides={u"REMOTE_USER": username.encode(u"ascii") if six.PY2 else username},
         )
         assert flask.g.user == username
         assert flask.g.userobj == test_user_obj
@@ -273,7 +260,7 @@ def test_user_objects_in_g_anon_user(self, app):
         An anon user request will have expected user objects added to request.
         """
         with app.flask_app.app_context():
-            app.get(u"/simple_flask", extra_environ={u"REMOTE_USER": str(u"")})
+            app.get(u"/simple_flask", environ_overrides={u"REMOTE_USER": str(u"")})
             assert flask.g.user == u""
             assert flask.g.userobj is None
             assert flask.g.author == u"Unknown IP Address"
@@ -289,70 +276,13 @@ def test_user_objects_in_g_sysadmin(self, app):
         test_user_obj = model.User.by_name(user[u"name"])
         app.get(
             u"/simple_flask",
-            extra_environ={u"REMOTE_USER": user[u"name"].encode(u"ascii") if six.PY2 else user[u"name"]},
+            environ_overrides={u"REMOTE_USER": user[u"name"].encode(u"ascii") if six.PY2 else user[u"name"]},
         )
         assert flask.g.user == user[u"name"]
         assert flask.g.userobj == test_user_obj
         assert flask.g.author == user[u"name"]
         assert flask.g.remote_addr == u"Unknown IP Address"
 
-    @pytest.mark.skipif(six.PY3, reason=u"Do not test AskAppDispatcherMiddleware in Py3")
-    def test_user_objects_in_c_normal_user(self, app):
-        """
-        A normal logged in user request will have expected user objects added
-        to request.
-        """
-        username = factories.User()[u"name"]
-        test_user_obj = model.User.by_name(username)
-
-        resp = app.get(
-            u"/from_pylons_extension_before_map",
-            extra_environ={u"REMOTE_USER": username.encode(u"ascii") if six.PY2 else username},
-        )
-
-        # tmpl_context available on response
-        assert resp.tmpl_context.user == username
-        assert resp.tmpl_context.userobj == test_user_obj
-        assert resp.tmpl_context.author == username
-        assert resp.tmpl_context.remote_addr == u"Unknown IP Address"
-
-    @pytest.mark.skipif(six.PY3, reason=u"Do not test AskAppDispatcherMiddleware in Py3")
-    def test_user_objects_in_c_anon_user(self, app):
-        """An anon user request will have expected user objects added to
-        request.
-        """
-
-        resp = app.get(
-            u"/from_pylons_extension_before_map",
-            extra_environ={u"REMOTE_USER": str(u"")},
-        )
-
-        # tmpl_context available on response
-        assert resp.tmpl_context.user == u""
-        assert resp.tmpl_context.userobj is None
-        assert resp.tmpl_context.author == u"Unknown IP Address"
-        assert resp.tmpl_context.remote_addr == u"Unknown IP Address"
-
-    @pytest.mark.skipif(six.PY3, reason=u"Do not test AskAppDispatcherMiddleware in Py3")
-    @pytest.mark.usefixtures(u"clean_db")
-    def test_user_objects_in_c_sysadmin(self, app):
-        """A sysadmin user request will have expected user objects added to
-        request.
-        """
-        username = factories.Sysadmin()[u"name"]
-        test_user_obj = model.User.by_name(username)
-
-        resp = app.get(
-            u"/from_pylons_extension_before_map",
-            extra_environ={u"REMOTE_USER": username.encode(u"ascii") if six.PY2 else username},
-        )
-
-        # tmpl_context available on response
-        assert resp.tmpl_context.user == username
-        assert resp.tmpl_context.userobj == test_user_obj
-        assert resp.tmpl_context.author == username
-        assert resp.tmpl_context.remote_addr == u"Unknown IP Address"
-
     @pytest.mark.skipif(six.PY3, reason=u"Do not test AskAppDispatcherMiddleware in Py3")
     @pytest.mark.ckan_config(
         u"ckan.use_pylons_response_cleanup_middleware", True
@@ -365,9 +295,9 @@ def test_pylons_route_with_cleanup_middleware_activated(self, app):
 
         response = app.get(url=u"/pylons_translated")
 
-        assert response.status_int == 200
+        assert response.status_code == 200
         # make sure we haven't overwritten the response too early.
-        assert u"cleanup middleware" not in response.body
+        assert u"cleanup middleware" not in response.data
 
 
 @pytest.mark.ckan_config(u"SECRET_KEY", u"super_secret_stuff")
@@ -417,8 +347,9 @@ def test_can_handle_request_with_environ(monkeypatch, app, rv, app_base):
 def test_ask_around_is_called(monkeypatch, app):
     ask = mock.MagicMock()
     monkeypatch.setattr(AskAppDispatcherMiddleware, u"ask_around", ask)
-    app.get(u"/", status=404)
+    res = app.get(u"/")
     assert ask.called
+    assert res.status_code == 404
 
 
 @pytest.mark.skipif(six.PY3, reason=u"Do not test AskAppDispatcherMiddleware in Py3")

ckan/tests/config/test_sessions.py

@@ -11,7 +11,7 @@
 
 
 @pytest.mark.ckan_config(u"ckan.plugins", u"test_flash_plugin")
-@pytest.mark.usefixtures("with_request_context")
+@pytest.mark.usefixtures(u"with_request_context")
 class TestWithFlashPlugin:
     # @pytest.mark.skipif(six.PY3, reason=u"There is no pylons app in Py3")
     def test_flash_populated_by_flask_redirect_to_flask(self, app):
@@ -20,15 +20,15 @@ def test_flash_populated_by_flask_redirect_to_flask(self, app):
         view.
         """
         url = u"/flask_add_flash_message_redirect_to_flask"
-        res = app.get(url).follow()
+        res = app.get(url)
         assert body_contains(res, u"This is a success message populated by Flask")
 
     @pytest.mark.skipif(six.PY3, reason=u"There is no pylons app in Py3")
     def test_flash_populated_in_pylons_action_redirect_to_flask(self, app):
         u"""
         Flash store is populated by pylons action is accessible by Flask view.
         """
-        res = app.get(u"/pylons_add_flash_message_redirect_view").follow()
+        res = app.get(u"/pylons_add_flash_message_redirect_view")
 
         assert body_contains(res, u"This is a success message populated by Pylons")
 
@@ -37,7 +37,7 @@ def test_flash_populated_in_flask_view_redirect_to_pylons(self, app):
         u"""
         Flash store is populated by flask view is accessible by pylons action.
         """
-        res = app.get(u"/flask_add_flash_message_redirect_pylons").follow()
+        res = app.get(u"/flask_add_flash_message_redirect_pylons")
 
         assert body_contains(res, u"This is a success message populated by Flask")
 

ckan/tests/helpers.py

@@ -27,6 +27,8 @@
 import re
 import smtplib
 
+from flask.testing import Client as FlaskClient
+
 import webtest
 import nose.tools
 import pytest
@@ -172,7 +174,7 @@ def body_contains(res, content):
     return content in body
 
 
-class CKANTestApp(webtest.TestApp):
+class CKANTestApp(object):
     """A wrapper around webtest.TestApp
 
     It adds some convenience methods for CKAN
@@ -189,13 +191,28 @@ def flask_app(self):
                 self._flask_app = self.app._wsgi_app
         return self._flask_app
 
+    def __init__(self, app):
+        self.app = app
+
+    def test_client(self, use_cookies=True):
+        return CKANTestClient(self.app, self.flask_app.response_class, use_cookies=use_cookies)
+        self.flask_app.test_client_class = CKANTestClient
+        return self.flask_app.test_client()
+
     def post(self, url, *args, **kwargs):
-        url = six.ensure_str(url)
-        return super(CKANTestApp, self).post(url, *args, **kwargs)
+        return self.test_client().post(url, *args, **kwargs)
 
     def get(self, url, *args, **kwargs):
-        url = six.ensure_str(url)
-        return super(CKANTestApp, self).get(url, *args, **kwargs)
+        return self.test_client().get(url, *args, **kwargs)
+
+
+class CKANTestClient(FlaskClient):
+    def open(self, *args, **kwargs):
+        if args and isinstance(args[0], six.string_types):
+            kwargs.setdefault('follow_redirects', True)
+            kwargs.setdefault('base_url', config['ckan.site_url'])
+        result = super(CKANTestClient, self).open(*args, **kwargs)
+        return result
 
 
 def _get_test_app():
@@ -221,6 +238,7 @@ def test_dataset_search(self, app):
     else:
         app = ckan.config.middleware.make_app(config, **config)
     app = CKANTestApp(app)
+
     return app
 
 

ckan/tests/test_none_root.py

@@ -1,13 +1,13 @@
 # encoding: utf-8
 
 import pytest
-
+from ckan.tests.helpers import body_contains
 
 @pytest.mark.ckan_config(u"ckan.root_path", u"/data/{{LANG}}")
 @pytest.mark.ckan_config(u"ckan.plugins", u"example_theme_v15_fanstatic")
 @pytest.mark.usefixtures(u"with_plugins")
 def test_resource_url(app):
     content = app.get(u"/")
-    if u"example_theme.css" not in content:
-        assert u"example_theme.min.css" in content
-    assert u'href="/data/webassets/example_theme' in content
+    if not body_contains(content, u"example_theme.css"):
+        assert body_contains(content, u"example_theme.min.css")
+    assert body_contains(content, u'href="/data/webassets/example_theme')

ckan/tests/test_robots_txt.py

@@ -1,9 +1,11 @@
 # encoding: utf-8
 
+from ckan.tests.helpers import body_contains
 
 def test_robots_txt(app):
-    response = app.get(u"/robots.txt", status=200)
+    res = app.get(u"/robots.txt")
+    assert res.status_code == 200
     assert (
-        response.headers.get(u"Content-Type") == u"text/plain; charset=utf-8"
+        res.headers.get(u"Content-Type") == u"text/plain; charset=utf-8"
     )
-    assert u"User-agent" in response
+    assert body_contains(res, u"User-agent")