testSSL.py 2.17 KB
Newer Older
Julien Muchembled's avatar
Julien Muchembled committed
1
#
Julien Muchembled's avatar
Julien Muchembled committed
2
# Copyright (C) 2015-2017  Nexedi SA
Julien Muchembled's avatar
Julien Muchembled committed
3 4 5 6 7 8 9 10 11 12 13 14 15 16 17
#
# 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
18
from neo.lib.protocol import Packets
Julien Muchembled's avatar
Julien Muchembled committed
19
from .. import SSL
20
from . import NEOCluster, test, testReplication
Julien Muchembled's avatar
Julien Muchembled committed
21 22 23 24 25 26 27 28 29 30 31 32 33 34 35


class SSLMixin:

    @classmethod
    def setUpClass(cls):
        NEOCluster.SSL = SSL

    @classmethod
    def tearDownClass(cls):
        NEOCluster.SSL = None


class SSLTests(SSLMixin, test.Test):
    # exclude expected failures
36 37
    testDeadlockAvoidance = None
    testUndoConflict = testUndoConflictDuringStore = None
Julien Muchembled's avatar
Julien Muchembled committed
38

39 40
    def testAbortConnection(self, after_handshake=1):
        with self.getLoopbackConnection() as conn:
41 42 43
            conn.ask(Packets.Ping())
            connector = conn.getConnector()
            del connector.connect_limit[connector.addr]
44
            conn.em.poll(1)
45 46
            self.assertTrue(isinstance(connector,
                connector.SSLHandshakeConnectorClass))
47
            self.assertNotIn(connector.getDescriptor(), conn.em.writer_set)
48 49
            if after_handshake:
                while not isinstance(connector, connector.SSLConnectorClass):
50
                    conn.em.poll(1)
51 52
            conn.abort()
            fd = connector.getDescriptor()
53 54
            while fd in conn.em.reader_set:
                conn.em.poll(1)
55 56
            self.assertIs(conn.getConnector(), None)

57 58 59
    def testAbortConnectionBeforeHandshake(self):
        self.testAbortConnection(0)

Julien Muchembled's avatar
Julien Muchembled committed
60 61 62 63 64 65 66
class SSLReplicationTests(SSLMixin, testReplication.ReplicationTests):
    # do not repeat slowest tests with SSL
    testBackupNodeLost = testBackupNormalCase = None


if __name__ == "__main__":
    unittest.main()