[tests,auth] Add tests for the auth profiles
Note that only the tests related to the currently loaded auth profile will be run.
This commit is contained in:
amercader
parent
4a7007460b
commit
4fe38ec49d
|
|
@ -0,0 +1,223 @@
|
|||
import logging
|
||||
from pprint import pprint
|
||||
from nose.plugins.skip import SkipTest;
|
||||
|
||||
from ckan import model
|
||||
from ckan.model import Package, Session
|
||||
from ckan.lib.helpers import url_for,json
|
||||
from ckan.lib.base import config
|
||||
|
||||
|
||||
from ckan.tests import CreateTestData
|
||||
from ckan.tests.functional.base import FunctionalTestCase
|
||||
|
||||
from ckanext.harvest.plugin import Harvest
|
||||
from ckanext.harvest.model import HarvestSource, HarvestJob, setup as harvest_model_setup
|
||||
|
||||
log = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class HarvestAuthBaseCase():
|
||||
@classmethod
|
||||
def setup_class(cls):
|
||||
harvest_model_setup()
|
||||
|
||||
@classmethod
|
||||
def teardown_class(cls):
|
||||
pass
|
||||
|
||||
def _test_auth_not_allowed(self,user_name = None, source = None, status = 401):
|
||||
|
||||
if not source:
|
||||
# Create harvest source
|
||||
source = HarvestSource(url=u'http://test-source.com',type='ckan')
|
||||
Session.add(source)
|
||||
Session.commit()
|
||||
|
||||
if user_name:
|
||||
extra_environ = {'REMOTE_USER': user_name.encode('utf8')}
|
||||
else:
|
||||
extra_environ = {}
|
||||
|
||||
# List
|
||||
res = self.app.get('/harvest', status=status, extra_environ=extra_environ)
|
||||
# Create
|
||||
res = self.app.get('/harvest/new', status=status, extra_environ=extra_environ)
|
||||
# Read
|
||||
res = self.app.get('/harvest/%s' % source.id, status=status, extra_environ=extra_environ)
|
||||
# Edit
|
||||
res = self.app.get('/harvest/edit/%s' % source.id, status=status, extra_environ=extra_environ)
|
||||
# Refresh
|
||||
res = self.app.get('/harvest/refresh/%s' % source.id, status=status, extra_environ=extra_environ)
|
||||
|
||||
def _test_auth_allowed(self,user_name,auth_profile=None):
|
||||
|
||||
extra_environ={'REMOTE_USER': user_name.encode('utf8')}
|
||||
|
||||
# List
|
||||
res = self.app.get('/harvest', extra_environ=extra_environ)
|
||||
assert 'Harvesting Sources' in res
|
||||
|
||||
# Create
|
||||
res = self.app.get('/harvest/new', extra_environ=extra_environ)
|
||||
assert 'New harvest source' in res
|
||||
if auth_profile == 'publisher':
|
||||
assert 'publisher_id' in res
|
||||
else:
|
||||
assert not 'publisher_id' in res
|
||||
|
||||
fv = res.forms['source-new']
|
||||
fv['url'] = u'http://test-source.com'
|
||||
fv['type'] = u'ckan'
|
||||
fv['title'] = u'Test harvest source'
|
||||
fv['description'] = u'Test harvest source'
|
||||
fv['config'] = u'{"a":1,"b":2}'
|
||||
|
||||
if auth_profile == 'publisher':
|
||||
fv['publisher_id'] = self.publisher1.id
|
||||
|
||||
res = fv.submit('save', extra_environ=extra_environ)
|
||||
assert not 'Error' in res, res
|
||||
|
||||
source = Session.query(HarvestSource).first()
|
||||
assert source.url == u'http://test-source.com'
|
||||
assert source.type == u'ckan'
|
||||
|
||||
# Read
|
||||
res = self.app.get('/harvest/%s' % source.id, extra_environ=extra_environ)
|
||||
assert 'Harvest Source Details' in res
|
||||
assert source.id in res
|
||||
assert source.title in res
|
||||
|
||||
# Edit
|
||||
res = self.app.get('/harvest/edit/%s' % source.id, extra_environ=extra_environ)
|
||||
assert 'Edit harvest source' in res
|
||||
if auth_profile == 'publisher':
|
||||
assert 'publisher_id' in res
|
||||
else:
|
||||
assert not 'publisher_id' in res
|
||||
|
||||
fv = res.forms['source-new']
|
||||
fv['title'] = u'Test harvest source Updated'
|
||||
|
||||
res = fv.submit('save', extra_environ=extra_environ)
|
||||
assert not 'Error' in res, res
|
||||
|
||||
source = Session.query(HarvestSource).first()
|
||||
assert source.title == u'Test harvest source Updated'
|
||||
|
||||
# Refresh
|
||||
res = self.app.get('/harvest/refresh/%s' % source.id, extra_environ=extra_environ)
|
||||
|
||||
job = Session.query(HarvestJob).first()
|
||||
assert job.source_id == source.id
|
||||
|
||||
|
||||
|
||||
|
||||
class TestAuthDefaultProfile(FunctionalTestCase,HarvestAuthBaseCase):
|
||||
|
||||
@classmethod
|
||||
def setup_class(cls):
|
||||
if (config.get('ckan.harvest.auth.profile','') != ''):
|
||||
raise SkipTest('Skipping default auth profile tests. Set ckan.harvest.auth.profile = \'\' to run them')
|
||||
|
||||
super(TestAuthDefaultProfile,cls).setup_class()
|
||||
|
||||
def setup(self):
|
||||
CreateTestData.create()
|
||||
self.sysadmin_user = model.User.get('testsysadmin')
|
||||
self.normal_user = model.User.get('annafan')
|
||||
|
||||
def teardown(self):
|
||||
model.repo.rebuild_db()
|
||||
|
||||
def test_auth_default_profile_sysadmin(self):
|
||||
self._test_auth_allowed(self.sysadmin_user.name)
|
||||
|
||||
def test_auth_default_profile_normal(self):
|
||||
self._test_auth_not_allowed(self.normal_user.name)
|
||||
|
||||
def test_auth_default_profile_notloggedin(self):
|
||||
self._test_auth_not_allowed(status=302)
|
||||
|
||||
class TestAuthPublisherProfile(FunctionalTestCase,HarvestAuthBaseCase):
|
||||
|
||||
@classmethod
|
||||
def setup_class(cls):
|
||||
if (config.get('ckan.harvest.auth.profile') != 'publisher'):
|
||||
raise SkipTest('Skipping publisher auth profile tests. Set ckan.harvest.auth.profile = \'publisher\' to run them')
|
||||
|
||||
super(TestAuthPublisherProfile,cls).setup_class()
|
||||
|
||||
def setup(self):
|
||||
|
||||
model.Session.remove()
|
||||
CreateTestData.create(auth_profile='publisher')
|
||||
self.sysadmin_user = model.User.get('testsysadmin')
|
||||
self.normal_user = model.User.get('annafan') # Does not belong to a publisher
|
||||
self.publisher1_user = model.User.by_name('russianfan')
|
||||
self.publisher2_user = model.User.by_name('tester')
|
||||
|
||||
# Create two Publishers
|
||||
rev = model.repo.new_revision()
|
||||
self.publisher1 = model.Group(name=u'test-publisher1',title=u'Test Publihser 1',type=u'publisher')
|
||||
Session.add(self.publisher1)
|
||||
self.publisher2 = model.Group(name=u'test-publisher2',title=u'Test Publihser 2',type=u'publisher')
|
||||
Session.add(self.publisher2)
|
||||
|
||||
member1 = model.Member(table_name = 'user',
|
||||
table_id = self.publisher1_user.id,
|
||||
group=self.publisher1,
|
||||
capacity='admin')
|
||||
Session.add(member1)
|
||||
member2 = model.Member(table_name = 'user',
|
||||
table_id = self.publisher2_user.id,
|
||||
group=self.publisher2,
|
||||
capacity='admin')
|
||||
Session.add(member2)
|
||||
|
||||
Session.commit()
|
||||
|
||||
def teardown(self):
|
||||
model.repo.rebuild_db()
|
||||
|
||||
def test_auth_publisher_profile_normal(self):
|
||||
self._test_auth_not_allowed(self.normal_user.name)
|
||||
|
||||
def test_auth_publisher_profile_notloggedin(self):
|
||||
self._test_auth_not_allowed(status=302)
|
||||
|
||||
def test_auth_publisher_profile_sysadmin(self):
|
||||
self._test_auth_allowed(self.sysadmin_user.name,auth_profile='publisher')
|
||||
|
||||
def test_auth_publisher_profile_publisher(self):
|
||||
self._test_auth_allowed(self.publisher1_user.name,auth_profile='publisher')
|
||||
|
||||
def test_auth_publisher_profile_different_publisher(self):
|
||||
|
||||
# Create a source for publisher 1
|
||||
source = HarvestSource(url=u'http://test-source.com',type='ckan',
|
||||
publisher_id=self.publisher1.id)
|
||||
Session.add(source)
|
||||
Session.commit()
|
||||
|
||||
extra_environ = {'REMOTE_USER': self.publisher2_user.name.encode('utf8')}
|
||||
|
||||
# List (Publihsers can see the sources list)
|
||||
res = self.app.get('/harvest', extra_environ=extra_environ)
|
||||
assert 'Harvesting Sources' in res
|
||||
# Create
|
||||
res = self.app.get('/harvest/new', extra_environ=extra_environ)
|
||||
assert 'New harvest source' in res
|
||||
assert 'publisher_id' in res
|
||||
|
||||
# Check that this publihser is not allowed to manage sources from other publishers
|
||||
status = 401
|
||||
# Read
|
||||
res = self.app.get('/harvest/%s' % source.id, status=status, extra_environ=extra_environ)
|
||||
# Edit
|
||||
res = self.app.get('/harvest/edit/%s' % source.id, status=status, extra_environ=extra_environ)
|
||||
# Refresh
|
||||
res = self.app.get('/harvest/refresh/%s' % source.id, status=status, extra_environ=extra_environ)
|
||||
|
||||
Loading…
Reference in New Issue