CKAN Unit Testing: How to Get Start

Writing tests for your code is as important as functionality. So it is not only about writing a piece of code that works but also answering this question: How are you sure that this works?

In my previous posts, I wrote about the Comprehensive Knowledge Archive Network (CKAN, https://ckan.org/), a web application that aims to facilitate storing and distributing datasets. Here are the links to those posts in case you like to know what is CKAN and how can you develop plugins and extensions for it:

In this writing, we will see how can we write our unit test for a CKAN plugin. I suggest having a look at part1 in case you don’t know what is a plugin in CKAN and how we can create one:

Note: This post is just to get started with the CKAN unit test. It does not cover the in-depth concepts. 

First, Let’s write something that needs Testing

In part1 I created a plugin named “my_cool_new_plugin”. Now before we write our test, let’s add a new controller function to it. Later we want to test this function. 

The function is supposed to only be accessible by the CKAN admin users. It accepts a dataset id and then returns a message to the system admin. 

import ckan.plugins.toolkit as toolkit
import ckan.model as model
import ckan.logic as logic

def only_admin_can_access_me():
        context = {'model': model,
                   'user': toolkit.g.user, 'auth_user_obj': toolkit.g.userobj}
        try:
            logic.check_access('sysadmin', context, {})
        except logic.NotAuthorized:
            toolkit.abort(403, 'Need to be system administrator to administer')
        
        dataset_id = request.form.get('dataset_id')
        return "Hello Admin with Dataset {}!".format(dataset_id)

We also need to define a new endpoint in the Blueprint to access it. If you are not familiar with Blueprints, I suggest taking a look at part1.

blueprint.add_url_rule(
            u'/cool_plugin/only_admin_can_access_me',
            u'only_admin_can_access_me',
            MyLogic.only_admin_can_access_me,
            methods=['POST']
        )

Everything is ready! Now test it!

CKAN Test Config

Before writing our test, we need to config our plugin and install pytest for CKAN. CKAN pytest is a wrapper for the Python pytest [link to pytest] library. Basically, they are the same but CKAN pytest has some extra features that suit CKAN. To install it, first, enable the CKAN virtual env:

source /usr/lib/ckan/default/bin/activate

Then move to your extension library and install dev-requirement.txt that contains the CKAN pytest: (when you create an extension in CKAN, this .txt file automatically gets created for you) 

- cd /TO/YOUR/EXTENSION/DIRECTORY
- pip install -r dev-requirement.txt

The next step is to add the plugin name to the text.ini

This file is exactly like ckan.ini that we use to config CKAN on running. We use test.ini to config the test running. The file gets created automatically when we create the extension via the CKAN CLI took. The only step to run a simple test is to add our plugin(s) to it. 

ckan.plugins = my_cool_new_plugin

This line enables our plugin when running the test.

Now we are ready to write the test script.

CKAN Plugin Test

The first step is to create a Python script for the test. We do it in the /tests/ directory in our extension. CKAN CLI already made it for us. It also has a sample test script inside. The name is test_plugin.py

Delete the content inside it to write our own script.

We start with importing the needed libraries:

import pytest
import ckan.tests.factories as factories
import ckan.lib.helpers as h
import ckan.model as model
import ckan.lib.create_test_data as ctd

Then we create our test class name TestCoolPlugin with a pytest decoration with two fixtures:

@pytest.mark.usefixtures('clean_db', 'with_plugins', 'with_request_context')
class TestCoolPlugin(object): 

• Clean_db: tells CKAN to clean the test database after running the test.

• With_plugins: Run the test with CKAN plugins enabled.

The next step is to write our test class constructor. The constructor needs the pytest decorator with the value autouse=True. This tells pytest that this function needs to get run before running the tests.

@pytest.fixture(autouse=True)
    def initial(self):
        ctd.CreateTestData.create()
        self.sysadmin_user = model.User.get("testsysadmin")
        self.post_data = {}
        self.dest_url = h.url_for('my_cool_new_plugin.only_admin_can_access_me')

• Line1: create test data in CKAN

• Line2: create a system admin account

• Line3: post data initiation 

• Line4: The endpoint that we want to test

Now we start to write the actual tests. We will write two tests. 

• Use a non-admin user: The function should return the message that tells “You need to be system administrator to administer”.
• Use an admin user: The function should return the code 200

This is the first test type:

def test_with_not_admin_user(self, app):       
        user = factories.User()
        owner_org = factories.Organization(users=[{
            'name': user['id'],
            'capacity': 'member'
        }])
        dataset = factories.Dataset(owner_org=owner_org['id'])       
        self.post_data['dataset_id'] = dataset['id']                
        response = app.post(self.dest_url, data=self.post_data)        
        assert "Need to be system administrator to administer" in response.body

• First we create a non-admin user

• Then we create an organization to assign our dataset to it.

• Then we create a dataset belonging to that organization 

• Then we set that dataset id as part of the POST data.

• Finally, we post our data and check that it includes our expected test result

The second test is when the user is admin:

def test_with_admin_user(self, app):       
        owner_org = factories.Organization(users=[{
            'name': self.sysadmin_user.id,
            'capacity': 'member'
        }])
        dataset = factories.Dataset(owner_org=owner_org['id'])         
        self.post_data['dataset_id'] = dataset['id']                             
        auth = {u"Authorization": str(self.sysadmin_user.apikey)}
        response = app.post(self.dest_url, data=self.post_data , extra_environ=auth)           
        assert response.status_code == 200     

• First, we create the organization and its dataset like the previous test and put it in the POST call payload.

• Then, we put the API Key for our system admin (that we created in the initial function) in the call context for the Authorization header value. 

• Finally, we send the post call. Here the expected response status code is 200 since the user is an admin. 

We wrote our first tests in CKAN. Hooray!

Here is the link to the plugin and the test script: https://github.com/Pooya-Oladazimi/ckanext-my-first-cool-extension/blob/main/ckanext/my_first_cool_extension/tests/test_plugin.py

The End.