Caliendo is a very simple interface for mocking APIs. It allows you to skip (potentially heavy) calls to your database or remote resources by storing sets of calls and caching responses based on the sequence of execution as well as function arguments. In some cases this improves unit test performance by several orders of magnitude.
If you have any questions or comments about caliendo or if you're interested in contributing feel free to contact Andrew at firstname.lastname@example.org with 'caliendo' in the subject line.
Caliendo is set up to install with
pip. You can install it directly from
GitHub by running:
pip install git+git://github.com/buzzfeed/caliendo.git#egg=caliendo
Or from pypi using the standard (to get the latest release version).
pip install caliendo
Alternatively if you have a copy of the source on your machine; cd to the parent directory and run:
pip install ./caliendo
To run tests you can use the standard unittest module. You'll have various prompts during the process. You can just hit ctrl+d to continue. To run all tests you should use nose with --nocapture. nose capturing interferes with the interactive prompts.
Your tests will need to be written is TestCases of some sort (classes). Caliendo uses the TestCase instance to figure out what module the test came from at runtime by referring to self, which is the first argument to the test methods.
python setup.py test
nosetests --all-modules --nocapture test/
Caliendo requires file read/write permissions for caching objects. The first time you invoke tests calling caliendo:
- Caliendo writes to the specified cache files. The default location is in the caliendo build, caliendo/cache, caliendo/evs, and caliendo/seeds, and caliendo/used. You can change where caliendo creates these directories and file by setting the environment variable:
- If you would like to be prompted to overwrite or modify existing cached values you can write the environment variable CALIENDO_PROMPT.
Configuration Best Practices
There are a lot of ways to set environment variables in your application. On our team we've come up with a few 'best practices' that work really well for us.
nosetestsset the variables in the
__init__.pyfile for your integration testing suite.
import os import sys os.environ['CALIENDO_CACHE_PREFIX'] = os.path.join(os.path.dirname(os.path.abspath(__file__)), 'caliendo') os.environ['PURGE_CALIENDO'] = 'True' os.environ['USE_CALIENDO'] = 'True' # Important!
unittestwith setup.py (invoked like
python setup.py test) set the variables as above in either setup.py or the
__init__.pyfile for your integration test folder.
fabricto run tests using either of these methods include the above in your
Here are a few basic examples of use.
Caliendo offers a cache which decorates callables. If you pass the cache the handle for the callable, and the args/kwargs; it will be 'cached'. The behavior is a little complex. Explained below:
- When the method is called the first time a counter is issued that is keyed on a hash of the stack trace and a serialization of the function parameters.
- If/When a matching hash is generated (e.g. a method is called with the same parameters by the same calling method the counter is incremented.
- With each unique counter the result of the function call is pickled and stored matching a CallDescriptor. If a return value can't be pickled caliendo will attempt to munge it. If caliendo fails to munge it an error will be thrown.
- When a method is called that matches an existing counter; the stored CallDescriptor rebuilds the original call and the original return value is returned by the cache.
from caliendo.facade import cache global side_effect side_effect = 0 def foo(): global side_effect side_effect += 1 return side_effect for i in range(3): assert cache(handle=foo) == i + 1 print side_effect
When the above example is run the first time; it will print 2. For every subsequent time it is run it will print 0 unless caliendo's cache is cleared.
An interface inspired greatly by python Mock is
patch() is intended to be used as a decorator for integration/unit tests that need to be decoupled from external services.
patch is called it returns the test it decorates in the context of the specified method replaced by it's
caliendo.facade.cache decorated version.
When the decorated test is invoked it is patched at runtime. After the test returns it is automatically unpatched.
patch, by default, uses
caliendo.facade.cache. If you pass an
rvalue as the second parameter; your patched method will return that value.
# Pretend these methods are all defined in various modules in the codebase. # Let foo() be defined in api.services.foos def foo(): return 'foo' # Let bar() be defined in api.services.bars def bar(): return foo() # Let baz be defined in api.bazs def baz(): return bar() # Now for our test suite. import unittest from caliendo.patch import patch from api.bazs import baz class ApiTest(unittest.TestCase): @patch('api.services.bars.bar', rvalue='biz') def test_baz(self): assert baz() == 'biz'
In the above example
bar is nested in the service layer of the architecture. We can import it once at the head of the test suite and effectively patch it at the test's invocation.
We set the rvalue to 'biz', but if we left it alone the value 'foo' would have been cached on the initial run. Every subsequent run would not have called the
bar method, and would have simply returned the cached value from the initial invokation of the test.
There are a bunch of idiomatic methods for testing that expected values match observed values. At the root of this functionality is the
The basic behavior of these methods is all the same:
- The observed value is passed for the first time.
- Caliendo will give the user an interactive shell to check the expected value (stored in the variable
- The user can modify the expected value by modifying
evin the shell.
- When the user quits with
ctrl+dthe expected value,
ev, will be cached by
- On this run the check is trivial. If the expected value stored is valid for
cacheing (e.g. vaguely
pickleable) the test will pass.
- When the
expected_valuemethod is invoked again in the same test/call the
cached value will be used for comparison to the new observed value.
The available methods are:
This is the most complicated method. As the first argument it takes a validator used to validate the observed value against the expected value.
The validator should take the expected value as the first argument and the observed value as the second. The return value doesn't really matter. The output of
expected_value.is_true_under() is whatever your return value is.
def validator(expected_value, observed_value): # Do a bunch of assertions here return anything
Just compares the observed value to the cached value.
Tests that the expected value is greater than the observed value.
Tests that the expected value is less than the observed value.
Sorry, this one isn't so idiomatic. Tests that the observed value contains
Sorry, this one isn't so idiomatic either. Tests that the observed value does not contain
Side effects can be run by patched methods.
If you pass an Exception that inherits from
StandardError your exception will be raised.
import unittest from caliendo.patch import patch from api.bazs import baz import sys class ApiTest(unittest.TestCase): @patch('api.services.bars.bar', side_effect=Exception("Things went foobar!")) def test_baz(self): with self.assertRaisesRegexp(Exception, r"Things went foobar!"): baz()
If you pass a callable as the side effect it will be called and the result returned. The arguments and keyword arguments to the method being patched will be passed.
import unittest from caliendo.patch import patch from api.bazs import baz import sys counter = 0 def example_side_effect(*args, **kwargs): global counter counter += 1 return 'foo' class ApiTest(unittest.TestCase): @patch('api.services.bars.bar', side_effect=example_side_effect) def test_baz(self): assert baz() == 'foo' assert counter == 1
Record and Replay calls to Callback Functions
When you patch a method with a callback function there's a small problem. When the cache hit occurs; the callback function never executes.
There's a decorator to allow you to execute the callback functions almost normally. If you have a patched method which calls a few callbacks before completing execution you can add a
replay decorator to indicate calls to the callbacks should be replayed.
""" In test/api/foobar.py """ def callback_for_method(a, b, c): assert a == 1 assert b == 2 assert c == 3 path = os.path.join(os.path.dirname(os.path.abspath(__file__)), 'callback_notes') if os.path.exists(path): with open(path, 'a') as f: f.write('.') else: with open(path, 'w+') as f: f.write('.') return path def method_with_callback(callback): return callback(1, 2, 3) """ In my test module """ @replay('test.api.foobar.callback_for_method') @patch('test.api.foobar.method_with_callback') def test(i): filename = method_with_callback(callback_for_method) with open(filename, 'rb') as f: assert f.read() == ('.' * (i+1))
If test(i) is run in many sessions, were i is in index of the session this test will always pass.
method_with_callback is only called the very first time the test is run, a hook is created for the callback such that each time there is a cache-hit associated with method_with_callback, the callback is executed with the expected arguments.
There are some downsides. Arguments must be pickleable. Furthermore; runtime resources such as database connections passed via closures will be lost. One workaround is to establish those in the callback at runtime.
Ignore and subsequent_rvalue: Patching runtime resources and dynamic parameters.
There are two situations it's particularly useful to ignore certain input parameters and return values.
- The input parameters or return value changes from one call to the next (even though the call is in the same order) or
- The input parameters or return value is a runtime resource (like a database cursor) and, hence, is not available when calls to the cache are made rather than to the services.
In the event you must pass a dynamic argument to a patched method you can use the
Ignore class to specify which paramters should be ignored.
from caliendo import Ignore from random import random from datetime import datetime dynamic_arguments = Ignore(args=[0,1], kwargs=['current_time']) class ApiTest(unittest.TestCase): @patch('api.services.bars.bar', ignore=dynamic_arguments) def test_baz(self): assert baz(random(), random(), 'a', mykwarg='b', current_time=datetime.now()) == 'foo'
The above test will always pass, even though positional args 0 and 1 change, as well as the current_time keyword argument.
Since the value is
Ignore'd it won't be pickled and hashed to for the CallDescriptor key. This means you can use
Ignore to skip pickling input parameters when the additional information is not needed to make the CallDescriptor hash unique.
When you need to avoid referencing a runtime resource when the cache is called (so the resource doesn't exist) you can use
subsequent_rvalue. This is a parameter to the
from caliendo import Ignore from my_database_client import find_with_cursor from my_application import from_cursor_to_list_of_models class ApiTest(unittest.TestCase): @patch('my_database_client.find_with_cursor', subsequent_rvalue=None) @patch('my_application.from_cursor_to_list_of_models', ignore=Ignore(args=)) def test_models(self): cursor = find_with_cursor('my query goes here') models = from_cursor_to_list_of_models(cursor) # Assert stuff about the models here.
In the above example we use
Ignore along with subsequent_rvalue to allow us to call our services to return models in such a way that we avoid using runtime resources in our tests entirely (after the first run).
find_with_cursor will return a cursor the first time it's called. Each subsequent time it will return
You can purge unused cache file from the cache by using the purge functionality at
By including a call to purge at the end of a full run of the tests; any unused portion of any part of the cache will be erased.
This is a good way to commit minimal files to your code base.
from caliendo.db.flatfiles import purge # Run all your tests: unittest.main() # Then purge unused files. purge()
If you have an api you want to run under Caliendo you can invoke it like so:
some_api = SomeAPI() caliendo_api = Facade( some_api ) # Note: caliendo is invoked with the INSTANCE, not the CLASS
As of revision v0.0.19 caliendo supports chaining so you can invoke it like:
caliendo_api = Facade(some_api) baz = caliendo_api.get_foo().get_bar().get_baz()
If type(baz) is not in ( float, long, str, int, dict, list, unicode ) it will be automatically wrapped by caliendo.
Some APIs check the types or class of the variables being passed in. A caliendo facade will have a class,
In order to unwrap an object to be type-checked by the target API you have to invoke the
wrapper__unwrap() method on the Facade'd API releasing the object to the target API.
A second method allows the implementer to specify a list of object to avoid Facading entirely. (Useful for exporting models).
facaded_api = Facade(SOMEAPI()) facaded_api.wrapper__ignore( somemodule.SomeClassDefinition )
The above example will ensure objects with
somemodule.SomeClassDefinition will never be wrapped.
To stop ignoring a particular class you can do:
facaded_api = Facade(SOMEAPI()) facaded_api.wrapper__ignore( somemodule.SomeClassDefinition ) facaded_api.wrapper__unignore( somemodule.SomeClassDefinition )
Once you have an instance of an API running under a
Facade you should be able
to call all the methods normally.
Hooks and Stacks (advanced)
When the cache is called the first time you can specify a hook that gets called each subsequent time a call matching that CallDescriptor is made. After a new
CallDescriptor is added to the
CallStack for the current
patch the method you specified will be called, and passed the most recent CallDescriptor as it's argument.
The callback has to be pickleable. For example:
import unittest from caliendo.patch import patch def callback(most_recent_call_descriptor): print "The last return value was: '%s'\n" % most_recent_call_descriptor.returnval class ApiTest(unittest.TestCase): @patch('api.services.bars.bar', callback=callback) def test_baz(self): assert baz() == 'baz'
The above example will print "The last return value was: 'baz'" followed by a newline to stdout.
If you start getting unexpected API results you should clear the cache by simply deleting all the rows in the
Caliendo doesn't support a large level of nested objects in arguments or return values. If you start getting unexpected results this may be the problem. Better nesting is to come.
If you alternate between calls to the
Facadeinstance of an API and the API itself you will probably see unexpected results. Caliendo maintains the state of the API by maintaining a reference to the API internally.
If you have a class inheriting from
dictyou'll need to define a
__setstate__method. Described: [http://stackoverflow.com/questions/5247250/why-does-pickle-getstate-accept-as-a-return-value-the-very-instance-it-requi]
If you're trying to mock a module that contains class definitions; you can use the classes normally except that the type will be that of a lambda function instead of a class.