Object Oriented Design - Call Center

In [50]:

from collections import deque

class CallCenter(object):

    def __init__(self, address, number, agents=None, managers=None):
        self.agents = agents
        self.managers = managers
        self.address = address
        self.number = number
        self.wait_queue = deque([])
        
    def handle_call(self, call):
        assigned = False
        if call.level == CallLevel.AGENT:
            assigned = self._handle_call_by_level(call, self.agents)
        if not assigned or call.level == CallLevel.MANAGER:
            assigned = self._handle_call_by_level(call, self.managers)
        if not assigned:
            self.wait_queue.append(call)
    
    def _handle_call_by_level(self, call, agents):
        for agent in agents:
            if agent.state == AgentState.READY:
                call.agent = agent
                agent.pickup_call(call)
                return True
        return False
    
    def notify_agent_available(self):
        if len(self.wait_queue) > 0:
            escalated_call = self.wait_queue.popleft()
            self.handle_call(escalated_call)

class Call(object):
    def __init__(self, customer):
        self.customer = customer
        self.state = CallState.INCOMING
        self.level = CallLevel.AGENT
        self.agent = None

class Agent(object):
    def __init__(self, name, call_center):
        self.name = name
        self.state = AgentState.READY
        self.call_center = call_center
        self.call = None
        
    def escalate(self):
        if self.call:
            self.call.state = CallState.ESCALATING
            self.call.level = CallLevel.MANAGER
            self.call_center.handle_call(self.call)
            self.state = AgentState.READY
        
    
    def pickup_call(self, call):
        self.call = call
        self.state = AgentState.ONCALL
        call.state = CallState.PROGRESS
    
    def complete_call(self):
        if self.call:
            self.call.state = CallState.FINISHED
            self.state = AgentState.READY
            self.call_center.notify_agent_available()
    
    def take_break(self):
        self.state = AgentState.OFFLINE

class Manager(Agent):
    def escalate(self, call):
        raise NotImplementedError('Manager can not escalate')


## enums
import enum

class CallState(enum.Enum):
    INCOMING = 1
    PROGRESS = 2
    ESCALATING = 3
    FINISHED = 4
    
class AgentState(enum.Enum):
    OFFLINE = 1
    READY = 2
    ONCALL = 3
    
class CallLevel(enum.Enum):
    AGENT = 1
    MANAGER = 2

## data objects
class Customer(object):
    def __init__(self, name, address):
        self.name = name
        self.address = address

class Address(object):
    def __init__(self, address_text, country, pincode):
        self.address_text = address_text
        self.country = country
        self.pincode = pincode

In [53]:

import unittest

class CallCenterTest(unittest.TestCase):
    def setUp(self):
        cc = CallCenter(Address('nukkad', 'India', '100001'), '9999888800')
        agents = [Agent('Ram', cc), Agent('Sham', cc)]
        managers = [Manager('Ramesh', cc), Manager('Shamshul', cc)]
        cc.agents = agents
        cc.managers = managers
        self.cc = cc
    
    def test_call_assignment(self):
        call = Call(Customer('Titu', Address('Dwarika', 'India', '100006')))
        self.cc.handle_call(call)
        self.assertEqual(call.agent.name, 'Ram')
    
    def test_call_enqueue(self):
        call1 = Call(Customer('Titu', Address('Dwarika', 'India', '100006')))
        call2 = Call(Customer('Mitu', Address('Dwarika', 'India', '100006')))
        call3 = Call(Customer('Jitu', Address('Dwarika', 'India', '100006')))
        call4 = Call(Customer('Kitu', Address('Dwarika', 'India', '100006')))
        call5 = Call(Customer('Pitu', Address('Dwarika', 'India', '100006')))
        
        self.cc.handle_call(call1)
        self.cc.handle_call(call2)
        self.cc.handle_call(call3)
        self.cc.handle_call(call4)
        self.cc.handle_call(call5)
        
        self.assertEqual(self.cc.wait_queue[0].customer.name, 'Pitu')
    
    def test_call_escalation(self):
        call = Call(Customer('Titu', Address('Dwarika', 'India', '100006')))
        self.cc.handle_call(call)
        call.agent.escalate()
        self.assertEqual(type(call.agent), Manager)
    
    def test_call_pickup_from_queue(self):
        call1 = Call(Customer('Titu', Address('Dwarika', 'India', '100006')))
        call2 = Call(Customer('Mitu', Address('Dwarika', 'India', '100006')))
        call3 = Call(Customer('Jitu', Address('Dwarika', 'India', '100006')))
        call4 = Call(Customer('Kitu', Address('Dwarika', 'India', '100006')))
        call5 = Call(Customer('Pitu', Address('Dwarika', 'India', '100006')))
        
        self.cc.handle_call(call1)
        self.cc.handle_call(call2)
        self.cc.handle_call(call3)
        self.cc.handle_call(call4)
        self.cc.handle_call(call5)
        
        self.assertEqual(self.cc.wait_queue[0].customer.name, 'Pitu')
        call1.agent.complete_call()
        self.assertEqual(call5.agent.name, 'Ram')
    
    def test_call_completion(self):
        call = Call(Customer('Titu', Address('Dwarika', 'India', '100006')))
        self.cc.handle_call(call)
        call.agent.complete_call()
        self.assertEqual(call.state, CallState.FINISHED)
    
    def test_all_agents_on_break(self):
        self.cc.agents[0].state = AgentState.OFFLINE
        self.cc.agents[1].state = AgentState.OFFLINE
        call = Call(Customer('Titu', Address('Dwarika', 'India', '100006')))
        self.cc.handle_call(call)
        self.assertEqual(type(call.agent), Manager)
    
    def tearDown(self):
        del self.cc

if __name__ == '__main__':
        unittest.main(argv=['first-arg-is-ignored'], exit=False)

......
----------------------------------------------------------------------
Ran 6 tests in 0.005s

OK