Commit 71e30fb9 authored by Julien Muchembled's avatar Julien Muchembled

Remove useless testEvent

Such kind of test has never helped to detect regressions and any bug in
EpollEventManager would be quickly reported by other tests.

testConnection may go the same way if it keeps annoying me too much.
parent f9df31be
...@@ -34,7 +34,6 @@ UNIT_TEST_MODULES = [ ...@@ -34,7 +34,6 @@ UNIT_TEST_MODULES = [
# generic parts # generic parts
'neo.tests.testBootstrap', 'neo.tests.testBootstrap',
'neo.tests.testConnection', 'neo.tests.testConnection',
'neo.tests.testEvent',
'neo.tests.testHandler', 'neo.tests.testHandler',
'neo.tests.testNodes', 'neo.tests.testNodes',
'neo.tests.testDispatcher', 'neo.tests.testDispatcher',
......
#
# Copyright (C) 2009-2015 Nexedi SA
#
# This program is free software; you can redistribute it and/or
# modify it under the terms of the GNU General Public License
# as published by the Free Software Foundation; either version 2
# of the License, or (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
import unittest
from mock import Mock
from . import NeoUnitTestBase
from select import epoll, EPOLLIN, EPOLLOUT
from neo.lib.event import EpollEventManager
class EventTests(NeoUnitTestBase):
def test_01_EpollEventManager(self):
# init one
em = EpollEventManager()
self.assertEqual(len(em.connection_dict), 0)
self.assertEqual(len(em.reader_set), 0)
self.assertEqual(len(em.writer_set), 0)
self.assertTrue(isinstance(em.epoll, epoll))
# use a mock object instead of epoll
em.epoll = Mock()
connector = self.getFakeConnector(descriptor=1014)
conn = self.getFakeConnection(connector=connector)
self.assertEqual(len(em.getConnectionList()), 0)
# test register/unregister
em.register(conn)
self.assertEqual(len(connector.mockGetNamedCalls("getDescriptor")), 1)
self.assertEqual(len(em.epoll.mockGetNamedCalls("register")), 1)
call = em.epoll.mockGetNamedCalls("register")[0]
data = call.getParam(0)
self.assertEqual(data, 1014)
self.assertEqual(len(em.getConnectionList()), 1)
self.assertEqual(em.getConnectionList()[0].getDescriptor(), conn.getDescriptor())
connector = self.getFakeConnector(descriptor=1014)
conn = self.getFakeConnection(connector=connector)
em.unregister(conn)
self.assertEqual(len(connector.mockGetNamedCalls("getDescriptor")), 1)
self.assertEqual(len(em.epoll.mockGetNamedCalls("unregister")), 1)
call = em.epoll.mockGetNamedCalls("unregister")[0]
data = call.getParam(0)
self.assertEqual(data, 1014)
self.assertEqual(len(em.getConnectionList()), 0)
# add/removeReader
conn = self.getFakeConnection()
self.assertEqual(len(em.reader_set), 0)
em.addReader(conn)
self.assertEqual(len(em.reader_set), 1)
self.assertEqual(len(em.epoll.mockGetNamedCalls("modify")), 1)
em.addReader(conn) # do not add if already present
self.assertEqual(len(em.reader_set), 1)
self.assertEqual(len(em.epoll.mockGetNamedCalls("modify")), 1)
em.removeReader(conn)
self.assertEqual(len(em.reader_set), 0)
self.assertEqual(len(em.epoll.mockGetNamedCalls("modify")), 2)
em.removeReader(conn)
self.assertEqual(len(em.reader_set), 0)
self.assertEqual(len(em.epoll.mockGetNamedCalls("modify")), 2)
# add/removeWriter
conn = self.getFakeConnection()
self.assertEqual(len(em.writer_set), 0)
em.addWriter(conn)
self.assertEqual(len(em.writer_set), 1)
self.assertEqual(len(em.epoll.mockGetNamedCalls("modify")), 3)
em.addWriter(conn) # do not add if already present
self.assertEqual(len(em.writer_set), 1)
self.assertEqual(len(em.epoll.mockGetNamedCalls("modify")), 3)
em.removeWriter(conn)
self.assertEqual(len(em.writer_set), 0)
self.assertEqual(len(em.epoll.mockGetNamedCalls("modify")), 4)
em.removeWriter(conn)
self.assertEqual(len(em.writer_set), 0)
self.assertEqual(len(em.epoll.mockGetNamedCalls("modify")), 4)
# poll
r_connector = self.getFakeConnector(descriptor=14515)
r_conn = self.getFakeConnection(connector=r_connector)
em.register(r_conn)
w_connector = self.getFakeConnector(descriptor=351621)
w_conn = self.getFakeConnection(connector=w_connector)
em.register(w_conn)
em.epoll = Mock({"poll":(
(r_connector.getDescriptor(), EPOLLIN),
(w_connector.getDescriptor(), EPOLLOUT),
)})
em.poll(1)
# check it called poll on epoll
self.assertEqual(len(em.epoll.mockGetNamedCalls("poll")), 1)
call = em.epoll.mockGetNamedCalls("poll")[0]
data = call.getParam(0)
self.assertEqual(data, -1)
# need to rebuild completely this test and the the packet queue
# check readable conn
#self.assertEqual(len(r_conn.mockGetNamedCalls("readable")), 1)
#self.assertEqual(len(r_conn.mockGetNamedCalls("writable")), 0)
# check writable conn
#self.assertEqual(len(w_conn.mockGetNamedCalls("readable")), 0)
#self.assertEqual(len(w_conn.mockGetNamedCalls("writable")), 1)
if __name__ == '__main__':
unittest.main()
Markdown is supported
0%
or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment