Commit bb0141e2 authored by Vincent Pelletier's avatar Vincent Pelletier

Move all common test functions into a single class, making it more reusable between tests.


git-svn-id: https://svn.erp5.org/repos/neo/branches/prototype3@1127 71dcc9de-d417-0410-9af5-da40c76e7ee4
parent bacc91d9
...@@ -21,18 +21,66 @@ import os ...@@ -21,18 +21,66 @@ import os
import sys import sys
import time import time
import signal import signal
import random
import MySQLdb import MySQLdb
import tempfile import tempfile
import traceback import traceback
# No need to protect this list with a lock, NEOProcess instanciations and from neo import setupLog
# killallNeo calls are done from a single thread. from neo import logging
neo_process_list = [] from neo.client.Storage import Storage
NEO_CONFIG_HEADER = """
[DEFAULT]
master_nodes: %(master_nodes)s
replicas: %(replicas)s
partitions: %(partitions)s
name: %(name)s
user: %(user)s
password: %(password)s
connector: SocketConnector
[admin]
server: 127.0.0.1:%(port)s
"""
NEO_CONFIG_MASTER = """
[%(id)s]
server: 127.0.0.1:%(port)s
"""
NEO_CONFIG_STORAGE = """
[%(id)s]
database: %(db)s
server: 127.0.0.1:%(port)s
"""
NEO_MASTER_ID = 'master%s'
NEO_STORAGE_ID = 'storage%s'
NEO_MASTER = 'neomaster'
NEO_STORAGE = 'neostorage'
NEO_ADMIN = 'neoadmin'
class AlreadyRunning(Exception):
pass
class AlreadyStopped(Exception):
pass
class NEOProcess: class NEOProcess:
pid = 0 pid = 0
def __init__(self, command, *args): def __init__(self, command, *args):
self.command = command
self.args = args
def start(self):
# Prevent starting when already forked and wait wasn't called.
if self.pid != 0:
raise AlreadyRunning, 'Already running with PID %r' % (self.pid, )
command = self.command
args = self.args
self.pid = os.fork() self.pid = os.fork()
if self.pid == 0: if self.pid == 0:
# Child # Child
...@@ -49,8 +97,6 @@ class NEOProcess: ...@@ -49,8 +97,6 @@ class NEOProcess:
# traceback, replace stdout & stderr. # traceback, replace stdout & stderr.
sys.stdout = sys.stderr = open('/dev/null', 'w') sys.stdout = sys.stderr = open('/dev/null', 'w')
raise KeyboardInterrupt raise KeyboardInterrupt
else:
neo_process_list.append(self)
def kill(self, sig=signal.SIGTERM): def kill(self, sig=signal.SIGTERM):
if self.pid: if self.pid:
...@@ -58,6 +104,8 @@ class NEOProcess: ...@@ -58,6 +104,8 @@ class NEOProcess:
os.kill(self.pid, sig) os.kill(self.pid, sig)
except OSError: except OSError:
traceback.print_last() traceback.print_last()
else:
raise AlreadyStopped
def __del__(self): def __del__(self):
# If we get killed, kill subprocesses aswell. # If we get killed, kill subprocesses aswell.
...@@ -70,182 +118,139 @@ class NEOProcess: ...@@ -70,182 +118,139 @@ class NEOProcess:
pass pass
def wait(self, options=0): def wait(self, options=0):
assert self.pid if self.pid == 0:
return os.WEXITSTATUS(os.waitpid(self.pid, options)[1]) raise AlreadyStopped
result = os.WEXITSTATUS(os.waitpid(self.pid, options)[1])
def killallNeo(): self.pid = 0
while len(neo_process_list): return result
process = neo_process_list.pop()
process.kill()
process.wait()
NEO_MASTER = 'neomaster' class NEOCluster(object):
NEO_STORAGE = 'neostorage' def __init__(self, db_list, master_node_count=1,
NEO_ADMIN = 'neoadmin' partitions=1, replicas=0, port_base=10000,
NEO_PORT_BASE = 10010 db_user='neo', db_password='neo',
NEO_CLUSTER_NAME = 'test' db_super_user='root', db_super_password=None,
NEO_MASTER_PORT_1 = NEO_PORT_BASE cleanup_on_delete=False):
NEO_MASTER_PORT_2 = NEO_MASTER_PORT_1 + 1 self.cleanup_on_delete = cleanup_on_delete
NEO_MASTER_PORT_3 = NEO_MASTER_PORT_2 + 1 self.db_super_user = db_super_user
NEO_STORAGE_PORT_1 = NEO_MASTER_PORT_3 + 1 self.db_super_password = db_super_password
NEO_STORAGE_PORT_2 = NEO_STORAGE_PORT_1 + 1 self.db_user = db_user
NEO_STORAGE_PORT_3 = NEO_STORAGE_PORT_2 + 1 self.db_password = db_password
NEO_STORAGE_PORT_4 = NEO_STORAGE_PORT_3 + 1 self.db_list = db_list
NEO_ADMIN_PORT = NEO_STORAGE_PORT_4 + 1 self.process_list = []
NEO_MASTER_NODES = '127.0.0.1:%(port_1)s 127.0.0.1:%(port_2)s 127.0.0.1:%(port_3)s' % { self.last_port = port_base
'port_1': NEO_MASTER_PORT_1, self.temp_dir = temp_dir = tempfile.mkdtemp(prefix='neo_')
'port_2': NEO_MASTER_PORT_2, print 'Using temp directory %r.' % (temp_dir, )
'port_3': NEO_MASTER_PORT_3 config_file_path = os.path.join(temp_dir, 'neo.conf')
} config_file = open(config_file_path, 'w')
NEO_SQL_USER = 'test' neo_admin_port = self.__allocatePort()
NEO_SQL_PASSWORD = '' self.cluster_name = cluster_name = 'neo_%s' % (random.randint(0, 100), )
NEO_SQL_DATABASE_1 = 'test_neo1' master_node_dict = {}
NEO_SQL_DATABASE_2 = 'test_neo2' for master in xrange(master_node_count):
NEO_SQL_DATABASE_3 = 'test_neo3' master_node_dict[NEO_MASTER_ID % (master, )] = \
NEO_SQL_DATABASE_4 = 'test_neo4' self.__allocatePort()
# Used to create & drop above databases and grant test users privileges. self.master_nodes = master_nodes = ' '.join('127.0.0.1:%s' %
SQL_ADMIN_USER = 'root' (x, ) for x in master_node_dict.itervalues())
SQL_ADMIN_PASSWORD = None config_file.write(NEO_CONFIG_HEADER % {
'master_nodes': master_nodes,
NEO_CONFIG = ''' 'replicas': replicas,
# Default parameters. 'partitions': partitions,
[DEFAULT] 'name': cluster_name,
# The list of master nodes. 'user': db_user,
master_nodes: %(master_nodes)s 'password': db_password,
# The number of replicas. 'port': neo_admin_port,
replicas: 2 })
# The number of partitions. self.__newProcess(NEO_ADMIN, '-vc', config_file_path, '-s', 'admin',
partitions: 1009 '-l', os.path.join(temp_dir, 'admin.log'))
# The name of this cluster. for config_id, port in master_node_dict.iteritems():
name: %(name)s config_file.write(NEO_CONFIG_MASTER % {
# The user name for the database. 'id': config_id,
user: %(user)s 'port': port,
# The password for the database. })
password: %(password)s self.__newProcess(NEO_MASTER, '-vc', config_file_path, '-s',
# The connector class used config_id, '-l',
connector: SocketConnector os.path.join(temp_dir, '%s.log' % (config_id)))
for storage, db in enumerate(db_list):
config_id = NEO_STORAGE_ID % (storage, )
config_file.write(NEO_CONFIG_STORAGE % {
'id': config_id,
'db': db,
'port': self.__allocatePort(),
})
self.__newProcess(NEO_STORAGE, '-vc', config_file_path, '-s',
config_id, '-l',
os.path.join(temp_dir, '%s.log' % (config_id)))
config_file.close()
self.neoctl = NeoCTL('127.0.0.1', neo_admin_port,
'SocketConnector')
# The admin node. def __newProcess(self, command, *args):
[admin] self.process_list.append(NEOProcess(command, *args))
server: 127.0.0.1:%(admin_port)s
# The first master.
[master1]
server: 127.0.0.1:%(master1_port)s
# The second master.
[master2]
server: 127.0.0.1:%(master2_port)s
# The third master.
[master3]
server: 127.0.0.1:%(master3_port)s
# The first storage.
[storage1]
database: %(storage1_db)s
server: 127.0.0.1:%(storage1_port)s
# The first storage.
[storage2]
database: %(storage2_db)s
server: 127.0.0.1:%(storage2_port)s
# The third storage.
[storage3]
database: %(storage3_db)s
server: 127.0.0.1:%(storage3_port)s
# The fourth storage.
[storage4]
database: %(storage4_db)s
server: 127.0.0.1:%(storage4_port)s
''' % {
'master_nodes': NEO_MASTER_NODES,
'name': NEO_CLUSTER_NAME,
'user': NEO_SQL_USER,
'password': NEO_SQL_PASSWORD,
'admin_port': NEO_ADMIN_PORT,
'master1_port': NEO_MASTER_PORT_1,
'master2_port': NEO_MASTER_PORT_2,
'master3_port': NEO_MASTER_PORT_3,
'storage1_port': NEO_STORAGE_PORT_1,
'storage1_db': NEO_SQL_DATABASE_1,
'storage2_port': NEO_STORAGE_PORT_2,
'storage2_db': NEO_SQL_DATABASE_2,
'storage3_port': NEO_STORAGE_PORT_3,
'storage3_db': NEO_SQL_DATABASE_3,
'storage4_port': NEO_STORAGE_PORT_4,
'storage4_db': NEO_SQL_DATABASE_4
}
temp_dir = tempfile.mkdtemp(prefix='neo_')
print 'Using temp directory %r.' % (temp_dir, )
config_file_path = os.path.join(temp_dir, 'neo.conf')
config_file = open(config_file_path, 'w')
config_file.write(NEO_CONFIG)
config_file.close()
m1_log = os.path.join(temp_dir, 'm1.log')
m2_log = os.path.join(temp_dir, 'm2.log')
m3_log = os.path.join(temp_dir, 'm3.log')
s1_log = os.path.join(temp_dir, 's1.log')
s2_log = os.path.join(temp_dir, 's2.log')
s3_log = os.path.join(temp_dir, 's3.log')
s4_log = os.path.join(temp_dir, 's4.log')
a_log = os.path.join(temp_dir, 'a.log')
def __allocatePort(self):
port = self.last_port
self.last_port += 1
return port
from neo import setupLog def setupDB(self):
client_log = os.path.join(temp_dir, 'c.log') # Cleanup or bootstrap databases
setupLog('CLIENT', filename=client_log, verbose=True) connect_arg_dict = {'user': self.db_super_user}
from neo import logging password = self.db_super_password
from neo.client.Storage import Storage if password is not None:
connect_arg_dict['passwd'] = password
sql_connection = MySQLdb.Connect(**connect_arg_dict)
cursor = sql_connection.cursor()
for database in self.db_list:
cursor.execute('DROP DATABASE IF EXISTS `%s`' % (database, ))
cursor.execute('CREATE DATABASE `%s`' % (database, ))
cursor.execute('GRANT ALL ON `%s`.* TO "%s"@"localhost" '\
'IDENTIFIED BY "%s"' % (database, self.db_user,
self.db_password))
cursor.close()
sql_connection.close()
neoctl = NeoCTL('127.0.0.1', NEO_ADMIN_PORT, 'SocketConnector') def start(self):
neoctl = self.neoctl
def startNeo(): assert len(self.process_list)
# Stop NEO cluster (if running) for process in self.process_list:
killallNeo() process.start()
# Cleanup or bootstrap databases # Try to put cluster in running state. This will succeed as soon as
connect_arg_dict = {'user': SQL_ADMIN_USER} # admin node could connect to the primary master node.
if SQL_ADMIN_PASSWORD is not None: while True:
connect_arg_dict['passwd'] = SQL_ADMIN_PASSWORD try:
sql_connection = MySQLdb.Connect(**connect_arg_dict) neoctl.startCluster()
cursor = sql_connection.cursor() except NotReadyException:
for database in (NEO_SQL_DATABASE_1, NEO_SQL_DATABASE_2, NEO_SQL_DATABASE_3, NEO_SQL_DATABASE_4): time.sleep(0.5)
cursor.execute('DROP DATABASE IF EXISTS %s' % (database, )) else:
cursor.execute('CREATE DATABASE %s' % (database, )) break
cursor.execute('GRANT ALL ON %s.* TO "%s"@"localhost" IDENTIFIED BY "%s"' % (database, NEO_SQL_USER, NEO_SQL_PASSWORD)) target_count = len(self.db_list)
cursor.close() while True:
sql_connection.close() storage_node_list = neoctl.getNodeList(
# Start NEO cluster node_type=protocol.STORAGE_NODE_TYPE)
NEOProcess(NEO_MASTER, '-vc', config_file_path, '-s', 'master1', '-l', m1_log) if len(storage_node_list) == target_count:
NEOProcess(NEO_MASTER, '-vc', config_file_path, '-s', 'master2', '-l', m2_log) break
NEOProcess(NEO_MASTER, '-vc', config_file_path, '-s', 'master3', '-l', m3_log)
NEOProcess(NEO_STORAGE, '-vRc', config_file_path, '-s', 'storage1', '-l', s1_log)
NEOProcess(NEO_STORAGE, '-vRc', config_file_path, '-s', 'storage2', '-l', s2_log)
NEOProcess(NEO_STORAGE, '-vRc', config_file_path, '-s', 'storage3', '-l', s3_log)
NEOProcess(NEO_STORAGE, '-vRc', config_file_path, '-s', 'storage4', '-l', s4_log)
NEOProcess(NEO_ADMIN, '-vc', config_file_path, '-s', 'admin', '-l', a_log)
# Try to put cluster in running state. This will succeed as soon as
# admin node could connect to the primary master node.
while True:
try:
neoctl.startCluster()
except NotReadyException:
time.sleep(0.5) time.sleep(0.5)
else: neoctl.enableStorageList([x[2] for x in storage_node_list])
break
while True: def stop(self):
storage_node_list = neoctl.getNodeList( for process in self.process_list:
node_type=protocol.STORAGE_NODE_TYPE) try:
if len(storage_node_list) == 4: process.kill()
break process.wait()
time.sleep(0.5) except AlreadyStopped:
neoctl.enableStorageList([x[2] for x in storage_node_list]) pass
def getNeoStorage(): def getNEOCTL(self):
return Storage( return self.neoctl
master_nodes=NEO_MASTER_NODES,
name=NEO_CLUSTER_NAME, def getStorage(self):
connector='SocketConnector') #setupLog('CLIENT', filename=os.path.join(self.temp_dir, 'client.log'),
# verbose=True)
return Storage(
master_nodes=self.master_nodes,
name=self.cluster_name,
connector='SocketConnector')
def __del__(self):
if self.cleanup_on_delete:
os.removedirs(self.temp_dir)
...@@ -23,7 +23,7 @@ from persistent import Persistent ...@@ -23,7 +23,7 @@ from persistent import Persistent
from persistent.mapping import PersistentMapping from persistent.mapping import PersistentMapping
import transaction import transaction
from neo.tests.functional import startNeo, getNeoStorage, killallNeo from neo.tests.functional import NEOCluster
class P(Persistent): class P(Persistent):
pass pass
...@@ -38,11 +38,17 @@ class DecoyIndependent(Persistent): ...@@ -38,11 +38,17 @@ class DecoyIndependent(Persistent):
def _p_independent(self): def _p_independent(self):
return 0 return 0
neo = NEOCluster(['test_neo1', 'test_neo2', 'test_neo3', 'test_neo4'],
partitions=1009, replicas=1, port_base=20000,
master_node_count=3)
class ZODBTests(unittest.TestCase): class ZODBTests(unittest.TestCase):
def setUp(self): def setUp(self):
startNeo() neo.stop()
self._storage = getNeoStorage() neo.setupDB()
neo.start()
self._storage = neo.getStorage()
self._db = ZODB.DB(self._storage) self._db = ZODB.DB(self._storage)
def populate(self): def populate(self):
...@@ -59,7 +65,7 @@ class ZODBTests(unittest.TestCase): ...@@ -59,7 +65,7 @@ class ZODBTests(unittest.TestCase):
def tearDown(self): def tearDown(self):
self._db.close() self._db.close()
self._storage.cleanup() self._storage.cleanup()
killallNeo() neo.stop()
def checkExportImport(self, abort_it=False): def checkExportImport(self, abort_it=False):
self.populate() self.populate()
......
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