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.
- First, let's rewrite the advice so that it is using a caching service, instead of handling the caching itself.
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() - Next, let's move the caching logic into a separate class.
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] - This requires an update to our IoC blue prints, so that we inject an instance of
CachingServiceinto theCachingInterceptor.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
CachingServiceinto a separate component, and promoted it along withCachingInterceptorto 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.
- Before we write any tests, let's create a testable version of the IoC container that isolates us from any layers above and below
WikiServiceand theProxyFactoryObjectthat contains our aspect.class WikiTestAppConfig(WikiProductionAppConfig): def __init__(self): super(WikiTestAppConfig, self).__init__() @Object def data_access(self): return StubDataAccess()
- This replaces
MysqlDataAccesswithStubDataAccesswhich 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.
- Let's write a test that verifies the values of the caching service.
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.