Skip to content

Commit

Permalink
Equivalent Pylons tests to identify user in request
Browse files Browse the repository at this point in the history
  • Loading branch information
@brew
brew committed Jun 9, 2016
1 parent 42fbc82 commit a644788
Show file tree
Hide file tree
Showing 2 changed files with 143 additions and 62 deletions.
201 changes: 141 additions & 60 deletions ckan/tests/config/test_middleware.py
View file
Expand Up @@ -5,12 +5,12 @@
from nose.tools import assert_not_equals, eq_
from routes import url_for
from flask import Blueprint
from flask import g

import ckan.model as model
import ckan.plugins as p
import ckan.tests.helpers as helpers
import ckan.tests.factories as factories
from ckan.common import c

from ckan.config.middleware import AskAppDispatcherMiddleware
from ckan.config.middleware.flask_app import CKANFlask
Expand Down Expand Up @@ -409,6 +409,146 @@ def test_flask_core_and_pylons_core_route_is_served_by_flask(self):
eq_(res.environ['ckan.app'], 'flask_app')


class TestFlaskUserIdentifiedInRequest(helpers.FunctionalTestBase):

'''Flask identifies user during each request.
API route has been migrated to Flask, so using as an example.
'''

def test_user_objects_in_g_normal_user(self):
'''
A normal logged in user request will have expected user objects added
to request.
'''
self.app = helpers._get_test_app()
flask_app = helpers.find_flask_app(self.app)
user = factories.User()
test_user_obj = model.User.by_name(user['name'])

with flask_app.test_request_context('/api/action/status_show'):
self.app.get(
'/api/action/status_show',
extra_environ={'REMOTE_USER': user['name'].encode('ascii')},)
eq_(c.user, user['name'])
eq_(c.userobj, test_user_obj)
eq_(c.author, user['name'])
eq_(c.remote_addr, 'Unknown IP Address')

def test_user_objects_in_g_anon_user(self):
'''
An anon user request will have expected user objects added to request.
'''
self.app = helpers._get_test_app()
flask_app = helpers.find_flask_app(self.app)

with flask_app.test_request_context('/api/action/status_show'):
self.app.get(
'/api/action/status_show',
extra_environ={'REMOTE_USER': ''},)
eq_(c.user, '')
eq_(c.userobj, None)
eq_(c.author, 'Unknown IP Address')
eq_(c.remote_addr, 'Unknown IP Address')

def test_user_objects_in_g_sysadmin(self):
'''
A sysadmin user request will have expected user objects added to
request.
'''
self.app = helpers._get_test_app()
flask_app = helpers.find_flask_app(self.app)
user = factories.Sysadmin()
test_user_obj = model.User.by_name(user['name'])

with flask_app.test_request_context('/api/action/status_show'):
self.app.get(
'/api/action/status_show',
extra_environ={'REMOTE_USER': user['name'].encode('ascii')},)
eq_(c.user, user['name'])
eq_(c.userobj, test_user_obj)
eq_(c.author, user['name'])
eq_(c.remote_addr, 'Unknown IP Address')


class TestPylonsUserIdentifiedInRequest(helpers.FunctionalTestBase):

'''Pylons identifies user during each request.
Using a route setup via an extension to ensure we're always testing a
Pylons-flavoured request.
'''

def test_user_objects_in_c_normal_user(self):
'''
A normal logged in user request will have expected user objects added
to request.
'''
if not p.plugin_loaded('test_routing_plugin'):
p.load('test_routing_plugin')

app = self._get_test_app()
user = factories.User()
test_user_obj = model.User.by_name(user['name'])

resp = app.get(
'/from_pylons_extension_before_map',
extra_environ={'REMOTE_USER': user['name'].encode('ascii')})

# tmpl_context available on response
eq_(resp.tmpl_context.user, user['name'])
eq_(resp.tmpl_context.userobj, test_user_obj)
eq_(resp.tmpl_context.author, user['name'])
eq_(resp.tmpl_context.remote_addr, 'Unknown IP Address')

p.unload('test_routing_plugin')

