Skip to content

Instantly share code, notes, and snippets.

@tobes
Created March 4, 2013 12:05
Show Gist options
  • Star 0 You must be signed in to star a gist
  • Fork 0 You must be signed in to fork a gist
  • Save tobes/5081840 to your computer and use it in GitHub Desktop.
Save tobes/5081840 to your computer and use it in GitHub Desktop.
Another look at tests
from webtest import TestApp
from pylons import config
import ckan.model as model
import ckan.lib.helpers as h
import ckan.config.middleware as middleware
class TestHelper(object):
def __init__(self):
self._app = None
self._sys_admin = None
self._original_config = config.copy()
self._config = config.copy()
self._users = {}
self._orgs = {}
self._datasets = {}
self._related_items = {}
def config_update(self, data):
''' Update the app config and restart the app using that config '''
self._config.update(data)
# reset the app
self._app = None
def config_reset(self):
''' Reset the config to the origional one suppied '''
self._config = self._original_config.copy()
# reset the app
self._app = None
def api_action(self, action, data=None, user=None, status=None):
''' Call the action api and return the dict '''
if not data:
data = {}
url = '/api/action/%s' % action
extra_environ = None
if user:
if isinstance(user, model.User):
apikey = user.apikey
if isinstance(user, dict):
apikey = user['apikey']
extra_environ = {'Authorization': str(apikey)}
resp = self._get_app().post_json(
url, data, status=status, extra_environ=extra_environ)
if resp.status_int == 200:
return resp.json['result']
return resp.json
def get(self, *args, **kw):
''' Request the page via a get call
eg
get('some_url')
get('named_route')
get('named_route', param=something)
get(controller='home', action='index')
returns a webtest response.
'''
status = kw.pop('status', None)
user = kw.pop('user', None)
url = h.url_for(*args, **kw)
return self._get_app().get(url, status=status)
def reset_db(self):
model.repo.rebuild_db()
def create_user(self, data):
name = data['name']
if 'password' not in data:
data['password'] = '%s password' % name
if 'email' not in data:
data['email'] = '%s@example.com' % name
sys_admin = self._get_sysadmin()
result = self.api_action('user_create', data, user=sys_admin)
self._users[name] = result
return result
def user(self, name):
return self._users[name]
def create_org(self, data, user=None):
name = data['name']
result = self.api_action('organization_create', data, user=user)
self._orgs[name] = result
return result
def org(self, name):
return self._orgs[name]
def create_dataset(self, data, user=None):
name = data['name']
result = self.api_action('package_create', data, user=user)
self._datasets[name] = result
return result
def dataset(self, name):
return self._datasets[name]
def create_related(self, data, user=None):
name = data['title']
if 'type' not in data:
data['type'] = 'Application'
result = self.api_action('related_create', data, user=user)
self._related_items[name] = result
return result
def related(self, name):
return self._related_items[name]
def _get_app(self):
if not self._app:
wsgiapp = middleware.make_app(
self._config['global_conf'], **self._config)
self._app = TestApp(wsgiapp)
return self._app
def _get_sysadmin(self):
# Find the sys admin and create if does not exist
if not self._sys_admin:
sys_admin = model.User.get('test_sysadmin')
if not sys_admin:
sys_admin = model.User(name='test_sysadmin', sysadmin=True)
model.Session.add(sys_admin)
model.Session.commit()
model.Session.remove()
self._sys_admin = sys_admin
self._users['test_sysadmin'] = self.api_action(
'user_show', {'id': 'test_sysadmin'}, user=sys_admin)
return self._sys_admin
t = TestHelper()
class TestNew(object):
@classmethod
def setup_class(cls):
t.create_user({'name': 'test_visitor'})
t.create_user({'name': 'test_org_owner'})
t.create_user({'name': 'test_org_member'})
t.create_user({'name': 'test_org_editor'})
t.create_user({'name': 'test_item_owner'})
t.create_org({'name': 'org'}, user=t.user('test_org_owner'))
t.api_action('organization_member_create',
{'id': t.org('org')['id'],
'username': 'test_org_member',
'role': 'member'},
user=t.user('test_org_owner'))
t.api_action('organization_member_create',
{'id': t.org('org')['id'],
'username': 'test_org_editor',
'role': 'editor'},
user=t.user('test_org_owner'))
ds = t.create_dataset({'name': 'dataset',
'owner_org':'org'},
user=t.user('test_org_owner'))
t.create_related({'title': 'rel1',
'dataset_id': ds['id']},
user=t.user('test_org_owner'))
t.create_related({'title': 'rel2',
'dataset_id': ds['id']},
user=t.user('test_org_owner'))
t.create_related({'title': 'rel3',
'dataset_id': ds['id']},
user=t.user('test_org_owner'))
t.create_related({'title': 'rel4',
'dataset_id': ds['id']},
user=t.user('test_org_owner'))
t.create_related({'title': 'rel5',
'dataset_id': ds['id']},
user=t.user('test_org_owner'))
t.create_related({'title': 'rel_no_dataset1'},
user=t.user('test_item_owner'))
t.create_related({'title': 'rel_no_dataset2'},
user=t.user('test_item_owner'))
t.create_related({'title': 'rel_no_dataset3'},
user=t.user('test_item_owner'))
@classmethod
def teardown_class(cls):
t.reset_db()
def test_visitor_cannot_delete_related_item(self):
t.api_action('related_delete',
t.related('rel1'),
user=t.user('test_visitor'),
status=403)
def test_dataset_owner_can_delete_related_item(self):
t.api_action('related_delete',
t.related('rel2'),
user=t.user('test_org_owner'),
status=200)
def test_item_owner_can_delete_related_item(self):
t.api_action('related_delete',
t.related('rel_no_dataset1'),
user=t.user('test_item_owner'),
status=200)
def test_others_cannot_delete_related_item(self):
t.api_action('related_delete',
t.related('rel_no_dataset2'),
user=t.user('test_visitor'),
status=403)
def test_sysadmin_can_delete_related_item(self):
t.api_action('related_delete',
t.related('rel3'),
user=t.user('test_sysadmin'),
status=200)
t.api_action('related_delete',
t.related('rel_no_dataset3'),
user=t.user('test_sysadmin'),
status=200)
def test_organization_member_cannot_delete_related_item(self):
t.api_action('related_delete',
t.related('rel4'),
user=t.user('test_org_member'),
status=403)
def test_organization_editor_can_delete_related_item(self):
t.api_action('related_delete',
t.related('rel5'),
user=t.user('test_org_owner'),
status=200)
@rclark
Copy link

rclark commented Mar 22, 2013

Couple of comments:

  • functions like create_dataset return objects that are dict-style, instead of class-style. e.g.

        pkg = create_dataset({...})
        pkg["id"] => something
        pkg.id => Error, although that's how you would call in in "regular" CKAN
    
  • cache issues: commonly there is a need to purge the database between tests. reset_db is great for this, but things remain in the class cache, e.g. _get_sysadmin will return a reference to a non-existent user.

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment