implemented chain of responsibility pattern.
This commit is contained in:
0
pypatterns/behavioral/__init__.py
Normal file
0
pypatterns/behavioral/__init__.py
Normal file
79
pypatterns/behavioral/chain.py
Normal file
79
pypatterns/behavioral/chain.py
Normal file
@@ -0,0 +1,79 @@
|
||||
from abc import ABCMeta, abstractmethod
|
||||
|
||||
|
||||
class ChainException(Exception):
|
||||
"""
|
||||
Exception for when a chain link could not handle a request.
|
||||
"""
|
||||
pass
|
||||
|
||||
|
||||
class ChainLink(object, metaclass=ABCMeta):
|
||||
"""
|
||||
Abstract ChainLink object as part of the Chain of Responsibility pattern.
|
||||
"""
|
||||
def __init__(self):
|
||||
"""
|
||||
Initialize a new ChainLink instance.
|
||||
"""
|
||||
self.successor = None
|
||||
|
||||
def set_successor(self, successor):
|
||||
"""
|
||||
Set a chain link to call if this chain link fails.
|
||||
|
||||
@param successor: The chain link to call if this chain link fails.
|
||||
@type successor: ChainLink
|
||||
"""
|
||||
self.successor = successor
|
||||
|
||||
def successor_handle(self, request):
|
||||
"""
|
||||
Have this chain links successor handle a request.
|
||||
|
||||
@param request: The request to handle.
|
||||
"""
|
||||
try:
|
||||
return self.successor.handle(request)
|
||||
except AttributeError:
|
||||
raise ChainException
|
||||
|
||||
@abstractmethod
|
||||
def handle(self, request):
|
||||
"""
|
||||
Handle a request.
|
||||
|
||||
@param request: The request to handle.
|
||||
"""
|
||||
pass
|
||||
|
||||
|
||||
class Chain(object, metaclass=ABCMeta):
|
||||
"""
|
||||
Abstract Chain class as part of the Chain of Responsibility pattern.
|
||||
"""
|
||||
def __init__(self, chainlink):
|
||||
"""
|
||||
Initialize a new Chain instance.
|
||||
|
||||
@param chainlink: The starting chain link.
|
||||
"""
|
||||
self.chainlink = chainlink
|
||||
|
||||
def handle(self, request):
|
||||
"""
|
||||
Handle a request.
|
||||
|
||||
@param request: The request to handle.
|
||||
"""
|
||||
try:
|
||||
return self.chainlink.handle(request)
|
||||
except ChainException:
|
||||
return self.fail()
|
||||
|
||||
@abstractmethod
|
||||
def fail(self):
|
||||
"""
|
||||
The method to call when the chain could not handle a request.
|
||||
"""
|
||||
pass
|
||||
4
setup.py
4
setup.py
@@ -7,6 +7,8 @@ setup(
|
||||
author='Tyler LaBerge',
|
||||
packages=[
|
||||
'pypatterns',
|
||||
'pypatterns.creational'
|
||||
'pypatterns.creational',
|
||||
'pypatterns.behavioral'
|
||||
],
|
||||
)
|
||||
|
||||
|
||||
0
tests/behavioral_tests/__init__.py
Normal file
0
tests/behavioral_tests/__init__.py
Normal file
140
tests/behavioral_tests/test_chain.py
Normal file
140
tests/behavioral_tests/test_chain.py
Normal file
@@ -0,0 +1,140 @@
|
||||
from unittest import TestCase
|
||||
from pypatterns.behavioral.chain import ChainException, ChainLink, Chain
|
||||
|
||||
|
||||
class ChainLinkTestCase(TestCase):
|
||||
"""
|
||||
Unit testing class for the ChainLink class.
|
||||
"""
|
||||
def setUp(self):
|
||||
"""
|
||||
Initialize testing data.
|
||||
"""
|
||||
class ConcreteChainLinkThree(ChainLink):
|
||||
|
||||
def handle(self, request):
|
||||
if request == 'handle_three':
|
||||
return "Handled in chain link three"
|
||||
else:
|
||||
return self.successor_handle(request)
|
||||
|
||||
class ConcreteChainLinkTwo(ChainLink):
|
||||
|
||||
def __init__(self):
|
||||
super().__init__()
|
||||
self.set_successor(ConcreteChainLinkThree())
|
||||
|
||||
def handle(self, request):
|
||||
if request == 'handle_two':
|
||||
return "Handled in chain link two"
|
||||
else:
|
||||
return self.successor_handle(request)
|
||||
|
||||
class ConcreteChainLinkOne(ChainLink):
|
||||
|
||||
def __init__(self):
|
||||
super().__init__()
|
||||
self.set_successor(ConcreteChainLinkTwo())
|
||||
|
||||
def handle(self, request):
|
||||
if request == 'handle_one':
|
||||
return "Handled in chain link one"
|
||||
else:
|
||||
return self.successor_handle(request)
|
||||
|
||||
self.chain_link_one_class = ConcreteChainLinkOne
|
||||
|
||||
def test_success_handle(self):
|
||||
"""
|
||||
Test the handle method with successful requests.
|
||||
|
||||
@raise AssertionError: If the test fails.
|
||||
"""
|
||||
handler = self.chain_link_one_class()
|
||||
|
||||
self.assertEquals("Handled in chain link one", handler.handle("handle_one"))
|
||||
self.assertEquals("Handled in chain link two", handler.handle("handle_two"))
|
||||
self.assertEquals("Handled in chain link three", handler.handle("handle_three"))
|
||||
|
||||
def test_fail_handle(self):
|
||||
"""
|
||||
Test the handle method with unsuccessful requests.
|
||||
|
||||
@raise AssertionError: If the test fails.
|
||||
"""
|
||||
handler = self.chain_link_one_class()
|
||||
with self.assertRaises(ChainException):
|
||||
handler.handle("foo")
|
||||
|
||||
|
||||
class ChainTestCase(TestCase):
|
||||
"""
|
||||
Unit testing class for the Chain class.
|
||||
"""
|
||||
def setUp(self):
|
||||
"""
|
||||
Initialize testing data.
|
||||
"""
|
||||
class ConcreteChainLinkThree(ChainLink):
|
||||
|
||||
def handle(self, request):
|
||||
if request == 'handle_three':
|
||||
return "Handled in chain link three"
|
||||
else:
|
||||
return self.successor_handle(request)
|
||||
|
||||
class ConcreteChainLinkTwo(ChainLink):
|
||||
|
||||
def __init__(self):
|
||||
super().__init__()
|
||||
self.set_successor(ConcreteChainLinkThree())
|
||||
|
||||
def handle(self, request):
|
||||
if request == 'handle_two':
|
||||
return "Handled in chain link two"
|
||||
else:
|
||||
return self.successor_handle(request)
|
||||
|
||||
class ConcreteChainLinkOne(ChainLink):
|
||||
|
||||
def __init__(self):
|
||||
super().__init__()
|
||||
self.set_successor(ConcreteChainLinkTwo())
|
||||
|
||||
def handle(self, request):
|
||||
if request == 'handle_one':
|
||||
return "Handled in chain link one"
|
||||
else:
|
||||
return self.successor_handle(request)
|
||||
|
||||
class ConcreteChain(Chain):
|
||||
|
||||
def __init__(self):
|
||||
super().__init__(ConcreteChainLinkOne())
|
||||
|
||||
def fail(self):
|
||||
return 'Fail'
|
||||
|
||||
self.chain_class = ConcreteChain
|
||||
|
||||
def test_success_handle(self):
|
||||
"""
|
||||
Test the handle method with a successful request
|
||||
|
||||
@raise AssertionError: If the test fails.
|
||||
"""
|
||||
chain = self.chain_class()
|
||||
|
||||
self.assertEquals("Handled in chain link one", chain.handle("handle_one"))
|
||||
self.assertEquals("Handled in chain link two", chain.handle("handle_two"))
|
||||
self.assertEquals("Handled in chain link three", chain.handle("handle_three"))
|
||||
|
||||
def test_fail_handle(self):
|
||||
"""
|
||||
Test the handle method with unsuccessful requests.
|
||||
|
||||
@raise AssertionError: If the test fails.
|
||||
"""
|
||||
chain = self.chain_class()
|
||||
|
||||
self.assertEquals("Fail", chain.handle("foo"))
|
||||
Reference in New Issue
Block a user