def test_user_objects_in_c_anon_user(self):
'''
An anon user request will have expected user objects added to request.
'''
if not p.plugin_loaded('test_routing_plugin'):
p.load('test_routing_plugin')

app = self._get_test_app()

resp = app.get(
'/from_pylons_extension_before_map',
extra_environ={'REMOTE_USER': ''})

# tmpl_context available on response
eq_(resp.tmpl_context.user, '')
eq_(resp.tmpl_context.userobj, None)
eq_(resp.tmpl_context.author, 'Unknown IP Address')
eq_(resp.tmpl_context.remote_addr, 'Unknown IP Address')

p.unload('test_routing_plugin')

def test_user_objects_in_c_sysadmin(self):
'''
A sysadmin user request will have expected user objects added to
request.
'''
if not p.plugin_loaded('test_routing_plugin'):
p.load('test_routing_plugin')

app = self._get_test_app()
user = factories.Sysadmin()
test_user_obj = model.User.by_name(user['name'])

resp = app.get(
'/from_pylons_extension_before_map',
extra_environ={'REMOTE_USER': user['name'].encode('ascii')})

# tmpl_context available on response
eq_(resp.tmpl_context.user, user['name'])
eq_(resp.tmpl_context.userobj, test_user_obj)
eq_(resp.tmpl_context.author, user['name'])
eq_(resp.tmpl_context.remote_addr, 'Unknown IP Address')

p.unload('test_routing_plugin')


class MockRoutingPlugin(p.SingletonPlugin):

p.implements(p.IRoutes)
Expand Down Expand Up @@ -455,62 +595,3 @@ class MockPylonsController(p.toolkit.BaseController):

def view(self):
return 'Hello World, this is served from a Pylons extension'


class TestUserIdentifiedInRequest(helpers.FunctionalTestBase):

'''Flask identifies user during each request.'''

def test_user_objects_in_g_normal_user(self):
'''
A normal logged in user request will have expected user objects added
to request.
'''
self.app = helpers._get_test_app()
flask_app = helpers.find_flask_app(self.app)
user = factories.User()
test_user_obj = model.User.by_name(user['name'])

with flask_app.test_request_context('/api/action/status_show'):
self.app.get(
'/api/action/status_show',
extra_environ={'REMOTE_USER': user['name'].encode('ascii')},)
eq_(g.user, user['name'])
eq_(g.userobj, test_user_obj)
eq_(g.author, user['name'])
eq_(g.remote_addr, 'Unknown IP Address')

def test_user_objects_in_g_anon_user(self):
'''
An anon user request will have expected user objects added to request.
'''
self.app = helpers._get_test_app()
flask_app = helpers.find_flask_app(self.app)

with flask_app.test_request_context('/api/action/status_show'):
self.app.get(
'/api/action/status_show',
extra_environ={'REMOTE_USER': ''},)
eq_(g.user, '')
eq_(g.userobj, None)
eq_(g.author, 'Unknown IP Address')
eq_(g.remote_addr, 'Unknown IP Address')

def test_user_objects_in_g_sysadmin(self):
'''
A sysadmin user request will have expected user objects added to
request.
'''
self.app = helpers._get_test_app()
flask_app = helpers.find_flask_app(self.app)
user = factories.Sysadmin()
test_user_obj = model.User.by_name(user['name'])

with flask_app.test_request_context('/api/action/status_show'):
self.app.get(
'/api/action/status_show',
extra_environ={'REMOTE_USER': user['name'].encode('ascii')},)
eq_(g.user, user['name'])
eq_(g.userobj, test_user_obj)
eq_(g.author, user['name'])
eq_(g.remote_addr, 'Unknown IP Address')
4 changes: 2 additions & 2 deletions ckan/views/__init__.py
View file
@@ -1,8 +1,8 @@
from flask import request, redirect
from flask import redirect

import ckan.model as model
from ckan.views.api import APIKEY_HEADER_NAME_DEFAULT
from ckan.common import c
from ckan.common import c, request
import ckan.plugins as p

import logging
Expand Down

0 comments on commit a644788

Please sign in to comment.