testStorageHandler.py 10.8 KB
Newer Older
Aurel's avatar
Aurel committed
1
#
Grégory Wisniewski's avatar
Grégory Wisniewski committed
2
# Copyright (C) 2009-2010  Nexedi SA
3
#
Aurel's avatar
Aurel committed
4 5 6 7
# 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.
8
#
Aurel's avatar
Aurel committed
9 10 11 12 13 14 15
# 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, write to the Free Software
16
# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA.
Aurel's avatar
Aurel committed
17

18 19
import unittest
from mock import Mock
Grégory Wisniewski's avatar
Grégory Wisniewski committed
20
from struct import pack
21
from neo.tests import NeoTestBase
22
from neo.protocol import NodeTypes, NodeStates, Packets
23
from neo.master.handlers.storage import StorageServiceHandler
24
from neo.master.handlers.client import ClientServiceHandler
25
from neo.master.app import Application
26
from neo.exception import OperationFailure
27

28
class MasterStorageHandlerTests(NeoTestBase):
29 30 31

    def setUp(self):
        # create an application object
32
        config = self.getMasterConfiguration(master_number=1, replicas=1)
33
        self.app = Application(config)
34
        self.app.pt.clear()
35
        self.app.em = Mock()
36
        self.service = StorageServiceHandler(self.app)
37
        self.client_handler = ClientServiceHandler(self.app)
38 39 40 41
        # define some variable to simulate client and storage node
        self.client_port = 11022
        self.storage_port = 10021
        self.master_port = 10010
42 43 44
        self.master_address = ('127.0.0.1', self.master_port)
        self.client_address = ('127.0.0.1', self.client_port)
        self.storage_address = ('127.0.0.1', self.storage_port)
45

46
    def tearDown(self):
47
        NeoTestBase.tearDown(self)
48

49 50 51 52 53
    def _allocatePort(self):
        self.port = getattr(self, 'port', 1000) + 1
        return self.port

    def _getClient(self):
54
        return self.identifyToMasterNode(node_type=NodeTypes.CLIENT,
55
                ip='127.0.0.1', port=self._allocatePort())
56

57 58 59 60
    def _getStorage(self):
        return self.identifyToMasterNode(node_type=NodeTypes.STORAGE,
                ip='127.0.0.1', port=self._allocatePort())

61 62 63
    def getLastUUID(self):
        return self.uuid

64
    def identifyToMasterNode(self, node_type=NodeTypes.STORAGE, ip="127.0.0.1",
65 66 67
                             port=10021):
        """Do first step of identification to MN
        """
68
        nm = self.app.nm
69
        uuid = self.getNewUUID()
70 71 72
        node = nm.createFromNodeType(node_type, address=(ip, port),
                uuid=uuid)
        conn = self.getFakeConnection(node.getUUID(),node.getAddress())
73
        node.setConnection(conn)
74
        return (node, conn)
75

76
    def test_answerInformationLocked_1(self):
77 78 79 80 81 82 83 84
        """
            Master must refuse to lock if the TID is greater than the last TID
        """
        tid1 = self.getNextTID()
        tid2 = self.getNextTID(tid1)
        self.app.tm.setLastTID(tid1)
        self.assertTrue(tid1 < tid2)
        node, conn = self.identifyToMasterNode()
85
        self.checkProtocolErrorRaised(self.service.answerInformationLocked,
86
                conn, tid2)
87
        self.checkNoPacketSent(conn)
88

89
    def test_answerInformationLocked_2(self):
90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105
        """
            Master must:
            - lock each storage
            - notify the client
            - invalidate other clients
            - unlock storages
        """
        # one client and two storages required
        client_1, client_conn_1 = self._getClient()
        client_2, client_conn_2 = self._getClient()
        storage_1, storage_conn_1 = self._getStorage()
        storage_2, storage_conn_2 = self._getStorage()
        uuid_list = storage_1.getUUID(), storage_2.getUUID()
        oid_list = self.getOID(), self.getOID()
        msg_id = 1
        # register a transaction
