ldapscan/tests/TestCommunicationManager.py

import unittest
import mock

from os.path import dirname, realpath
from sys import argv, path
path.append(dirname(dirname(realpath(__file__))) + '/lib')

from Communication.Manager import CommunicationManager

class TestCommunicationManager(unittest.TestCase):
    def setUp(self):
        self._manager   = CommunicationManager()
        self._endpoint  = mock.Mock()
        self._transport = mock.Mock()
        self._event     = mock.Mock()

        self._endpoint.getHandle.return_value    = 123
        self._endpoint.getTransport.return_value = self._transport
        self._event.subject                      = self._endpoint
        self._manager.issueEvent                 = mock.Mock()

    def testEndPointAlreadyHandled(self):
        """
        there really should be a test for this to ensure the if
        is working correctly.
        """
        pass

    def testAddEndPoint(self):
        self._transport.isListen.return_value = False
        self._manager._rcons = mock.Mock()
        self._manager.addEndPoint(self._endpoint)
        self.assertIn(123, self._manager._cons)
        self.assertEqual(self._endpoint, self._manager._cons[123])
        self._manager._rcons.append.assert_called_with(123)

    def testAddListenEndPoint(self):
        self._transport.isListen.return_value = True
        self._manager._rcons = mock.Mock()
        self._manager.addEndPoint(self._endpoint)
        self.assertIn(123, self._manager._listen)
        self.assertEqual(self._endpoint, self._manager._listen[123])
        self._manager._rcons.append.assert_called_with(123)

    def test_addCon(self):
        self._manager.addEndPoint = mock.Mock()
        self.assertTrue(self._manager._addCon(self._event))
        self._manager.addEndPoint.assert_called_once_with(self._endpoint)

    def test_enableWriteOnWriteFinTransport(self):
        self._transport._fin_state = 2
        self.assertTrue(self._manager._enableWrite(self._event))
        self.assertNotIn(123, self._manager._wcons)

    def test_enableWrite(self):
        self._transport._fin_state = 0
        self.assertTrue(self._manager._enableWrite(self._event))
        self.assertIn(123, self._manager._wcons)

    def test_disableWriteNoShutdownRead(self):
        self._transport._fin_state = 0
        self.assertTrue(self._manager._disableWrite(self._event))
        self.assertNotIn(123, self._manager._wcons)
        self.test_enableWrite()
        self.assertTrue(self._manager._disableWrite(self._event))
        self.assertNotIn(123, self._manager._wcons)

    def test_disableWriteNoShutdownRead(self):
        self._transport._fin_state = 1
        self.assertTrue(self._manager._disableWrite(self._event))
        self.assertNotIn(123, self._manager._wcons)
        self.test_enableWrite()
        self.assertTrue(self._manager._disableWrite(self._event))
        self.assertNotIn(123, self._manager._wcons)
        self._manager.issueEvent.assert_called_with(
            self._endpoint, 'shutdown_write')

    def test_shutdown(self):
        self._transport.isListen.return_value = True
        endpoint2  = mock.Mock()
        transport2 = mock.Mock()
        endpoint2.getTransport.return_value = transport2
        endpoint2.getHandle.return_value    = 321
        transport2.isListen.return_value    = False
        self._manager.addEndPoint(self._endpoint)
        self._manager.addEndPoint(endpoint2)
        self.assertFalse(self._manager._shutdown(None))
        self._manager.issueEvent.assert_any_call(self._endpoint, 'close')
        self._manager.issueEvent.assert_any_call(endpoint2, 'close')
        self.assertEqual(self._manager._rcons, [])
        self.assertEqual(self._manager._wcons, [])

    def test_shutdownReadReadyToClose(self):
        self._manager._rcons.append(123)
        self._transport.shutdownRead.return_value  = 3
        self._endpoint.hasPendingData.return_value = False
        self.assertFalse(self._manager._shutdownRead(self._event))
        self._manager.issueEvent.assert_called_with(self._endpoint, 'close')

    def test_shutdownReadReadyToShutdownWrite(self):
        self._manager._rcons.append(123)
        self._transport.shutdownRead.return_value  = 0
        self._endpoint.hasPendingData.return_value = False
        self.assertFalse(self._manager._shutdownRead(self._event))
        self._manager.issueEvent.assert_called_with(self._endpoint, 'shutdown_write')

    def test_shutdownReadMarkAsClose(self):
        self._manager._rcons.append(123)
        self._transport.shutdownRead.return_value  = 0
        self._endpoint.hasPendingData.return_value = True
        self.assertFalse(self._manager._shutdownRead(self._event))
        self._endpoint.setClose.assert_called_once_with()

    def test_shutdownWriteReadyToClose(self):
        self._manager._wcons.append(123)
        self._transport.shutdownWrite.return_value = 3
        self.assertFalse(self._manager._shutdownWrite(self._event))
        self._manager.issueEvent.assert_called_once_with(self._endpoint, 'close')

    def test_shutdownWrite(self):
        self._manager._wcons.append(123)
        self._transport.shutdownWrite.return_value = 0
        self.assertFalse(self._manager._shutdownWrite(self._event))

    def test_closeCon(self):
        self._manager._wcons.append(123)
        self._manager._rcons.append(123)
        self._manager._cons[123] = self._endpoint
        self.assertFalse(self._manager._close(self._event))
        self._transport.shutdown.assert_called_with()
        self._transport.close.assert_called_with()
        self.assertNotIn(123, self._manager._wcons)
        self.assertNotIn(123, self._manager._rcons)
        self.assertNotIn(123, self._manager._cons)

    def test_closeListen(self):
        self._manager._wcons.append(123)
        self._manager._rcons.append(123)
        self._manager._listen[123] = self._endpoint
        self.assertFalse(self._manager._close(self._event))
        self._transport.shutdown.assert_called_with()
        self._transport.close.assert_called_with()
        self.assertNotIn(123, self._manager._wcons)
        self.assertNotIn(123, self._manager._rcons)
        self.assertNotIn(123, self._manager._listen)

def suite():
    return unittest.TestLoader().loadTestsFromTestCase(TestCommunicationManager)

if __name__ == '__main__':
    unittest.TextTestRunner(verbosity=2).run(suite())

# vim: set ft=python et ts=8 sw=4 sts=4: