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都沒開啟，所以還不用擔心，之後注意就好
    • 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



1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64



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}
以上參考於自己的筆記

留言

張貼留言