229 lines
8.0 KiB
Python
229 lines
8.0 KiB
Python
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
|
|
#TODO: remove references to old tests
|
|
try:
|
|
from ckan.tests.functional.base import FunctionalTestCase
|
|
except ImportError:
|
|
from ckan.tests.legacy.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):
|
|
raise SkipTest()
|
|
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)
|
|
|