Update lib tests

.gitignore

@@ -7,6 +7,7 @@ syntax: glob
 ckan.egg-info/*
 sandbox/*
 dist
+.mypy_cache
 
 # pylons
 development.ini*

bin/running_stats.py

@@ -57,7 +57,7 @@ def report_value(self, category):
 
     def report(self, indent=1):
         lines = []
-        categories = self.keys()
+        categories = list(self.keys())
         categories.sort()
         indent_str = '\t' * indent
         for category in categories:

ckan/lib/helpers.py

@@ -503,11 +503,15 @@ def fix_arg(arg):
         url_is_relative = (url.scheme == '' and url.netloc == '' and
                            not url.path.startswith('/'))
         if url_is_relative:
-            return '/' + url.geturl()
-        return url.geturl()
+            return False, '/' + url.geturl()
+
+        return bool(url.scheme), url.geturl()
 
     if args:
-        args = (fix_arg(args[0]), ) + args[1:]
+        is_external, fixed_url = fix_arg(args[0])
+        if is_external:
+            return fixed_url
+        args = (fixed_url, ) + args[1:]
     if kw.get('qualified', False):
         kw['protocol'], kw['host'] = get_site_protocol_and_host()
     kw['locale'] = 'default'

ckan/lib/i18n.py

@@ -346,7 +346,7 @@ def _build_js_translation(lang, source_filenames, entries, dest_filename):
                     plural.append(msgstr)
     with open(dest_filename, u'w') as f:
         s = json.dumps(result, sort_keys=True, indent=2, ensure_ascii=False)
-        f.write(s.encode(u'utf-8'))
+        f.write(six.ensure_str(s))
 
 
 def build_js_translations():

ckan/lib/navl/dictization_functions.py

@@ -46,7 +46,7 @@ class State(object):
 
 class DictizationError(Exception):
     def __str__(self):
-        return text_type(self).encode('utf8')
+        return six.ensure_str(self.__unicode__())
 
     def __unicode__(self):
         if hasattr(self, 'error') and self.error:
@@ -267,7 +267,7 @@ def validate(data, schema, context=None):
 
     # create a copy of the context which also includes the schema keys so
     # they can be used by the validators
-    validators_context = dict(context, schema_keys=schema.keys())
+    validators_context = dict(context, schema_keys=list(schema.keys()))
 
     flattened = flatten_dict(data)
     converted_data, errors = _validate(flattened, schema, validators_context)

ckan/tests/config/test_middleware.py

@@ -250,24 +250,22 @@ def test_pylons_extension_route_is_served_by_pylons(self, patched_app):
             res.body == u"Hello World, this is served from a Pylons extension"
         )
 
-    @pytest.mark.usefixtures(u"clean_db")
+    @pytest.mark.usefixtures(u"clean_db", u"with_request_context")
     def test_user_objects_in_g_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)
-
-        with app.flask_app.app_context():
-            app.get(
-                u"/simple_flask",
-                extra_environ={u"REMOTE_USER": username.encode(u"ascii") if six.PY2 else username},
-            )
-            assert flask.g.user == username
-            assert flask.g.userobj == test_user_obj
-            assert flask.g.author == username
-            assert flask.g.remote_addr == u"Unknown IP Address"
+        app.get(
+            u"/simple_flask",
+            extra_environ={u"REMOTE_USER": username.encode(u"ascii") if six.PY2 else username},
+        )
+        assert flask.g.user == username
+        assert flask.g.userobj == test_user_obj
+        assert flask.g.author == username
+        assert flask.g.remote_addr == u"Unknown IP Address"
 
     @pytest.mark.usefixtures(u"clean_db")
     def test_user_objects_in_g_anon_user(self, app):
@@ -281,24 +279,22 @@ def test_user_objects_in_g_anon_user(self, app):
             assert flask.g.author == u"Unknown IP Address"
             assert flask.g.remote_addr == u"Unknown IP Address"
 
-    @pytest.mark.usefixtures(u"clean_db")
+    @pytest.mark.usefixtures(u"clean_db", u"with_request_context")
     def test_user_objects_in_g_sysadmin(self, app):
         """
         A sysadmin user request will have expected user objects added to
         request.
         """
         user = factories.Sysadmin()
         test_user_obj = model.User.by_name(user[u"name"])
-
-        with app.flask_app.app_context():
-            app.get(
-                u"/simple_flask",
-                extra_environ={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"
+        app.get(
+            u"/simple_flask",
+            extra_environ={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):

ckan/tests/helpers.py

@@ -121,12 +121,6 @@ def call_action(action_name, context=None, **kwargs):
     context.setdefault("user", "127.0.0.1")
     context.setdefault("ignore_auth", True)
 
-    if six.PY3:
-        from ckan.lib.helpers import _get_auto_flask_context
-        _auto_flask_context = _get_auto_flask_context()
-        if _auto_flask_context:
-            _auto_flask_context.push()
-
     return logic.get_action(action_name)(context=context, data_dict=kwargs)
 
 

ckan/tests/lib/dictization/test_model_dictize.py

@@ -11,7 +11,7 @@
 from ckan.tests import helpers, factories
 
 
-@pytest.mark.usefixtures("clean_db", "clean_index")
+@pytest.mark.usefixtures("clean_db", "clean_index", "with_request_context")
 class TestGroupListDictize:
     def test_group_list_dictize(self):
         group = factories.Group()
@@ -134,7 +134,7 @@ def test_group_list_dictize_including_groups(self):
         assert child_dict["groups"][0]["name"] == "parent"
 
 
-@pytest.mark.usefixtures("clean_db", "clean_index")
+@pytest.mark.usefixtures("clean_db", "clean_index", "with_request_context")
 class TestGroupDictize:
     def test_group_dictize(self):
         group = factories.Group(name="test_dictize")
@@ -325,11 +325,11 @@ def test_group_dictize_for_org_with_package_count(self):
         assert org["package_count"] == 1
 
 
-@pytest.mark.usefixtures("clean_db")
+@pytest.mark.usefixtures("clean_db", "with_request_context")
 class TestPackageDictize:
     def remove_changable_values(self, dict_):
         dict_ = copy.deepcopy(dict_)
-        for key, value in dict_.items():
+        for key, value in list(dict_.items()):
             if key.endswith("id") and key != "license_id":
                 dict_.pop(key)
             if key == "created":
@@ -642,7 +642,7 @@ def test_vocabulary_dictize_not_including_datasets(self):
             assert len(tag.get("packages", [])) == 0
 
 
-@pytest.mark.usefixtures("clean_db")
+@pytest.mark.usefixtures("clean_db", "with_request_context")
 class TestActivityDictize(object):
     def test_include_data(self):
         dataset = factories.Dataset()

ckan/tests/lib/navl/test_dictization_functions.py

@@ -1,6 +1,7 @@
 # encoding: utf-8
 
-import nose
+import pytest
+import six
 from six import text_type
 from ckan.lib.navl.dictization_functions import validate, Invalid
 
@@ -44,6 +45,7 @@ def test_str_error(self):
         err_obj = Invalid("Some ascii error")
         assert str(err_obj) == "Invalid: 'Some ascii error'"
 
+    @pytest.mark.skipif(six.PY3, reason="Skip unicode checks in Py3")
     def test_unicode_error(self):
         err_obj = Invalid("Some ascii error")
         assert text_type(err_obj) == u"Invalid: 'Some ascii error'"
@@ -54,14 +56,17 @@ def test_repr_error(self):
 
     # Error msgs should be ascii, but let's just see what happens for unicode
 
+    @pytest.mark.skipif(six.PY3, reason="Skip unicode checks in Py3")
     def test_str_unicode_error(self):
         err_obj = Invalid(u"Some unicode \xa3 error")
         assert str(err_obj) == "Invalid: u'Some unicode \\xa3 error'"
 
+    @pytest.mark.skipif(six.PY3, reason="Skip unicode checks in Py3")
     def test_unicode_unicode_error(self):
         err_obj = Invalid(u"Some unicode \xa3 error")
         assert text_type(err_obj) == "Invalid: u'Some unicode \\xa3 error'"
 
+    @pytest.mark.skipif(six.PY3, reason="Skip unicode checks in Py3")
     def test_repr_unicode_error(self):
         err_obj = Invalid(u"Some unicode \xa3 error")
         assert repr(err_obj) == "<Invalid u'Some unicode \\xa3 error'>"
@@ -70,6 +75,7 @@ def test_str_blank(self):
         err_obj = Invalid("")
         assert str(err_obj) == "Invalid"
 
+    @pytest.mark.skipif(six.PY3, reason="Skip unicode checks in Py3")
     def test_unicode_blank(self):
         err_obj = Invalid("")
         assert text_type(err_obj) == u"Invalid"

ckan/tests/lib/search/test_index.py

@@ -5,6 +5,7 @@
 import json
 import nose
 import pytest
+import six
 from ckan.common import config
 import ckan.lib.search as search
 
@@ -46,9 +47,9 @@ def test_index_basic(self):
         assert response.docs[0]["title"] == "Monkey"
 
         index_id = hashlib.md5(
-            "{0}{1}".format(
+            six.b("{0}{1}".format(
                 self.base_package_dict["id"], config["ckan.site_id"]
-            )
+            ))
         ).hexdigest()
 
         assert response.docs[0]["index_id"] == index_id

ckan/tests/lib/test_base.py

@@ -1,6 +1,6 @@
 # encoding: utf-8
 
-
+import six
 import pytest
 import ckan.tests.helpers as helpers
 
@@ -26,15 +26,15 @@ def test_apikey_missing(app):
     app.get("/dataset/new", headers=request_headers, status=403)
 
 
-@pytest.mark.usefixtures("clean_db")
+@pytest.mark.usefixtures("clean_db", "with_request_context")
 def test_apikey_in_authorization_header(app):
     user = factories.Sysadmin()
     request_headers = {"Authorization": str(user["apikey"])}
 
     app.get("/dataset/new", headers=request_headers)
 
 
-@pytest.mark.usefixtures("clean_db")
+@pytest.mark.usefixtures("clean_db", "with_request_context")
 def test_apikey_in_x_ckan_header(app):
     user = factories.Sysadmin()
     # non-standard header name is defined in test-core.ini
@@ -53,7 +53,7 @@ def test_apikey_contains_unicode(app):
 
 def test_options(app):
     response = app.options(url="/", status=200)
-    assert len(str(response.body)) == 0, "OPTIONS must return no content"
+    assert len(six.ensure_str(response.body)) == 0, "OPTIONS must return no content"
 
 
 def test_cors_config_no_cors(app):
@@ -230,7 +230,7 @@ def test_cors_config_origin_allow_all_false_with_whitelist_not_containing_origin
 @pytest.mark.usefixtures("with_plugins")
 def test_options_2(app):
     response = app.options(url="/simple_flask", status=200)
-    assert len(str(response.body)) == 0, "OPTIONS must return no content"
+    assert len(six.ensure_str(response.body)) == 0, "OPTIONS must return no content"
 
 
 @pytest.mark.ckan_config("ckan.plugins", "test_routing_plugin")

ckan/tests/lib/test_changes.py

@@ -106,6 +106,7 @@ def test_change_extra(self):
         assert changes[0]["old_value"] == u"science"
         assert changes[0]["new_value"] == u"scientific"
 
+
     def test_change_multiple_extras(self):
         changes = []
         original = Dataset(
@@ -132,7 +133,7 @@ def test_change_multiple_extras(self):
             if change["key"] == u"subject":
                 assert change["new_value"] == u"scientific"
             else:
-                assert changes[0]["key"] == u"topic"
+                assert change["key"] == u"topic"
                 assert change["new_value"] == u"rain"
 
     # TODO how to test change2?
@@ -259,6 +260,7 @@ def test_remove_notes(self):
         assert changes[0]["method"] == u"remove"
 
     @pytest.mark.ckan_config(u"ckan.auth.create_unowned_dataset", True)
+    @pytest.mark.usefixtures(u"with_request_context")
     def test_add_org(self):
         changes = []
         original = Dataset(owner_org=None)
@@ -273,6 +275,7 @@ def test_add_org(self):
         assert changes[0]["method"] == u"add"
         assert changes[0]["new_org_id"] == new_org["id"]
 
+    @pytest.mark.usefixtures(u"with_request_context")
     def test_change_org(self):
         changes = []
         old_org = Organization()
@@ -290,6 +293,7 @@ def test_change_org(self):
         assert changes[0]["new_org_id"] == new_org["id"]
 
     @pytest.mark.ckan_config(u"ckan.auth.create_unowned_dataset", True)
+    @pytest.mark.usefixtures(u"with_request_context")
     def test_remove_org(self):
         changes = []
         old_org = Organization()
@@ -308,6 +312,7 @@ def test_remove_org(self):
         assert changes[0]["type"] == u"org"
         assert changes[0]["method"] == u"remove"
 
+    @pytest.mark.usefixtures(u"with_request_context")
     def test_make_private(self):
         changes = []
         old_org = Organization()
@@ -322,6 +327,7 @@ def test_make_private(self):
         assert changes[0]["type"] == u"private"
         assert changes[0]["new"] == u"Private"
 
+    @pytest.mark.usefixtures(u"with_request_context")
     def test_make_public(self):
         changes = []
         old_org = Organization()

ckan/tests/lib/test_config_tool.py

@@ -224,14 +224,14 @@ def test_parse_sections(self):
             "\n"
         )
 
-        out = config_tool.parse_config(input_lines)
+        out = sorted(config_tool.parse_config(input_lines).items())
 
         assert (
-            str(out) == "{"
-            "'logger-keys': <Option [logger] keys = root, ckan, ckanext>, "
-            "'app:main-ckan.site_title': <Option [app:main] ckan.site_title = CKAN>, "
-            "'logger-level': <Option [logger] level = WARNING>"
-            "}"
+            str(out) == "["
+            "('app:main-ckan.site_title', <Option [app:main] ckan.site_title = CKAN>), "
+            "('logger-keys', <Option [logger] keys = root, ckan, ckanext>), "
+            "('logger-level', <Option [logger] level = WARNING>)"
+            "]"
         )
 
     def test_parse_comment(self):

ckan/tests/lib/test_datapreview.py

@@ -125,7 +125,7 @@ def test_both_plugins_in_config_only_datastore(self):
 
 
 @pytest.mark.ckan_config("ckan.plugins", "image_view test_datastore_view")
-@pytest.mark.usefixtures("clean_db", "with_plugins")
+@pytest.mark.usefixtures("clean_db", "with_plugins", "with_request_context")
 class TestDatapreview(object):
     def test_get_view_plugins(self):