95 lines
2.5 KiB
Python
95 lines
2.5 KiB
Python
import unittest
|
|
from replication import ReplicatedDatablock, ReplicatedDataFactory
|
|
import umsgpack
|
|
import logging
|
|
from replication_client import Client, Server
|
|
import time
|
|
|
|
|
|
log = logging.getLogger(__name__)
|
|
|
|
class SampleData():
|
|
def __init__(self):
|
|
self.map = {
|
|
"sample":"data",
|
|
"sample":"data",
|
|
"sample":"data",
|
|
"sample":"data"
|
|
}
|
|
|
|
class RepSampleData(ReplicatedDatablock):
|
|
def serialize(self,data):
|
|
import pickle
|
|
|
|
return pickle.dumps(data)
|
|
|
|
def deserialize(self,data):
|
|
import pickle
|
|
|
|
return pickle.loads(data)
|
|
|
|
class TestDataFactory(unittest.TestCase):
|
|
def test_data_factory(self):
|
|
factory = ReplicatedDataFactory()
|
|
factory.register_type(SampleData, RepSampleData)
|
|
data_sample = SampleData()
|
|
rep_sample = factory.construct_from_dcc(data_sample)(owner="toto", data=data_sample)
|
|
|
|
self.assertEqual(isinstance(rep_sample,RepSampleData), True)
|
|
|
|
|
|
class TestDataReplication(unittest.TestCase):
|
|
def __init__(self,methodName='runTest'):
|
|
unittest.TestCase.__init__(self, methodName)
|
|
|
|
self.factory = ReplicatedDataFactory()
|
|
self.factory.register_type(SampleData, RepSampleData)
|
|
|
|
self.server = Server(factory=self.factory)
|
|
self.server.serve()
|
|
|
|
self.client = Client(factory=self.factory, id="client_1")
|
|
self.client.connect()
|
|
|
|
self.client2 = Client(factory=self.factory, id="client_2")
|
|
self.client2.connect()
|
|
|
|
|
|
def test_register_client_data(self):
|
|
data_sample_key = self.client.register(SampleData())
|
|
|
|
#Waiting for server to receive the datas
|
|
time.sleep(2)
|
|
|
|
test_key = self.client2._rep_store[data_sample_key]
|
|
|
|
#Check if the server receive them
|
|
self.assertNotEqual(test_key, None)
|
|
|
|
def test_register_client_data2(self):
|
|
data_sample_key = self.client.register(SampleData())
|
|
|
|
#Waiting for server to receive the datas
|
|
time.sleep(2)
|
|
|
|
test_key = self.client2._rep_store[data_sample_key]
|
|
|
|
#Check if the server receive them
|
|
self.assertNotEqual(test_key, None)
|
|
|
|
|
|
|
|
|
|
|
|
def suite():
|
|
suite = unittest.TestSuite()
|
|
suite.addTest(TestDataFactory('test_data_factory'))
|
|
suite.addTest(TestDataReplication('test_register_client_data'))
|
|
suite.addTest(TestDataReplication('test_register_client_data2'))
|
|
|
|
return suite
|
|
|
|
if __name__ == '__main__':
|
|
# unittest.main()
|
|
runner = unittest.TextTestRunner(failfast=True)
|
|
runner.run(suite()) |