Merge branch 'master' into 4766-allow-empty-arrays-and-objects-in-jso…

…n-fields-with-datastore_create

ckan/authz.py

@@ -206,7 +206,8 @@ def is_authorized(action, context, data_dict=None):
             return {
                 'success': False,
                 'msg': 'Action {0} requires an authenticated user'.format(
-                    action)
+                    (auth_function if not isinstance(auth_function, functools.partial)
+                        else auth_function.func).__name__)
             }
 
         return auth_function(context, data_dict)

ckan/lib/helpers.py

@@ -505,7 +505,7 @@ def _local_url(url_to_amend, **kw):
             default_locale = True
 
     root = ''
-    if kw.get('qualified', False):
+    if kw.get('qualified', False) or kw.get('_external', False):
         # if qualified is given we want the full url ie http://...
         protocol, host = get_site_protocol_and_host()
         root = _routes_default_url_for('/',

ckan/tests/lib/test_helpers.py

@@ -226,18 +226,39 @@ def test_url_for_flask_route_new_syntax_external(self):
         generated_url = h.url_for('api.get_api', ver=3, _external=True)
         eq_(generated_url, url)
 
+    @helpers.change_config('ckan.site_url', 'http://example.com')
+    @helpers.change_config('ckan.root_path', '/{{LANG}}/data')
+    def test_url_for_flask_route_new_syntax_external_with_root_path(self):
+        url = 'http://example.com/data/api/3'
+        generated_url = h.url_for('api.get_api', ver=3, _external=True)
+        eq_(generated_url, url)
+
     @helpers.change_config('ckan.site_url', 'http://example.com')
     def test_url_for_flask_route_old_syntax_external(self):
         url = 'http://example.com/api/3'
         generated_url = h.url_for(controller='api', action='get_api', ver=3, _external=True)
         eq_(generated_url, url)
 
+    @helpers.change_config('ckan.site_url', 'http://example.com')
+    @helpers.change_config('ckan.root_path', '/{{LANG}}/data')
+    def test_url_for_flask_route_old_syntax_external_with_root_path(self):
+        url = 'http://example.com/data/api/3'
+        generated_url = h.url_for(controller='api', action='get_api', ver=3, _external=True)
+        eq_(generated_url, url)
+
     @helpers.change_config('ckan.site_url', 'http://example.com')
     def test_url_for_flask_route_old_syntax_qualified(self):
         url = 'http://example.com/api/3'
         generated_url = h.url_for(controller='api', action='get_api', ver=3, qualified=True)
         eq_(generated_url, url)
 
+    @helpers.change_config('ckan.site_url', 'http://example.com')
+    @helpers.change_config('ckan.root_path', '/{{LANG}}/data')
+    def test_url_for_flask_route_old_syntax_qualified_with_root_path(self):
+        url = 'http://example.com/data/api/3'
+        generated_url = h.url_for(controller='api', action='get_api', ver=3, qualified=True)
+        eq_(generated_url, url)
+
     @helpers.change_config('ckan.site_url', 'http://example.com')
     def test_url_for_flask_route_new_syntax_site_url(self):
         url = '/api/3'

ckanext/datastore/tests/test_chained_auth_functions.py

@@ -1,17 +1,22 @@
 # -*- coding: utf-8 -*-
 import nose
 import ckan.plugins as p
-from ckan.logic import check_access
+from ckan.logic import check_access, NotAuthorized
+from ckan.logic.auth.get import user_list as core_user_list
 import ckan.lib.create_test_data as ctd
 import ckan.tests.helpers as helpers
+import ckan.tests.factories as factories
+from ckan import model
 
 from ckanext.datastore.tests.helpers import DatastoreFunctionalTestBase
 
 assert_equals = nose.tools.assert_equals
 assert_raises = nose.tools.assert_raises
+assert_is_instance = nose.tools.assert_is_instance
 
 
 auth_message = u'No search for you'
+user_list_message = u'Nothing to see here'
 
 
 class TestAuthException(Exception):
@@ -28,15 +33,23 @@ def datastore_search_sql_auth(up_func, context, data_dict):
 
 
 @p.toolkit.chained_auth_function
-def user_create(up_func, context, data_dict):
-    return up_func(context, data_dict)
+def user_list(next_auth, context, data_dict):
+    # check it's received the core function as the first arg
+    assert_equals(next_auth, core_user_list)
+    raise TestAuthException(user_list_message)
+
+
+@p.toolkit.chained_auth_function
+def user_create(next_auth, context, data_dict):
+    return next_auth(context, data_dict)
 
 
 class ExampleDataStoreSearchSQLPlugin(p.SingletonPlugin):
     p.implements(p.IAuthFunctions)
 
     def get_auth_functions(self):
-        return {u'datastore_search_sql': datastore_search_sql_auth}
+        return {u'datastore_search_sql': datastore_search_sql_auth,
+                u'user_list': user_list}
 
 
 class TestChainedAuth(DatastoreFunctionalTestBase):
@@ -54,6 +67,18 @@ def test_datastore_search_sql_auth(self):
         # check that exception returned has the message from our auth function
         assert_equals(raise_context.exception.message, auth_message)
 
+    def test_chain_core_auth_functions(self):
+        user = factories.User()
+        context = {u'user': user[u'name']}
+        with assert_raises(TestAuthException) as raise_context:
+            check_access(u'user_list', context, {})
+        assert_equals(raise_context.exception.message, user_list_message)
+        # check that the 'auth failed' msg doesn't fail because it's a partial
+        assert_raises(NotAuthorized,
+                      lambda: check_access(u'user_list',
+                                           {u'ignore_auth': False,
+                                            u'user': u'not_a_real_user'}, {}))
+
 
 class ExampleExternalProviderPlugin(p.SingletonPlugin):
     p.implements(p.IAuthFunctions)
@@ -71,3 +96,4 @@ def test_user_create_chained_auth(self):
         ctd.CreateTestData.create()
         # check if chained auth fallbacks to built-in user_create
         check_access(u'user_create', {u'user': u'annafan'}, {})
+