If you're a professional software developer, you'll be feeling a little nervous at this point. We've written quite a bit of code, in a number of aspects, and things have 'just worked'. We all know this is rarely the case.
Aspects are just like any other piece of code, they need to be tested. To mitigate risks, and to keep to best practices, we should write automated tests for our aspects. There are different types of tests to pursue. For our caching example, it would be useful to isolate the caching functionality to make sure it meets our requirements; commonly referred to as unit testing. Also, ensuring that the right advice is being applied to the right functions is critical to confirming that AOP is working and this can be exercised using an integration test.
In this chapter we have developed an aspect that generalizes caching. We showed a later enhancement with several more aspects. For the rest of this example, we are going to revert to the earlier configuration that has only our CachingInterceptor
plugged in.
Our aspect happens to be tightly coupled with its caching service. Even though the caching solution we have coded is simple, let's assume that we are planning to replace it with something more sophisticated. To make it easier to code and test enhancements to our caching service, let's go ahead and break it out into a separate module.
class CachingInterceptor(MethodInterceptor): def __init__(self, caching_service=None): self.caching_service = caching_service def invoke(self, invocation): if invocation.method_name.startswith("get"): if invocation.args not in self.caching_service.keys(): self.caching_service.store(invocation.args, invocation.proceed()) return self.caching_service.get(invocation.args) elif invocation.method_name.startswith("store"): self.caching_service.del(invocation.args) invocation.proceed()
class CachingService(object): def __init__(self): self.cache = {} def keys(self): return self.cache.keys def store(self, key, value): self.cache[key] = value def get(self, key): return self.cache[key] def del(self, key): del self.cache[key]
CachingService
into the CachingInterceptor
.class WikiProductionAppConfig(PythonConfig): def __init__(self): super(WikiProductionAppConfig, self).__init__() @Object def data_access(self): return MysqlDataAccess() @Object def caching_service(self): return CachingService() @Object def interceptor(self): return CachingInterceptor(self.caching_service()) @Object def wiki_service(self): advisor = RegexpMethodPointcutAdvisor( advice=[self.interceptor()], patterns=[".*get.*", ".*store.*"]) return ProxyFactoryObject( target=WikiService(self.data_access()), interceptors=advisor)
The following diagram shows how we have pulled CachingService
into a separate component, and promoted it along with CachingInterceptor
to fully named Spring Python objects.
Now that we have pulled our caching service into a separate module, it is easy to write some automated tests.
WikiService
and the ProxyFactoryObject
that contains our aspect.class WikiTestAppConfig(WikiProductionAppConfig): def __init__(self): super(WikiTestAppConfig, self).__init__() @Object def data_access(self): return StubDataAccess()
MysqlDataAccess
with StubDataAccess
which runs quicker, avoids database contention with other developers, and has pre-formatted responses for each method. With Python's unit test framework taking the place of the view layer as the caller, we have isolated our code base for testing.class CachedWikiTest(unittest.TestCase): def testCachingService(self): context = ApplicationContext(WikiTestAppConfig()) caching_service = context.get_object("caching_service") self.assertEquals(len(caching_service.keys()), 0) caching_service.store("key", "value") self.assertEquals(len(caching_service.keys()), 1) self.assertEquals(caching_service.get("key"), "value") caching_service.del("key") self.assertEquals(len(caching_service.keys()), 0)
In this test method, we fetch a copy of caching_service
from our IoC container. Then, we verify it's empty. Next, we store a simple key/value pair, and verify the size and content of the cache. Finally, we exercise caching_service's del()
method, and verify that the cache has been properly emptied.
I admit that CachingService
is a bit over engineered, considering it's just a Python dictionary. I normally wouldn't write unit tests for language-level structures like this. However, the purpose of this example is to show that we can move our solution into a separate module, free of any AOP machinery, and then enhance it with more sophisticated features. We could modify it to be a distributed cache that would persist across multiple nodes without impacting either WikiService
or CachingInterceptor
.
Testing the caching service is valuable because it keeps bugs from creeping back into the code base. But we also need to know that our aspect is being correctly woven with the Wiki API that we have coded.
We have confirmed that the caching service works by isolating it and writing an automated test. The final task we need to complete is verifying that we have wired the caching service into our API correctly.
Let's add another test method to CachedWikiTest
showing that WikiService
is being properly advised.
def testWikiServiceWithCaching(self): context = ApplicationContext(WikiTestAppConfig()) caching_service = context.get("caching_service") self.assertEquals(len(caching_service.keys()), 0) wiki_service = context.get_object("wiki_service") wiki_service.statistics("Spring Python") self.assertEquals(len(caching_service.keys()), 0) html = wiki_service.get_article("Spring Python") self.assertEquals(len(caching_service.keys()), 1) wiki_service.store_article("Spring Python") self.assertEquals(len(caching_service.keys()), 0)
In this test, we fetch a copy of caching_service
from our IoC container and verify that it's empty. Next, we fetch a copy of wiki_service
from our IoC container. Inside our IoC container, we know that caching_service
is linked to wiki_service
through some AOP advice. We call statistics
, and assert that the cache is still empty, since the advice doesn't apply to that method. Next, we call get_article
, and verify that the cache has a new entry. Finally, we call store_article
, and verify that it cleared the cache.
Combining this test with the earlier one, we clearly show that our CachingInterceptor
advice is working as expected. Having used the IoC container, we have decoupled things nicely, and it is now easy to adjust one class with no impact to the other.