Commit f658d4ae authored by Grégory Wisniewski's avatar Grégory Wisniewski

Add tests for storage identification handler.

git-svn-id: https://svn.erp5.org/repos/neo/trunk@2138 71dcc9de-d417-0410-9af5-da40c76e7ee4
parent f2b24081
#
# Copyright (C) 2009-2010 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, write to the Free Software
# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA.
import unittest
from mock import Mock
from neo.tests import NeoTestBase
from neo.protocol import NodeTypes, NotReadyError, BrokenNodeDisallowedError
from neo.pt import PartitionTable
from neo.storage.app import Application
from neo.storage.handlers.identification import IdentificationHandler
class StorageIdentificationHandlerTests(NeoTestBase):
def setUp(self):
config = self.getStorageConfiguration(master_number=1)
self.app = Application(config)
self.app.name = 'NEO'
self.app.ready = True
self.app.pt = PartitionTable(4, 1)
self.identification = IdentificationHandler(self.app)
def test_requestIdentification1(self):
""" nodes are rejected during election or if unknown storage """
self.app.ready = False
self.assertRaises(
NotReadyError,
self.identification.requestIdentification,
self.getFakeConnection(),
NodeTypes.CLIENT,
self.getNewUUID(),
None,
self.app.name,
)
self.app.ready = True
self.assertRaises(
NotReadyError,
self.identification.requestIdentification,
self.getFakeConnection(),
NodeTypes.STORAGE,
self.getNewUUID(),
None,
self.app.name,
)
def test_requestIdentification3(self):
""" broken nodes must be rejected """
uuid = self.getNewUUID()
conn = self.getFakeConnection(uuid=uuid)
node = self.app.nm.createClient(uuid=uuid)
node.setBroken()
self.assertRaises(BrokenNodeDisallowedError,
self.identification.requestIdentification,
conn,
NodeTypes.CLIENT,
uuid,
None,
self.app.name,
)
def test_requestIdentification2(self):
""" accepted client must be connected and running """
uuid = self.getNewUUID()
conn = self.getFakeConnection(uuid=uuid)
node = self.app.nm.createClient(uuid=uuid)
self.identification.requestIdentification(conn, NodeTypes.CLIENT, uuid,
None, self.app.name)
self.assertTrue(node.isRunning())
self.assertTrue(node.isConnected())
self.assertEqual(node.getUUID(), uuid)
self.assertTrue(node.getConnection() is conn)
self.checkUUIDSet(conn, uuid)
args = self.checkAcceptIdentification(conn, decode=True)
node_type, address, _np, _nr, _uuid = args
self.assertEqual(node_type, NodeTypes.STORAGE)
self.assertEqual(address, None)
self.assertEqual(_uuid, uuid)
if __name__ == "__main__":
unittest.main()
......@@ -54,6 +54,7 @@ UNIT_TEST_MODULES = [
'neo.tests.storage.testStorageHandler',
'neo.tests.storage.testStorageMySQLdb',
'neo.tests.storage.testVerificationHandler',
'neo.tests.storage.testIdentificationHandler',
'neo.tests.storage.testTransactions',
# client application
'neo.tests.client.testClientApp',
......
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