106 107
        tid = self.app.tm.begin()
        self.app.tm.prepare(client_1, tid, oid_list, uuid_list, msg_id)
108 109
        self.assertTrue(tid in self.app.tm)
        # the first storage acknowledge the lock
110
        self.service.answerInformationLocked(storage_conn_1, tid)
111 112 113 114 115
        self.checkNoPacketSent(client_conn_1)
        self.checkNoPacketSent(client_conn_2)
        self.checkNoPacketSent(storage_conn_1)
        self.checkNoPacketSent(storage_conn_2)
        # then the second
116
        self.service.answerInformationLocked(storage_conn_2, tid)
117
        self.checkAnswerTransactionFinished(client_conn_1)
118
        self.checkInvalidateObjects(client_conn_2)
119 120
        self.checkNotifyUnlockInformation(storage_conn_1)
        self.checkNotifyUnlockInformation(storage_conn_2)
121

122
    def test_12_askLastIDs(self):
123
        service = self.service
124
        node, conn = self.identifyToMasterNode()
125
        # give a uuid
126
        conn = self.getFakeConnection(node.getUUID(), self.storage_address)
127
        ptid = self.app.pt.getID()
128 129 130
        oid = self.getOID(1)
        tid = self.getNextTID()
        self.app.tm.setLastOID(oid)
131
        self.app.tm.setLastTID(tid)
132 133
        service.askLastIDs(conn)
        packet = self.checkAnswerLastIDs(conn)
134
        loid, ltid, lptid = packet.decode()
135 136 137
        self.assertEqual(loid, oid)
        self.assertEqual(ltid, tid)
        self.assertEqual(lptid, ptid)
138

139
    def test_13_askUnfinishedTransactions(self):
140
        service = self.service
141
        node, conn = self.identifyToMasterNode()
142
        # give a uuid
143 144
        service.askUnfinishedTransactions(conn)
        packet = self.checkAnswerUnfinishedTransactions(conn)
145 146
        tid_list, = packet.decode()
        self.assertEqual(tid_list, [])
147
        # create some transaction
148
        node, conn = self.identifyToMasterNode(node_type=NodeTypes.CLIENT,
149
                                                port=self.client_port)
150 151 152 153 154 155 156
        def create_transaction(index):
            tid = self.getNextTID()
            oid_list = [self.getOID(index)]
            self.app.tm.prepare(node, tid, oid_list, [node.getUUID()], index)
        create_transaction(1)
        create_transaction(2)
        create_transaction(3)
157
        conn = self.getFakeConnection(node.getUUID(), self.storage_address)
158 159
        service.askUnfinishedTransactions(conn)
        packet = self.checkAnswerUnfinishedTransactions(conn)
160
        (tid_list, ) = packet.decode()
161
        self.assertEqual(len(tid_list), 3)
162

Grégory Wisniewski's avatar
Grégory Wisniewski committed
163
    def _testWithMethod(self, method, state):
164 165 166 167 168 169 170
        # define two nodes
        node1, conn1 = self.identifyToMasterNode()
        node2, conn2 = self.identifyToMasterNode()
        node1.setRunning()
        node2.setRunning()
        self.assertEquals(node1.getState(), NodeStates.RUNNING)
        self.assertEquals(node2.getState(), NodeStates.RUNNING)
171
        # filled the pt
172
        self.app.pt.make(self.app.nm.getStorageList())
173 174
        self.assertTrue(self.app.pt.filled())
        self.assertTrue(self.app.pt.operational())
175
        # drop one node
176
        lptid = self.app.pt.getID()
177 178
        method(conn1)
        self.assertEquals(node1.getState(), state)
179
        self.assertTrue(lptid < self.app.pt.getID())
180
        # drop the second, no storage node left
181
        lptid = self.app.pt.getID()
182 183 184
        self.assertEquals(node2.getState(), NodeStates.RUNNING)
        self.assertRaises(OperationFailure, method, conn2)
        self.assertEquals(node2.getState(), state)
185
        self.assertEquals(lptid, self.app.pt.getID())
