Try   HackMD

Test Automation of CKAN Customize Extension

目前,已經自行建立了一個 ckanext-ytdataservice extension
接著,我們以下兩篇的官方教學與範例程式碼為基底
實做客製化套件 (e.g., ytdataservice) 的自動化測試,結果不含CKAN核心測試

相關檔案列表

以下是要使用到的檔案列表,我們假設所在目錄為 ~/ckan/lib/default/src/

新增:

  • ./ckanext-ytdataservice/ckanext/ytdataservice/tests/test_ytdataservice.py
  • ./ckanext-ytdataservice/ckanext/ytdataservice/plugin_v4.py
  • ./ckanext-ytdataservice/ckanext/ytdataservice/plugin_v5_custom_config_setting.py
  • ./ckanext-ytdataservice/ckanext/ytdataservice/plugin_v6_parent_auth_functions.py

修改:

  • ./ckan/test-core.ini
  • ./ckan/setup.py

參考:

  • ./ckanext-ytdataservice/ckanext/ytdataservice/plugin.py
  • 特別注意,這個跟假設所在目錄不同 ~/ckan/etc/default/development.ini
  • Installing CKAN from source

前置準備

  1. 架設好CKAN 2.7.2的 Development 環境
  2. 已經自行創建名為 ytdataservice 的extension,做到 We're done!,確認完成plugin.py
  3. Extension的所在目錄為 ~/ckan/lib/default/src/ckanext-ytdataservice,就不用改路徑
  4. 確認在python虛擬環境(Virtualenv),再執行指令
  5. 進入虛擬環境後在 ~/ckan/lib/default/src/ckan/ ~/ckan/lib/default/src/ckanext-ytdataservice/ 目錄下
    執行以下指令,初始設定開發環境
    ​​​​python setup.py develop
    

初始化測試環境

  1. 基於development.ini修改 test-core.ini

    ~/ckan/lib/default/src/ckan/test-core.ini

    只列出要修改的部分

    ​​​​# Specify the Postgres database for SQLAlchemy to use
    ​​​​sqlalchemy.url = postgresql://ckan_default:<YOUR_PASSWORD>@localhost/ckan_test
    
    ​​​​## Datastore
    ​​​​ckan.datastore.write_url = postgresql://ckan_default:<YOUR_PASSWORD>@localhost/datastore_test
    ​​​​ckan.datastore.read_url = postgresql://datastore_default:<YOUR_PASSWORD>@localhost/datastore_test
    

    官網有特別提醒,測試用的Redis的設定要與Production分離
    但其實預設設定不管Development還是Production都沒開啟,所以還不用擔心,之後注意就好

  2. 完成到Configure Solr Multi-core 這一步

    • Set the permissions 這裡是在~/ckan/lib/default/src/ckan/下執行
  3. 初始化測試用資料庫,在 ~/ckan/lib/default/src/ckan目錄下,執行以下指令

    ​​​​paster db init -c test-core.ini
    

撰寫測試檔

