PK †ÙD·m]·†1 †1 python-aspectlib-1.1/index.html
aspectlib is an aspect-oriented programming, monkey-patch and decorators library. It is useful when changing behavior in existing code is desired.
aspectlib provides two core tools to do AOP: Aspects and a weaver.
An aspect can be created by decorating a generator with an Aspect. The generator yields advices - simple behavior changing instructions.
An Aspect instance is a simple function decorator. Decorating a function with an aspect will change the function’s behavior according to the advices yielded by the generator.
Example:
@aspectlib.Aspect
def strip_return_value():
result = yield aspectlib.Proceed
yield aspectlib.Return(result.strip())
@strip_return_value
def read(name):
return open(name).read()
You can use these advices:
Patches classes and functions with the given aspect. When used with a class it will patch all the methods. In AOP parlance these patched functions and methods are referred to as cut-points.
Returns a Rollback object that can be used a context manager. It will undo all the changes at the end of the context.
Example:
@aspectlib.Aspect
def mock_open():
yield aspectlib.Return(StringIO("mystuff"))
with aspectlib.weave(open, mock_open):
assert open("/doesnt/exist.txt").read() == "mystuff"
You can use aspectlib.weave() on: classes, instances, builtin functions, module level functions, methods, classmethods, staticmethods, instance methods etc.
pip install aspectlib
Or, if you live in the stone age:
easy_install aspectlib
For your convenience there is a python-aspectlib meta-package that will just install aspectlib, in case you run pip install python-aspectlib by mistake.
OS: | Any |
---|---|
Runtime: | Python 2.6, 2.7, 3.3, 3.4 or PyPy |
Python 3.2, 3.1 and 3.0 are NOT supported (some objects are too crippled).
Lightweight spies and mock responses
Example usage, suppose you want to test this class:
>>> class ProductionClass(object):
... def method(self):
... return 'stuff'
>>> real = ProductionClass()
With aspectlib.test.mock and aspectlib.test.record:
>>> from aspectlib import weave, test
>>> patch = weave(real.method, [test.mock(3), test.record])
>>> real.method(3, 4, 5, key='value')
3
>>> assert real.method.calls == [(real, (3, 4, 5), {'key': 'value'})]
As a bonus, you have an easy way to rollback all the mess:
>>> patch.rollback()
>>> real.method()
'stuff'
With mock:
>>> from mock import Mock
>>> real = ProductionClass()
>>> real.method = Mock(return_value=3)
>>> real.method(3, 4, 5, key='value')
3
>>> real.method.assert_called_with(3, 4, 5, key='value')
Elaborate tools for testing difficult code
Writing tests using the Story is viable when neither integration tests or unit tests seem adequate:
- Integration tests are too difficult to integrate in your test harness due to automation issues, permissions or plain lack of performance.
- Unit tests are too difficult to write due to design issues (like too many dependencies, dependency injects is too hard etc) or take too much time to write to get good code coverage.
The Story is the middle-ground, bringing those two types of testing closer. It allows you to start with integration tests and later mock/stub with great ease all the dependencies.
Warning
The Story is not intended to patch and mock complex libraries that keep state around. E.g.: requests keeps a connection pool around - there are better choices.
Note
Using the Story on imperative, stateless interfaces is best.
Suppose we implement this simple GNU tree clone:
>>> import os
>>> def tree(root, prefix=''):
... if not prefix:
... print("%s%s" % (prefix, os.path.basename(root)))
... for pos, name in reversed(list(enumerate(sorted(os.listdir(root), reverse=True)))):
... print("%s%s%s" % (prefix, "├── " if pos else "└── ", name))
... absname = os.path.join(root, name)
... if os.path.isdir(absname):
... tree(absname, prefix + ("│ " if pos else " "))
Lets suppose we would make up some directories and files for our tests:
>>> if not os.path.exists('some/test/dir'): os.makedirs('some/test/dir')
>>> if not os.path.exists('some/test/empty'): os.makedirs('some/test/empty')
>>> with open('some/test/dir/file.txt', 'w') as fh:
... pass
And we’ll assert that tree has this output:
>>> tree('some')
some
└── test
├── dir
│ └── file.txt
└── empty
But now we’re left with some garbage and have to clean it up:
>>> import shutil
>>> shutil.rmtree('some')
This is not very practical - we’ll need to create many scenarios, and some are not easy to create automatically (e.g: tests for permissions issues - not easy to change permissions from within a test).
Normally, to handle this we’d have have to manually monkey-patch the os module with various mocks or add dependency-injection in the tree function and inject mocks. Either approach we’ll leave us with very ugly code.
With dependency-injection tree would look like this:
def tree(root, prefix='', basename=os.path.basename, listdir=os.listdir, join=os.path.join, isdir=os.path.isdir):
...
One could argue that this is overly explicit, and the function’s design is damaged by testing concerns. What if we need to check for permissions ? We’d have to extend the signature. And what if we forget to do that ? In some situations one cannot afford all this (re-)engineering (e.g: legacy code, simplicity goals etc).
The aspectlib.test.Story is designed to solve this problem in a neat way.
We can start with some existing test data in the filesystem:
>>> os.makedirs('some/test/dir')
>>> os.makedirs('some/test/empty')
>>> with open('some/test/dir/file.txt', 'w') as fh:
... pass
Write an empty story and examine the output:
>>> from aspectlib.test import Story
>>> with Story(os, methods="^(?!error)[a-z]+$") as story:
... pass
>>> with story.replay(strict=False) as replay:
... tree('some')
some
└── test
├── dir
│ └── file.txt
└── empty
STORY/REPLAY DIFF:
--- expected...
+++ actual...
@@ ... @@
+os.listdir('some') == ['test'] # returns
+os.stat('some/test') == os.stat_result((...)) # returns
+os.listdir('some/test') == [...'dir'...] # returns
+os.stat('some/test/dir') == os.stat_result((...)) # returns
+os.listdir('some/test/dir') == ['file.txt'] # returns
+os.stat('some/test/dir/file.txt') == os.stat_result((...)) # returns
+os.stat('some/test/empty') == os.stat_result((...)) # returns
+os.listdir('some/test/empty') == [] # returns
ACTUAL:
os.listdir('some') == ['test'] # returns
os.stat('some/test') == os.stat_result((...)) # returns
os.listdir('some/test') == [...'dir'...] # returns
os.stat('some/test/dir') == os.stat_result((...)) # returns
os.listdir('some/test/dir') == ['file.txt'] # returns
os.stat('some/test/dir/file.txt') == os.stat_result((...)) # returns
os.stat('some/test/empty') == os.stat_result((...)) # returns
os.listdir('some/test/empty') == [] # returns
We can quickly get whatever we would need to put in the story with aspectlib.test.Replay.unexpected:
>>> print(replay.unexpected) os.listdir('some') == ['test'] # returns os.stat('some/test') == os.stat_result((...)) # returns os.listdir('some/test') == [...'dir'...] # returns os.stat('some/test/dir') == os.stat_result((...)) # returns os.listdir('some/test/dir') == ['file.txt'] # returns os.stat('some/test/dir/file.txt') == os.stat_result((...)) # returns os.stat('some/test/empty') == os.stat_result((...)) # returns os.listdir('some/test/empty') == [] # returns
Now we can remove the test directories and fill the story:
>>> import shutil
>>> shutil.rmtree('some')
The story:
>>> with Story(os, methods="^(?!error)[a-z]+$") as story: ... os.listdir('some') == ['test'] ... os.stat('some/test') == os.stat_result((16893, 6691875, 2049, 3, 1000, 1000, 4096, 1399131539, 1399131539, 1399131539)) ... os.listdir('some/test') == ['empty', 'dir'] # returns ... os.stat('some/test/dir') == os.stat_result((16893, 6691876, 2049, 2, 1000, 1000, 4096, 1399131539, 1399131539, 1399131539)) ... os.listdir('some/test/dir') == ['file.txt'] ... os.stat('some/test/dir/file.txt') == os.stat_result((33204, 6691877, 2049, 1, 1000, 1000, 9, 1399131539, 1399131539, 1399131539)) ... os.stat('some/test/empty') == os.stat_result((16893, 6691877, 2049, 2, 1000, 1000, 4096, 1399132977, 1399132977, 1399132977)) # returns ... os.listdir('some/test/empty') == [] # returnsAnd the strict replay:
>>> with story.replay(proxy=False) as replay: ... tree('some') some └── test ├── dir │ └── file.txt └── empty
If we diverge a bit from the story (or we’d have some unexpected change in the tree function) we’d get something like this:
>>> with Story(os, methods="^(?!error)[a-z]+$") as story:
... os.listdir('some') == ['test']
... os.listdir('bogus') == ['some bogus directory']
... os.stat('some/test') == os.stat_result((16893, 6691875, 2049, 3, 1000, 1000, 4096, 1399131539, 1399131539, 1399131539))
... os.listdir('some/test') == ['empty', 'dir'] # returns
... os.stat('some/test/dir') == os.stat_result((16893, 6691876, 2049, 2, 1000, 1000, 4096, 1399131539, 1399131539, 1399131539))
... os.listdir('some/test/dir') == ['file.txt']
... os.stat('some/test/dir/file.txt') == os.stat_result((33204, 6691877, 2049, 1, 1000, 1000, 9, 1399131539, 1399131539, 1399131539))
... os.stat('some/test/empty') == os.stat_result((16893, 6691877, 2049, 2, 1000, 1000, 4096, 1399132977, 1399132977, 1399132977)) # returns
... os.listdir('some/test/empty') == [] # returns
>>> with story.replay(proxy=False) as replay:
... tree('some')
Traceback (most recent call last):
...
AssertionError: --- expected...
+++ actual...
@@ ... @@
os.listdir('some') == ['test'] # returns
-os.listdir('bogus') == ['some bogus directory'] # returns
os.stat('some/test') == os.stat_result((16893, 6691875, 2049, 3, 1000, 1000, 4096, 1399131539, 1399131539, 1399131539)) # returns
os.listdir('some/test') == [...'dir'...] # returns
os.stat('some/test/dir') == os.stat_result((16893, 6691876, 2049, 2, 1000, 1000, 4096, 1399131539, 1399131539, 1399131539)) # returns
There are perfectly sane use cases for monkey-patching (aka weaving):
Then in those situations:
Because it does more things that just patching. Depending on the target object it will patch and/or create one or more subclasses and objects.
Some frameworks don’t resort to monkey patching but instead force the user to use ridiculous amounts of abstractions and wrapping in order to make weaving possible. Notable example: spring-python.
For all intents and purposes I think it’s wrong to have such high amount of boilerplate in Python.
Also, aspectlib is targeting a different stage of development: the maintenance stage - where the code is already written and needs additional behavior, in a hurry :)
Where code is written from scratch and AOP is desired there are better choices than both aspectlib and spring-python.
aspectlib was initially written because I was tired of littering other people’s code with prints and logging statements just to fix one bug or understand how something works. aspectlib.debug.log is aspectlib‘s crown jewel. Of course, aspectlib has other applications, see the Rationale.
TODO: Make a more configurable retry decorator and add it in aspectlib.contrib.
class Client(object):
def __init__(self, address):
self.address = address
self.connect()
def connect(self):
# establish connection
def action(self, data):
# do some stuff
def retry(retries=(1, 5, 15, 30, 60), retry_on=(IOError, OSError), prepare=None):
assert len(retries)
@aspectlib.Aspect
def retry_aspect(*args, **kwargs):
durations = retries
while True:
try:
yield aspectlib.Proceed
break
except retry_on as exc:
if durations:
logging.warn(exc)
time.sleep(durations[0])
durations = durations[1:]
if prepare:
prepare(*args, **kwargs)
else:
raise
return retry_aspect
Now patch the Client class to have the retry functionality on all its methods:
aspectlib.weave(Client, retry())
or with different retry options (reconnect before retry):
aspectlib.weave(Client, retry(prepare=lambda self, *_: self.connect())
or just for one method:
aspectlib.weave(Client.action, retry())
You can see here the advantage of having reusable retry functionality. Also, the retry handling is decoupled from the Client class.
... those damn sockets:
>>> import aspectlib, socket, sys
>>> with aspectlib.weave(
... socket.socket,
... aspectlib.debug.log(
... print_to=sys.stdout,
... stacktrace=None,
... ),
... lazy=True,
... ):
... s = socket.socket()
... s.connect(('example.com', 80))
... s.send(b'GET / HTTP/1.1\r\nHost: example.com\r\n\r\n')
... s.recv(8)
... s.close()
...
{socket...}.connect(('example.com', 80))
{socket...}.connect => None
{socket...}.send(...'GET / HTTP/1.1\r\nHost: example.com\r\n\r\n')
{socket...}.send => 37
37
{socket...}.recv(8)
{socket...}.recv => ...HTTP/1.1...
...'HTTP/1.1'
...
The output looks a bit funky because it is written to be run by doctest - so you don’t use broken examples :)
Mock behavior for tests:
class MyTestCase(unittest.TestCase):
def test_stuff(self):
@aspectlib.Aspect
def mock_stuff(self, value):
if value == 'special':
yield aspectlib.Return('mocked-result')
else:
yield aspectlib.Proceed
with aspectlib.weave(foo.Bar.stuff, mock_stuff):
obj = foo.Bar()
self.assertEqual(obj.stuff('special'), 'mocked-result')
Safe toolkit for writing decorators (hereby called aspects)
aspectlib.Aspect | Container for the advice yielding generator. |
aspectlib.Proceed | Instruction for calling the decorated function. |
aspectlib.Return | Instruction for returning a optional value. |
Power tools for patching functions (hereby glorified as weaving)
aspectlib.ALL_METHODS | Compiled regular expression objects |
aspectlib.NORMAL_METHODS | Compiled regular expression objects |
aspectlib.weave | Send a message to a recipient |
aspectlib.Rollback | When called, rollbacks all the patches and changes the weave() has done. |
Instruction for calling the decorated function. Can be used multiple times.
If not used as an instance then the default args and kwargs are used.
Instruction for returning a optional value.
If not used as an instance then None is returned.
Container for the advice yielding generator. Can be used as a decorator on other function to change behavior according to the advices yielded from the generator.
Parameters: |
|
---|
Usage:
>>> @Aspect
... def my_decorator(*args, **kwargs):
... print("Got called with args: %s kwargs: %s" % (args, kwargs))
... result = yield
... print(" ... and the result is: %s" % (result,))
>>> @my_decorator
... def foo(a, b, c=1):
... print((a, b, c))
>>> foo(1, 2, c=3)
Got called with args: (1, 2) kwargs: {'c': 3}
(1, 2, 3)
... and the result is: None
Normally you don’t have access to the cutpoints (the functions you’re going to use the aspect/decorator on) because you don’t and should not call them directly. There are situations where you’d want to get the name or other data from the function. This is where you use the bind=True option:
>>> @Aspect(bind=True)
... def my_decorator(cutpoint, *args, **kwargs):
... print("`%s` got called with args: %s kwargs: %s" % (cutpoint.__name__, args, kwargs))
... result = yield
... print(" ... and the result is: %s" % (result,))
>>> @my_decorator
... def foo(a, b, c=1):
... print((a, b, c))
>>> foo(1, 2, c=3)
`foo` got called with args: (1, 2) kwargs: {'c': 3}
(1, 2, 3)
... and the result is: None
You can use these advices:
Note
The Aspect will correctly handle generators and coroutines (consume them, capture result).
Example:
>>> from aspectlib import Aspect
>>> @Aspect
... def log_errors(*args, **kwargs):
... try:
... yield
... except Exception as exc:
... print("Raised %r for %s/%s" % (exc, args, kwargs))
... raise
Will work as expected with generators (and coroutines):
>>> @log_errors
... def broken_generator():
... yield 1
... raise RuntimeError()
>>> from pytest import raises
>>> raises(RuntimeError, lambda: list(broken_generator()))
Raised RuntimeError() for ()/{}
...
>>> @log_errors
... def broken_function():
... raise RuntimeError()
>>> raises(RuntimeError, broken_function)
Raised RuntimeError() for ()/{}
...
And it will handle results:
>>> from aspectlib import Aspect
>>> @Aspect
... def log_results(*args, **kwargs):
... try:
... value = yield
... except Exception as exc:
... print("Raised %r for %s/%s" % (exc, args, kwargs))
... raise
... else:
... print("Returned %r for %s/%s" % (value, args, kwargs))
>>> @log_results
... def weird_function():
... yield 1
... raise StopIteration('foobar') # in Python 3 it's the same as: return 'foobar'
>>> list(weird_function())
Returned 'foobar' for ()/{}
[1]
When called, rollbacks all the patches and changes the weave() has done.
Alias of __exit__.
Alias of __exit__.
Compiled regular expression objects
Compiled regular expression objects
Send a message to a recipient
Parameters: |
|
---|---|
Returns: | aspectlib.Rollback instance |
Raises TypeError: | |
If target is a unacceptable object, or the specified options are not available for that type of object. |
Changed in version 0.4.0: Replaced only_methods, skip_methods, skip_magicmethods options with methods. Renamed on_init option to lazy. Added aliases option. Replaced skip_subclasses option with subclasses.
aspectlib.debug.log | Decorates func to have logging. |
aspectlib.debug.format_stack | Returns a one-line string with the current callstack. |
aspectlib.debug.frame_iterator | Yields frames till there are no more. |
aspectlib.debug.strip_non_ascii | Convert to string (using str) and replace non-ascii characters with a dot (.). |
Returns a one-line string with the current callstack.
Decorates func to have logging.
Parameters: |
|
---|---|
Returns: | A decorator or a wrapper. |
Example
>>> @log(print_to=sys.stdout)
... def a(weird=False):
... if weird:
... raise RuntimeError('BOOM!')
>>> a()
a() <<< ...
a => None
>>> try:
... a(weird=True)
... except Exception:
... pass # naughty code !
a(weird=True) <<< ...
a ~ raised RuntimeError('BOOM!',)
You can conveniently use this to logs just errors, or just results, example:
>>> import aspectlib
>>> with aspectlib.weave(float, log(call=False, result=False, print_to=sys.stdout)):
... try:
... float('invalid')
... except Exception as e:
... pass # naughty code !
float('invalid') <<< ...
float ~ raised ValueError(...float...invalid...)
This makes debugging naughty code easier.
PS: Without the weaving it looks like this:
>>> try:
... log(call=False, result=False, print_to=sys.stdout)(float)('invalid')
... except Exception:
... pass # naughty code !
float('invalid') <<< ...
float ~ raised ValueError(...float...invalid...)
Changed in version 0.5.0: Renamed arguments to call_args. Renamed arguments_repr to call_args_repr. Added call option.
This module aims to be a lightweight and flexible alternative to the popular mock framework and more.
aspectlib.test.record | Factory or decorator (depending if func is initially given). |
aspectlib.test.mock | Factory for a decorator that makes the function return a given return_value. |
aspectlib.test.Story | This a simple yet flexible tool that can do “capture-replay mocking” or “test doubles” [1]. |
aspectlib.test.Replay | Object implementing the replay transaction. |
Factory or decorator (depending if func is initially given).
Parameters: |
|
---|---|
Returns: | A wrapper that has a calls property. |
The decorator returns a wrapper that records all calls made to func. The history is available as a call property. If access to the function is too hard then you need to specify the history manually.
Example
>>> @record
... def a(x, y, a, b):
... pass
>>> a(1, 2, 3, b='c')
>>> a.calls
[Call(self=None, args=(1, 2, 3), kwargs={'b': 'c'})]
Or, with your own history list:
>>> calls = []
>>> @record(calls=calls)
... def a(x, y, a, b):
... pass
>>> a(1, 2, 3, b='c')
>>> a.calls
[Call(self=None, args=(1, 2, 3), kwargs={'b': 'c'})]
>>> calls is a.calls
True
Changed in version 0.9.0: Renamed history option to calls. Renamed call option to iscalled. Added callback option. Added extended option.
Factory for a decorator that makes the function return a given return_value.
Parameters: |
|
---|---|
Returns: | A decorator. |
This a simple yet flexible tool that can do “capture-replay mocking” or “test doubles” [1]. It leverages aspectlib‘s powerful weaver.
Parameters: |
|
---|
The Story allows some testing patterns that are hard to do with other tools:
The Story works in two of transactions:
The story: You describe what calls you want to mocked. Initially you don’t need to write this. Example:
>>> import mymod
>>> with Story(mymod) as story:
... mymod.func('some arg') == 'some result'
... mymod.func('bad arg') ** ValueError("can't use this")
The replay: You run the code uses the interfaces mocked in the story. The replay always starts from a story instance.
Changed in version 0.9.0: Added in.
[1] | (1, 2) http://www.martinfowler.com/bliki/TestDouble.html |
Parameters: |
|
---|---|
Returns: | A aspectlib.test.Replay object. |
Example
>>> import mymod
>>> with Story(mymod) as story:
... mymod.func('some arg') == 'some result'
... mymod.func('other arg') == 'other result'
>>> with story.replay(strict=False):
... print(mymod.func('some arg'))
... mymod.func('bogus arg')
some result
Got bogus arg in the real code !
STORY/REPLAY DIFF:
--- expected...
+++ actual...
@@ -1,2 +1,2 @@
mymod.func('some arg') == 'some result' # returns
-mymod.func('other arg') == 'other result' # returns
+mymod.func('bogus arg') == None # returns
ACTUAL:
mymod.func('some arg') == 'some result' # returns
mymod.func('bogus arg') == None # returns
Object implementing the replay transaction.
This object should be created by Story‘s replay method.
Returns a pretty text representation of the unexpected and missing calls.
Most of the time you don’t need to directly use this. This is useful when you run the replay in strict=False mode and want to do custom assertions.
Returns a pretty text representation of just the unexpected calls.
The output should be usable directly in the story (just copy-paste it). Example:
>>> import mymod
>>> with Story(mymod) as story:
... pass
>>> with story.replay(strict=False, dump=False) as replay:
... mymod.func('some arg')
... try:
... mymod.badfunc()
... except ValueError as exc:
... print(exc)
Got some arg in the real code !
boom!
>>> print(replay.unexpected)
mymod.func('some arg') == None # returns
mymod.badfunc() ** ValueError('boom!',) # raises
We can just take the output and paste in the story:
>>> import mymod
>>> with Story(mymod) as story:
... mymod.func('some arg') == None # returns
... mymod.badfunc() ** ValueError('boom!') # raises
>>> with story.replay():
... mymod.func('some arg')
... try:
... mymod.badfunc()
... except ValueError as exc:
... print(exc)
boom!
class BaseProcessor(object):
def process_foo(self, data):
# do some work
def process_bar(self, data):
# do some work
class ValidationConcern(aspectlib.Concern):
@aspectlib.Aspect
def process_foo(self, data):
# validate data
if is_valid_foo(data):
yield aspectlib.Proceed
else:
raise ValidationError()
@aspectlib.Aspect
def process_bar(self, data):
# validate data
if is_valid_bar(data):
yield aspectlib.Proceed
else:
raise ValidationError()
aspectlib.weave(BaseProcesor, ValidationConcern)
class MyProcessor(BaseProcessor):
def process_foo(self, data):
# do some work
def process_bar(self, data):
# do some work
# MyProcessor automatically inherits BaseProcesor's ValidationConcern
Changed aspectlib.test.record:
- Renamed history option to calls.
- Renamed call option to iscalled.
- Added callback option.
- Added extended option.
Changed aspectlib.weave:
- Allow weaving everything in a module.
- Allow weaving instances of new-style classes.
Added aspectlib.test.Story class for capture-replay and stub/mock testing.
Changed aspectlib.debug.log:
- Renamed arguments to call_args.
- Renamed arguments_repr to call_args_repr.
- Added call option.
- Fixed issue with logging from old-style methods (object name was a generic “instance”).
Fixed issues with weaving some types of builtin methods.
Allow to apply multiple aspects at the same time.
Validate string targets before weaving. aspectlib.weave('mod.invalid name', aspect) now gives a clear error (invalid name is not a valid identifier)
Various documentation improvements and examples.
Changed aspectlib.weave:
- Replaced only_methods, skip_methods, skip_magicmethods options with methods.
- Renamed on_init option to lazy.
- Added aliases option.
- Replaced skip_subclasses option with subclasses.
Fixed weaving methods from a string target.