186

Grégory Wisniewski's avatar
Grégory Wisniewski committed
187 188 189
    def test_15_peerBroken(self):
        self._testWithMethod(self.service.peerBroken, NodeStates.BROKEN)

190
    def test_16_timeoutExpired(self):
Grégory Wisniewski's avatar
Grégory Wisniewski committed
191
        self._testWithMethod(self.service.timeoutExpired,
192
                NodeStates.TEMPORARILY_DOWN)
193 194

    def test_17_connectionClosed(self):
Grégory Wisniewski's avatar
Grégory Wisniewski committed
195
        self._testWithMethod(self.service.connectionClosed,
196
                NodeStates.TEMPORARILY_DOWN)
197

198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219
    def test_nodeLostAfterAskLockInformation(self):
        # 2 storage nodes, one will die
        node1, conn1 = self._getStorage()
        node2, conn2 = self._getStorage()
        # client nodes, to distinguish answers for the sample transactions
        client1, cconn1 = self._getClient()
        client2, cconn2 = self._getClient()
        client3, cconn3 = self._getClient()
        tid1 = self.getNextTID()
        tid2 = self.getNextTID(tid1)
        tid3 = self.getNextTID(tid2)
        oid_list = [self.getOID(), ]

        # Some shortcuts to simplify test code
        self.app.pt = Mock({'operational': True})
        self.app.outdateAndBroadcastPartition = lambda: None

        # Register some transactions
        tm = self.app.tm
        # Transaction 1: 2 storage nodes involved, one will die and the other
        # already answered node lock
        msg_id_1 = 1
220
        tm.prepare(client1, tid1, oid_list, [node1.getUUID(), node2.getUUID()], msg_id_1)
221 222 223
        tm.lock(tid1, node2.getUUID())
        # Transaction 2: 2 storage nodes involved, one will die
        msg_id_2 = 2
224
        tm.prepare(client2, tid2, oid_list, [node1.getUUID(), node2.getUUID()], msg_id_2)
225 226
        # Transaction 3: 1 storage node involved, which won't die
        msg_id_3 = 3
227
        tm.prepare(client3, tid3, oid_list, [node2.getUUID(), ], msg_id_3)
228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247

        # Assert initial state
        self.checkNoPacketSent(cconn1)
        self.checkNoPacketSent(cconn2)
        self.checkNoPacketSent(cconn3)

        # Storage 1 dies
        node1.setTemporarilyDown()
        self.service.nodeLost(conn1, node1)

        # Check state after node lost
        # T1: last locking node lost, client receives AnswerTransactionFinished
        self.checkAnswerTransactionFinished(cconn1)
        # ...and notifications are sent to other clients
        self.checkInvalidateObjects(cconn2)
        self.checkInvalidateObjects(cconn3)
        # T2: pending locking answer, client keeps waiting
        self.checkNoPacketSent(cconn2, check_notify=False)
        # T3: action not significant to this transacion, so no response
        self.checkNoPacketSent(cconn3, check_notify=False)
248

249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272
    def test_answerPack(self):
        # Note: incomming status has no meaning here, so it's left to False.
        node1, conn1 = self._getStorage()
        node2, conn2 = self._getStorage()
        self.app.packing = None
        # Does nothing
        self.service.answerPack(None, False)

        client_conn = Mock({
            'getPeerId': 512,
        })
        client_peer_id = 42
        self.app.packing = (client_conn, client_peer_id, set([conn1.getUUID(),
            conn2.getUUID()]))
        self.service.answerPack(conn1, False)
        self.checkNoPacketSent(client_conn)
        self.assertEqual(self.app.packing[2], set([conn2.getUUID(), ]))
        self.service.answerPack(conn2, False)
        status = self.checkAnswerPacket(client_conn, Packets.AnswerPack,
            decode=True)[0]
        # TODO: verify packet peer id
        self.assertTrue(status)
        self.assertEqual(self.app.packing, None)

273 274 275
if __name__ == '__main__':
    unittest.main()