若是完全按照官網新增ckanext-iauthfunctions的 extension
可能就不用置換名詞與修改Setup.py
~/ckan/lib/default/src/ckan/ckanext/example_iauthfunctions 也已經有自動化測試程式碼
本文件是希望切割已有的範例檔,獨自跑過一遍

  1. 基於這篇官方文件的test_iauthfunctions.py
    置換example_iauthfunctions成ytdataservice,新增並改寫成 test_ytdataservice.py
    其他程式碼都不變,都貼上只是希望方便參考,因為要改的地方不少

    ./ckanext-ytdataservice/ckanext/ytdataservice/tests/test_ytdataservice.py

    ​​​​# encoding: utf-8 ​​​​'''Tests for the ckanext.ytdataservice extension. ​​​​''' ​​​​from nose.tools import assert_raises ​​​​from nose.tools import assert_equal ​​​​import ckan.model as model ​​​​import ckan.plugins ​​​​from ckan.plugins.toolkit import NotAuthorized, ObjectNotFound ​​​​import ckan.tests.factories as factories ​​​​import ckan.logic as logic ​​​​import ckan.tests.helpers as helpers ​​​​class TestytdataservicePluginV6ParentAuthFunctions(object): ​​​​ '''Tests for the ckanext.ytdataservice.plugin module. ​​​​ Specifically tests that overriding parent auth functions will cause ​​​​ child auth functions to use the overridden version. ​​​​ ''' ​​​​ @classmethod ​​​​ def setup_class(cls): ​​​​ '''Nose runs this method once to setup our test class.''' ​​​​ # Test code should use CKAN's plugins.load() function to load plugins ​​​​ # to be tested. ​​​​ ckan.plugins.load('ytdataservice_v6_parent_auth_functions') ​​​​ def teardown(self): ​​​​ '''Nose runs this method after each test method in our test class.''' ​​​​ # Rebuild CKAN's database after each test method, so that each test ​​​​ # method runs with a clean slate. ​​​​ model.repo.rebuild_db() ​​​​ @classmethod ​​​​ def teardown_class(cls): ​​​​ '''Nose runs this method once after all the test methods in our class ​​​​ have been run. ​​​​ ''' ​​​​ # We have to unload the plugin we loaded, so it doesn't affect any ​​​​ # tests that run after ours. ​​​​ ckan.plugins.unload('ytdataservice_v6_parent_auth_functions') ​​​​ def test_resource_delete_editor(self): ​​​​ '''Normally organization admins can delete resources ​​​​ Our plugin prevents this by blocking delete organization. ​​​​ Ensure the delete button is not displayed (as only resource delete ​​​​ is checked for showing this) ​​​​ ''' ​​​​ user = factories.User() ​​​​ owner_org = factories.Organization( ​​​​ users=[{'name': user['id'], 'capacity': 'admin'}] ​​​​ ) ​​​​ dataset = factories.Dataset(owner_org=owner_org['id']) ​​​​ resource = factories.Resource(package_id=dataset['id']) ​​​​ with assert_raises(logic.NotAuthorized) as e: ​​​​ logic.check_access('resource_delete', {'user': user['name']}, {'id': resource['id']}) ​​​​ assert_equal(e.exception.message, 'User %s not authorized to delete resource %s' % (user['name'], resource['id'])) ​​​​ def test_resource_delete_sysadmin(self): ​​​​ '''Normally organization admins can delete resources ​​​​ Our plugin prevents this by blocking delete organization. ​​​​ Ensure the delete button is not displayed (as only resource delete ​​​​ is checked for showing this) ​​​​ ''' ​​​​ user = factories.Sysadmin() ​​​​ owner_org = factories.Organization( ​​​​ users=[{'name': user['id'], 'capacity': 'admin'}] ​​​​ ) ​​​​ dataset = factories.Dataset(owner_org=owner_org['id']) ​​​​ resource = factories.Resource(package_id=dataset['id']) ​​​​ assert_equal(logic.check_access('resource_delete', {'user': user['name']}, {'id': resource['id']}), True) ​​​​class TestytdataserviceCustomConfigSetting(object): ​​​​ '''Tests for the plugin_v5_custom_config_setting module. ​​​​ ''' ​​​​ @classmethod ​​​​ def setup_class(cls): ​​​​ if not ckan.plugins.plugin_loaded('ytdataservice_v5_custom_config_setting'): ​​​​ ckan.plugins.load('ytdataservice_v5_custom_config_setting') ​​​​ @classmethod ​​​​ def teardown_class(cls): ​​​​ ckan.plugins.unload('ytdataservice_v5_custom_config_setting') ​​​​ def teardown(self): ​​​​ # Delete any stuff that's been created in the db, so it doesn't ​​​​ # interfere with the next test. ​​​​ model.repo.rebuild_db() ​​​​ @helpers.change_config('ckan.ytdataservice.users_can_create_groups', False) ​​​​ def test_sysadmin_can_create_group_when_config_is_False(self): ​​​​ sysadmin = factories.Sysadmin() ​​​​ context = { ​​​​ 'ignore_auth': False, ​​​​ 'user': sysadmin['name'] ​​​​ } ​​​​ helpers.call_action('group_create', context, name='test-group') ​​​​ @helpers.change_config('ckan.ytdataservice.users_can_create_groups', False) ​​​​ def test_user_cannot_create_group_when_config_is_False(self): ​​​​ user = factories.User() ​​​​ context = { ​​​​ 'ignore_auth': False, ​​​​ 'user': user['name'] ​​​​ } ​​​​ assert_raises( ​​​​ NotAuthorized, helpers.call_action, 'group_create', ​​​​ context, name='test-group') ​​​​ @helpers.change_config('ckan.ytdataservice.users_can_create_groups', False) ​​​​ def test_visitor_cannot_create_group_when_config_is_False(self): ​​​​ context = { ​​​​ 'ignore_auth': False, ​​​​ 'user': None ​​​​ } ​​​​ assert_raises( ​​​​ NotAuthorized, helpers.call_action, 'group_create', ​​​​ context, name='test-group') ​​​​ @helpers.change_config('ckan.ytdataservice.users_can_create_groups', True) ​​​​ def test_sysadmin_can_create_group_when_config_is_True(self): ​​​​ sysadmin = factories.Sysadmin() ​​​​ context = { ​​​​ 'ignore_auth': False, ​​​​ 'user': sysadmin['name'] ​​​​ } ​​​​ helpers.call_action('group_create', context, name='test-group') ​​​​ @helpers.change_config('ckan.ytdataservice.users_can_create_groups', True) ​​​​ def test_user_can_create_group_when_config_is_True(self): ​​​​ user = factories.User() ​​​​ context = { ​​​​ 'ignore_auth': False, ​​​​ 'user': user['name'] ​​​​ } ​​​​ helpers.call_action('group_create', context, name='test-group') ​​​​ @helpers.change_config('ckan.ytdataservice.users_can_create_groups', True) ​​​​ def test_visitor_cannot_create_group_when_config_is_True(self): ​​​​ context = { ​​​​ 'ignore_auth': False, ​​​​ 'user': None ​​​​ } ​​​​ assert_raises( ​​​​ NotAuthorized, helpers.call_action, 'group_create', ​​​​ context, name='test-group') ​​​​class BaseTest(object): ​​​​ def teardown(self): ​​​​ # Rebuild CKAN's database after each test method, so that each test ​​​​ # method runs with a clean slate. ​​​​ model.repo.rebuild_db() ​​​​ def _make_curators_group(self): ​​​​ '''This is a helper method for test methods to call when they want ​​​​ the 'curators' group to be created. ​​​​ ''' ​​​​ sysadmin = factories.Sysadmin() ​​​​ # Create a user who will *not* be a member of the curators group. ​​​​ noncurator = factories.User() ​​​​ # Create a user who will be a member of the curators group. ​​​​ curator = factories.User() ​​​​ # Create the curators group, with the 'curator' user as a member. ​​​​ users = [{'name': curator['name'], 'capacity': 'member'}] ​​​​ context = { ​​​​ 'ignore_auth': False, ​​​​ 'user': sysadmin['name'] ​​​​ } ​​​​ curators_group = helpers.call_action( ​​​​ 'group_create', context, name='curators', users=users) ​​​​ return (noncurator, curator, curators_group) ​​​​class TestytdataservicePluginV4(BaseTest): ​​​​ '''Tests for the ckanext.ytdataservice.plugin module. ​​​​ ''' ​​​​ @classmethod ​​​​ def setup_class(cls): ​​​​ '''Nose runs this method once to setup our test class.''' ​​​​ # Test code should use CKAN's plugins.load() function to load plugins ​​​​ # to be tested. ​​​​ if not ckan.plugins.plugin_loaded('ytdataservice_v4'): ​​​​ ckan.plugins.load('ytdataservice_v4') ​​​​ @classmethod ​​​​ def teardown_class(cls): ​​​​ '''Nose runs this method once after all the test methods in our class ​​​​ have been run. ​​​​ ''' ​​​​ # We have to unload the plugin we loaded, so it doesn't affect anyexample_iauthfunctions ​​​​ # tests that run after ours. ​​​​ ckan.plugins.unload('ytdataservice_v4') ​​​​ def test_group_create_with_no_curators_group(self): ​​​​ '''Test that group_create doesn't crash when there's no curators group. ​​​​ ''' ​​​​ sysadmin = factories.Sysadmin() ​​​​ # Make sure there's no curators group. ​​​​ assert 'curators' not in helpers.call_action('group_list', {}) ​​​​ # Make our sysadmin user create a group. CKAN should not crash. ​​​​ context = { ​​​​ 'ignore_auth': False, ​​​​ 'user': sysadmin['name'] ​​​​ } ​​​​ helpers.call_action('group_create', context, name='test-group') ​​​​ def test_group_create_with_visitor(self): ​​​​ '''A visitor (not logged in) should not be able to create a group. ​​​​ Note: this also tests that the group_create auth function doesn't ​​​​ crash when the user isn't logged in. ​​​​ ''' ​​​​ noncurator, curator, curators_group = self._make_curators_group() ​​​​ context = { ​​​​ 'ignore_auth': False, ​​​​ 'user': None ​​​​ } ​​​​ assert_raises( ​​​​ NotAuthorized, helpers.call_action, 'group_create', ​​​​ context, name='this_group_should_not_be_created') ​​​​ def test_group_create_with_non_curator(self): ​​​​ '''A user who isn't a member of the curators group should not be able ​​​​ to create a group. ​​​​ ''' ​​​​ noncurator, curator, curators_group = self._make_curators_group() ​​​​ context = { ​​​​ 'ignore_auth': False, ​​​​ 'user': noncurator['name'] ​​​​ } ​​​​ assert_raises( ​​​​ NotAuthorized, helpers.call_action, 'group_create', ​​​​ context, name='this_group_should_not_be_created') ​​​​ def test_group_create_with_curator(self): ​​​​ '''A member of the curators group should be able to create a group. ​​​​ ''' ​​​​ noncurator, curator, curators_group = self._make_curators_group() ​​​​ name = 'my-new-group' ​​​​ context = { ​​​​ 'ignore_auth': False, ​​​​ 'user': curator['name'] ​​​​ } ​​​​ result = helpers.call_action( ​​​​ 'group_create', context, name=name) ​​​​ assert result['name'] == name
  2. 參考這裡的程式碼,修改並於對應目錄新增以下檔案:
    plugin_v4.py、plugin_v5_custom_config_setting.py 與 plugin_v6_parent_auth_functions.py

    ./ckanext-ytdataservice/ckanext/ytdataservice/plugin_v4.py

    ​​​​# encoding: utf-8 ​​​​import ckan.plugins as plugins ​​​​import ckan.plugins.toolkit as toolkit ​​​​def group_create(context, data_dict=None): ​​​​ # Get the user name of the logged-in user. ​​​​ user_name = context['user'] ​​​​ # Get a list of the members of the 'curators' group. ​​​​ try: ​​​​ members = toolkit.get_action('member_list')( ​​​​ data_dict={'id': 'curators', 'object_type': 'user'}) ​​​​ except toolkit.ObjectNotFound: ​​​​ # The curators group doesn't exist. ​​​​ return {'success': False, ​​​​ 'msg': "The curators groups doesn't exist, so only sysadmins " ​​​​ "are authorized to create groups."} ​​​​ # 'members' is a list of (user_id, object_type, capacity) tuples, we're ​​​​ # only interested in the user_ids. ​​​​ member_ids = [member_tuple[0] for member_tuple in members] ​​​​ # We have the logged-in user's user name, get their user id. ​​​​ convert_user_name_or_id_to_id = toolkit.get_converter( ​​​​ 'convert_user_name_or_id_to_id') ​​​​ try: ​​​​ user_id = convert_user_name_or_id_to_id(user_name, context) ​​​​ except toolkit.Invalid: ​​​​ # The user doesn't exist (e.g. they're not logged-in). ​​​​ return {'success': False, ​​​​ 'msg': 'You must be logged-in as a member of the curators ' ​​​​ 'group to create new groups.'} ​​​​ # Finally, we can test whether the user is a member of the curators group. ​​​​ if user_id in member_ids: ​​​​ return {'success': True} ​​​​ else: ​​​​ return {'success': False, ​​​​ 'msg': 'Only curators are allowed to create groups'} ​​​​class YtdataservicePlugin(plugins.SingletonPlugin): ​​​​ plugins.implements(plugins.IAuthFunctions) ​​​​ def get_auth_functions(self): ​​​​ return {'group_create': group_create}

    ./ckanext-ytdataservice/ckanext/ytdataservice/plugin_v5_custom_config_setting.py

    ​​​​# encoding: utf-8 ​​​​from ckan.common import config ​​​​import ckan.plugins as plugins ​​​​import ckan.plugins.toolkit as toolkit ​​​​def group_create(context, data_dict=None): ​​​​ # Get the value of the ckan.iauthfunctions.users_can_create_groups ​​​​ # setting from the CKAN config file as a string, or False if the setting ​​​​ # isn't in the config file. ​​​​ users_can_create_groups = config.get( ​​​​ 'ckan.ytdataservice.users_can_create_groups', False) ​​​​ # Convert the value from a string to a boolean. ​​​​ users_can_create_groups = toolkit.asbool(users_can_create_groups) ​​​​ if users_can_create_groups: ​​​​ return {'success': True} ​​​​ else: ​​​​ return {'success': False, ​​​​ 'msg': 'Only sysadmins can create groups'} ​​​​class YtdataservicePlugin(plugins.SingletonPlugin): ​​​​ plugins.implements(plugins.IAuthFunctions) ​​​​ def get_auth_functions(self): ​​​​ return {'group_create': group_create}

    ./ckanext-ytdataservice/ckanext/ytdataservice/plugin_v6_parent_auth_functions.py

    ​​​​# encoding: utf-8 ​​​​from ckan.common import config ​​​​import ckan.plugins as plugins ​​​​import ckan.plugins.toolkit as toolkit ​​​​def package_delete(context, data_dict=None): ​​​​ return {'success': False, ​​​​ 'msg': 'Only sysadmins can delete packages'} ​​​​class YtdataservicePlugin(plugins.SingletonPlugin): ​​​​ plugins.implements(plugins.IAuthFunctions) ​​​​ def get_auth_functions(self): ​​​​ return {'package_delete': package_delete}
  3. 設定 setup.py 以準備載入對應的plugin

    ./ckan/setup.py

    'ckan.plugins': [ 區塊內的最下方新增三個測試用的plugin

    ​​​​    'ckan.plugins': [
    ​​​​        'synchronous_search = ckan.lib.search:SynchronousSearchPlugin',
    ​​​​        'stats = ckanext.stats.plugin:StatsPlugin',
    ​​​​        'publisher_form = ckanext.publisher_form.forms:PublisherForm',
    ​​​​        ...
    ​​​​        ...
    ​​​​        ...
    ​​​​        'ytdataservice_v4 = ckanext.ytdataservice.plugin_v4:YtdataservicePlugin',
    ​​​​        'ytdataservice_v5_custom_config_setting = ckanext.ytdataservice.plugin_v5_custom_config_setting:YtdataservicePlugin',
    ​​​​        'ytdataservice_v6_parent_auth_functions = ckanext.ytdataservice.plugin_v6_parent_auth_functions:YtdataservicePlugin',
    ​​​​    ],
    
  4. ~/ckan/lib/default/src/ckan 下,執行以下指令

    ​​​​python setup.py develop
    
  5. ~/ckan/lib/default/src/ckanext-ytdataservice 下,執行以下指令

    ​​​​nosetests --ckan --with-pylons=test.ini ckanext/ytdataservice/tests
    

    如果順利測試完成,結果應該如下

至於要怎麼寫自己要用的測試案例,之後有機會再研究
以上是如何撰寫客製化套件之自動化測試,提供參考


參考用程式碼,裡面有包含之前製作tab的 routing rule,只要看 4-38, 40, 43, 63, 64行即可

./src/ckanext-ytdataservice/ckanext/ytdataservice/plugin.py

import ckan.plugins as plugins import ckan.plugins.toolkit as toolkit def group_create(context, data_dict=None): # Get the user name of the logged-in user. user_name = context['user'] # Get a list of the members of the 'curators' group. try: members = toolkit.get_action('member_list')( data_dict={'id': 'curators', 'object_type': 'user'}) except toolkit.ObjectNotFound: # The curators group doesn't exist. return {'success': False, 'msg': "The curators groups doesn't exist, so only sysadmins " "are authorized to create groups."} # 'members' is a list of (user_id, object_type, capacity) tuples, we're # only interested in the user_ids. member_ids = [member_tuple[0] for member_tuple in members] # We have the logged-in user's user name, get their user id. convert_user_name_or_id_to_id = toolkit.get_converter( 'convert_user_name_or_id_to_id') try: user_id = convert_user_name_or_id_to_id(user_name, context) except toolkit.Invalid: # The user doesn't exist (e.g. they're not logged-in). return {'success': False, 'msg': 'You must be logged-in as a member of the curators ' 'group to create new groups.'} # Finally, we can test whether the user is a member of the curators group. if user_id in member_ids: return {'success': True} else: return {'success': False, 'msg': 'Only curators are allowed to create groups'} class YtdataservicePlugin(plugins.SingletonPlugin): plugins.implements(plugins.IConfigurer) plugins.implements(plugins.IRoutes) plugins.implements(plugins.IAuthFunctions) # IConfigurer def update_config(self, config_): toolkit.add_template_directory(config_, 'templates') toolkit.add_public_directory(config_, 'public') toolkit.add_resource('fanstatic', 'ytdataservice') # IRoutes def before_map(self, map): map.connect('cookData','/cookingData', controller='ckanext.ytdataservice.controllers:cookDataController', action='firstPage') map.connect('browseData','/cookingData/browseData', controller='ckanext.ytdataservice.controllers:cookDataController', action='browseData') map.connect('addData','/cookingData/addData', controller='ckanext.ytdataservice.controllers:cookDataController', action='addData') map.connect('cleanData','/cookingData/cleanData', controller='ckanext.ytdataservice.controllers:cookDataController', action='cleanData') map.connect('mergeData','/cookingData/mergeData', controller='ckanext.ytdataservice.controllers:cookDataController', action='mergeData') return map def after_map(self, map): return map def get_auth_functions(self): return {'group_create': group_create}