Newer
Older
'''
Created on Dec 20, 2012
@author: hvanwelbergen
'''
import unittest
from mockito import mock
from mockito import verify
from mockito import any
from mockito import when
from mockito import times
from notifier import ComponentNotifier
from hamcrest.core.base_matcher import BaseMatcher
from ipaaca import IUEventType
from ipaaca import Message
from ipaaca import InputBuffer
from ipaaca import OutputBuffer
import time
import os
class IUCategoryMatcher(BaseMatcher):
def __init__(self, expected_cat):
self.expected_cat = expected_cat
def _matches_(self, iu):
return iu.category==self.expected_cat
def describe_to(self, description):
description.append_text("IU with category :"+self.expected_cat)
class ComponentNotifierTest(unittest.TestCase):
RECV_CAT = set(["testrec1", "testrc2"])
SEND_CAT = set(["testsnd1", "testsnd2", "testsnd3"])
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
def setUp(self):
self.mockOutBuffer = mock()
self.mockInBuffer = mock()
self.notifier = ComponentNotifier("testcomp","testfunc", ComponentNotifierTest.SEND_CAT, ComponentNotifierTest.RECV_CAT, self.mockOutBuffer, self.mockInBuffer)
self.notifier.initialize()
def tearDown(self):
pass
def _sendNotify(self, state, receiveCats):
mockIUNotify = Message(ComponentNotifier.NOTIFY_CATEGORY)
mockIUNotify.payload[ComponentNotifier.STATE] = state;
mockIUNotify.payload[ComponentNotifier.NAME] = "namex";
mockIUNotify.payload[ComponentNotifier.SEND_CATEGORIES] = "";
mockIUNotify.payload[ComponentNotifier.RECEIVE_CATEGORIES] = ",".join(receiveCats);
self.notifier._handle_iu_event(mockIUNotify, IUEventType.ADDED, False)
def testNotifyAtInit(self):
verify(self.mockOutBuffer).add(any())
#TODO: python mockito cannot yet use hamcrest matchers, so cannot easily test if the message is correct :(
#assertEquals(ComponentNotifier.NOTIFY_CATEGORY, iu.getCategory());
#assertEquals("new", iu.getPayload().get(ComponentNotifier.STATE));
#assertThat(ImmutableSet.copyOf(iu.getPayload().get(ComponentNotifier.RECEIVE_CATEGORIES).split(",")),
# IsIterableContainingInAnyOrder.containsInAnyOrder(RECV_CAT.toArray(new String[0])));
#assertThat(ImmutableSet.copyOf(iu.getPayload().get(ComponentNotifier.SEND_CATEGORIES).split(",")),
# IsIterableContainingInAnyOrder.containsInAnyOrder(SEND_CAT.toArray(new String[0])));
def testNotifyAtNotifyNew(self):
self._sendNotify("new", {"testsnd1"});
verify(self.mockOutBuffer, times(2)).add(any())
#TODO: python mockito cannot yet use hamcrest matchers, so cannot easily test if the message is correct :(
#ArgumentCaptor<LocalIU> argument = ArgumentCaptor.forClass(LocalIU.class);
#verify(mockOutBuffer, times(2)).add(argument.capture());
#LocalIU iu = argument.getAllValues().get(1);
#assertEquals("componentNotify", iu.getCategory());
#assertEquals("old", iu.getPayload().get("state"));
def testNoNotifyAtNotifyOld(self):
self._sendNotify("old", {"testsnd1"});
verify(self.mockOutBuffer, times(1)).add(any())
class MyListener(object):
def __init__(self):
self.numCalled = 0
def handle(self, iu, mytype, local):
self.numCalled += 1
class ComponentNotifierIntegrationTest(unittest.TestCase):
def setUp(self):
pass
def tearDown(self):
pass
def _setupCompNotifier(self, idx, sendList, recvList):
inBuffer = InputBuffer(idx+"in", {ComponentNotifier.NOTIFY_CATEGORY})
outBuffer = OutputBuffer(idx+"out")
return ComponentNotifier(idx, "testfunction", sendList, recvList, outBuffer, inBuffer)
# bug: this somehow remains active after running
# def testSelf(self):
# notifier = self._setupCompNotifier("not", {"a1","b1"}, {"a3","b1"});
# listener = MyListener()
# notifier.add_notification_handler(listener.handle);
#
# notifier.initialize();
# time.sleep(0.5);
#
# self.assertEquals(0, listener.numCalled);
def testTwo(self):
notifier1 = self._setupCompNotifier("not1", {"a1", "b1"}, {"a3", "b2"});
notifier2 = self._setupCompNotifier("not2", {"a2", "b2"}, {"a3", "b1"});
listener1 = MyListener()
listener2 = MyListener()
notifier1.add_notification_handler(listener1.handle)
notifier2.add_notification_handler(listener2.handle)
notifier1.initialize()
time.sleep(0.5)
notifier2.initialize()
time.sleep(0.5)
self.assertEqual(1, listener1.numCalled)
self.assertEqual(1, listener2.numCalled)
if __name__ == "__main__":
#import sys;sys.argv = ['', 'Test.testName']
unittest.main()
os._exit(0)