Documentation

Testing Your Code Programmatically

Abstract

SummaryHow to write, run, and test your Python code outside of the Code Engine UI. Includes sections on installing alooma.py (Python API client), authenticating via the API, obtaining and testing via sample data, running against live data, and integration and unit testing.

In addition to testing your Code Engine code in the UI, you can also test your code programmatically. Alooma supports all the testing experiences you would expect:

Example code

For this example, let's assume we have some mobile applications sending data to Alooma, and we have an event type which contains a field called price. Most of the time, price contains numeric values, but in some versions of the mobile application price values are sent as strings. Moreover, in some rare cases, these strings are prefixed by a dollar sign. So, possible values of price include 0, 1, 1.1, "1.1", and "$1.1". If you map the price field in the Mapper to a column of type float, the mapper will handle the first four values, but it will fail to handle the last one, "$1.1".

You come up with the following function to fix this issue:

def fix_numeric(num):
  if str(num).startswith('$'):
    return float(num[1:])
  return float(num)

and you call it by using:

def transform(event):
  event['price'] = fix_numeric(event['price'])
  return event

Save your code (the functions transform and fix_numeric shown above), as a file called my_transform.py. We now describe how to test the code above.

Unit tests

Unit testing your Code Engine code means testing it locally on your own computer before deploying it to Alooma. This should not be any different from how you would test any other code, so feel free to use your favorite testing tool. For this example we will use basic unittest functionality only.

Create another file called my_tests.py with the following content:

import my_transform
import unittest

class TestCodeEngine(unittest.TestCase):

    def test_fix_numeric(self):
        self.assertEqual(0, my_transform.fix_numeric(0))
        self.assertEqual(1, my_transform.fix_numeric(1))
        self.assertEqual(1.1, my_transform.fix_numeric(1.1))
        self.assertEqual(1.1, my_transform.fix_numeric('1.1'))
        self.assertEqual(1.1, my_transform.fix_numeric('$1.1'))

You can run the tests using:

python -m unittest my_tests

Which should print something like:

.
--------------------
Ran 1 test in 0.000s

OK
Getting some real data for your tests

You can get sample events from your Alooma instance to be used in your unit tests. To do this, you need to install Alooma's Python API, which can be done by:

pip install alooma

Then, launch a Python interpreter, and run the following:

import alooma
alooma_api = alooma.Client(
    '<your-username>',
    '<your-password>',
    '<your-account>')
samples = alooma_api.get_samples(
    event_type='<event-type-you-wish-to-test>')

Alooma.get_samples, when used without any parameters, as we have done here, simply returns 10 randomly selected events from your stream. Each event is represented by a dictionary containing the following keys: sample - a Python object containing the event itself; timestamp - the time in which this event was sampled, in milliseconds since epoch; status - the status of this event; eventType - the event type of this event.

Alooma.get_samples can also receive 2 optional parameters: event_type - a string containing a name of an event type you wish to samples. If this parameter is provided, only events of this event type will be returned in the sample.error_codes - a list of strings representing status codes you wish to sample. The full list of possible status codes can be retrieved by Alooma.get_sample_status_codes().

Now samples contains a list of events that actually flowed through Alooma, which you can use in your tests. For example:

import json
json.dumps(samples[0]['sample'])

The above code would print the event as follows:

'{"price": "$20", 
  "user_id": "l5c0Fs3RbkYjk7gvnK4t", 
  "_metadata": {
    "@uuid": "caae9d4f-6bae-4188-b82e-17b45d7449c3", 
    "event_type": "test", 
    "@timestamp": "2016-06-09T13:28:44.451Z"
  }
}'

You can use this serialized event in your unit test for the transform function, by modifying my_tests.py like this:

import my_transform
import unittest
import json

class TestCodeEngine(unittest.TestCase):

    event_a = '{"price": "$20", 
                "user_id": "l5c0Fs3RbkYjk7gvnK4t", 
                "_metadata": {
                  "@uuid": "caae9d4f-6bae-4188-b82e-17b45d7449c3", 
                  "event_type": "test", 
                  "@timestamp": "2016-06-09T13:28:44.451Z"
                  }
                }'

    def test_transform(self):
        event = json.loads(self.event_a)
        transformed_event = my_transform.transform(event)
        self.assertEqual(20, transformed_event['price'])
        # add more assertions here...

There are 2 reasons for keeping a serialized version of the event in the test code (instead of getting a sample from Alooma when running the test):

  1. It runs much faster, because you don't have to send an HTTP request to Alooma to get the event.

  2. The Alooma.get_samples function may return different events every time you call it, because it updates its sample from events streaming through your Alooma system continuously. Therefore, if you change your code, run the tests on a new sample, and get a test failure, it will be difficult to determine the cause: It might be caused by the change in your code that introduced a regression, or it could be that a new event that came in caused the failure.

Tests using live data

Even with the 2 reasons mentioned above, it is also a good idea to test your code against live samples. An example could run like this:

import alooma
import my_transform
import unittest

class TestCodeEngineLiveData(unittest.TestCase):

    def test_on_live_data(self):
        alooma_api = alooma.Client(
            '<your-username>',
            '<your-password>',
            '<your-account>')
        samples = alooma_api.get_samples(
            event_type='<event-type-you-wish-to-test>')

        for sample in samples:
            try:
                my_transform.transform(sample['sample'])
            except Exception:
                self.fail(
                    'Failed on event "%s"\n%s' % (
                        sample['sample'],
                        traceback.format_exc()))

Notice that this test does not assert anything about the output of the transform but only verifies that it runs without raising exceptions. Also note that Alooma.get_samples returns 10 sample events. To run on more events, you would have to introduce another loop that will call get_samples again.

Integration tests

Before deploying new code, it is recommended to run it in Alooma. This can be done by reading the code as text, and submitting it with a sample event to test on, to Alooma.test_transform as follows:

import alooma
import unittest

class TestCodeEngineIntegration(unittest.TestCase):

    def test_on_alooma(self):
        with open('my_transform.py') as f:
            code = f.read()

        alooma_api = alooma.Client(
            '<your-username>',
            '<your-password>',    
            '<your-account>')
        samples = alooma_api.get_samples(
            event_type='<event-type-you-wish-to-test>')

        for sample in samples:
            output = alooma_api.test_transform(
                sample['sample'], code)

            if 'errorMessage' in output:
                self.fail(
                    'Failed on event "%s"\n%s' % (
                        sample['sample'],
                        output['errorMessage']))

Alooma.test_transform returns a dictionary with the following keys:

  • output - strings printed to stdout by the code while being executed

  • result - the resulting event

  • runtime - execution time of the code, in milliseconds

  • errorMessage - an optional key that exists only if an error occurred during the execution of the code

In this test we only check the existence of errorMessage in the returned dictionary. It is also possible to make assertions regarding the structure of the returned event (which can be accessed through output['result']).

We'd love to hear your ideas for improving our testing infrastructure, and we're always here to answer your questions.

Happy testing!

Search results

    No results found