Change in vdsm[master]: jsonrpc: Use the AsyncoreClient implementation in tests too
smizrahi at redhat.com
smizrahi at redhat.com
Fri Feb 1 22:16:10 UTC 2013
Saggi Mizrahi has uploaded a new change for review.
Change subject: jsonrpc: Use the AsyncoreClient implementation in tests too
......................................................................
jsonrpc: Use the AsyncoreClient implementation in tests too
No need to have two implementation for the same thing
Change-Id: I6f98075c73000c1a1bea7228eb0590c0678ea900
Signed-off-by: Saggi Mizrahi <smizrahi at redhat.com>
---
M tests/jsonRpcTests.py
M tests/jsonRpcUtils.py
M vdsm_api/jsonrpc/asyncoreReactor.py
M vdsm_api/jsonrpc/client.py
4 files changed, 61 insertions(+), 56 deletions(-)
git pull ssh://gerrit.ovirt.org:29418/vdsm refs/changes/25/11625/1
diff --git a/tests/jsonRpcTests.py b/tests/jsonRpcTests.py
index 5a08da0..5779334 100644
--- a/tests/jsonRpcTests.py
+++ b/tests/jsonRpcTests.py
@@ -89,9 +89,6 @@
with constructReactor(reactorType) as \
(reactor, clientFactory, laddr):
- t = threading.Thread(target=serve, args=(reactor,))
- t.setDaemon(True)
- t.start()
t = threading.Thread(target=echosrv.serve)
t.setDaemon(True)
@@ -99,33 +96,35 @@
reactor.createListener(laddr, echosrv.accept)
- clientNum = 10
- repeats = 10
- subRepeats = 10
+ clientNum = 4
+ repeats = 2
+ subRepeats = 4
clients = []
try:
for i in range(clientNum):
- client = clientFactory()
- client.connect()
- clients.append(client)
+ c = clientFactory()
+ c.connect()
+ clients.append(c)
for i in range(repeats):
for client in clients:
for i in range(subRepeats):
self.log.info("Sending message...")
- client.sendMessage(data, CALL_TIMEOUT)
+ client.send(data, CALL_TIMEOUT)
for i in range(repeats * subRepeats):
for client in clients:
self.log.info("Waiting for reply...")
- retData = client.recvMessage(CALL_TIMEOUT)
+ retData = client.recv(CALL_TIMEOUT)
self.log.info("Asserting reply...")
- self.assertEquals(retData, data)
+ self.assertEquals(
+ retData, data,
+ "Data is not as expected " +
+ "'%s...%s' != '%s...%s'" %
+ (retData[:10], retData[-10:],
+ data[:10], data[-10:]))
finally:
- for client in clients:
- client.close()
-
queue.put((None, None))
diff --git a/tests/jsonRpcUtils.py b/tests/jsonRpcUtils.py
index f29de18..d5ffc06 100644
--- a/tests/jsonRpcUtils.py
+++ b/tests/jsonRpcUtils.py
@@ -11,7 +11,6 @@
asyncoreReactor
from jsonrpc.client import \
JsonRpcClient, \
- TCPReactorClient, \
ProtonReactorClient
protonReactor = None
@@ -53,11 +52,20 @@
port = getFreePort()
address = ("localhost", port)
reactor = asyncoreReactor.AsyncoreReactor()
+ clientsReactor = asyncoreReactor.AsyncoreReactor()
+
+ t = threading.Thread(target=clientsReactor.process_requests)
+ t.setDaemon(True)
+ t.start()
+
+ def clientFactory(address):
+ return TestClientWrapper(clientsReactor.createClient(address))
try:
- yield reactor, partial(TCPReactorClient, address), address
+ yield reactor, partial(clientFactory, address), address
finally:
reactor.stop()
+ clientsReactor.stop()
@contextmanager
@@ -84,8 +92,32 @@
@contextmanager
def constructReactor(tp):
- with REACTOR_CONSTRUCTORS[tp]() as res:
- yield res
+ with REACTOR_CONSTRUCTORS[tp]() as (serverReactor, clientFactory, laddr):
+
+ t = threading.Thread(target=serverReactor.process_requests)
+ t.setDaemon(True)
+ t.start()
+
+ yield serverReactor, clientFactory, laddr
+
+
+class TestClientWrapper(object):
+ def __init__(self, client):
+ self._client = client
+ self._queue = Queue()
+ self._client.setInbox(self._queue)
+
+ def send(self, data, timeout=None):
+ self._client.send(data)
+
+ def recv(self, timeout=None):
+ return self._queue.get(timeout=timeout)[1]
+
+ def connect(self):
+ return self._client.connect()
+
+ def close(self):
+ return self._client.close()
@contextmanager
@@ -93,10 +125,6 @@
queue = Queue()
server = JsonRpcServer(bridge, queue)
with constructReactor(tp) as (reactor, clientFactory, laddr):
- t = threading.Thread(target=reactor.process_requests)
- t.setDaemon(True)
- t.start()
-
def _accept(listener, client):
client.setInbox(queue)
diff --git a/vdsm_api/jsonrpc/asyncoreReactor.py b/vdsm_api/jsonrpc/asyncoreReactor.py
index c486c4b..1ef6264 100644
--- a/vdsm_api/jsonrpc/asyncoreReactor.py
+++ b/vdsm_api/jsonrpc/asyncoreReactor.py
@@ -95,6 +95,9 @@
def __init__(self, client):
self._client = client
+ def connect(self):
+ self._client.connect(self._client._addr)
+
def setInbox(self, inbox):
self._client.set_inbox(inbox)
@@ -192,6 +195,10 @@
self.wakeup()
return l
+ def createClient(self, address):
+ client = AsyncoreClient(None, self, address)
+ return AsyncoreClientWrapper(client)
+
def process_requests(self):
self._isRunning = True
while self._isRunning:
diff --git a/vdsm_api/jsonrpc/client.py b/vdsm_api/jsonrpc/client.py
index 797fafe..282738a 100644
--- a/vdsm_api/jsonrpc/client.py
+++ b/vdsm_api/jsonrpc/client.py
@@ -31,12 +31,12 @@
'params': params,
'id': reqId}
- self._transport.sendMessage(json.dumps(msg, 'utf-8'), timeout=timeout)
+ self._transport.send(json.dumps(msg, 'utf-8'), timeout=timeout)
# Notifications have no repsonse
if reqId is None:
return
- resp = self._transport.recvMessage(timeout=timeout)
+ resp = self._transport.recv(timeout=timeout)
resp = json.loads(resp)
if resp.get('error') is not None:
raise JsonRpcError(resp['error']['code'],
@@ -46,35 +46,6 @@
def close(self):
self._transport.close()
-
-
-class TCPReactorClient(object):
- def __init__(self, address):
- self.sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
- self.address = address
-
- def connect(self):
- self.sock.connect(self.address)
-
- def sendMessage(self, msg, timeout=None):
- msg = _Size.pack(len(msg)) + msg
- self.sock.settimeout(timeout)
- while msg:
- sent = self.sock.send(msg)
- msg = msg[sent:]
-
- def recvMessage(self, timeout=None):
- self.sock.settimeout(timeout)
- rawSize = self.sock.recv(_Size.size)
- size = _Size.unpack(rawSize)[0]
- buff = ""
- while (size - len(buff)) > 0:
- buff += self.sock.recv(size)
-
- return buff
-
- def close(self):
- self.sock.close()
class ProtonReactorClient(object):
@@ -87,7 +58,7 @@
def connect(self):
self._msngr.start()
- def sendMessage(self, data, timeout=None):
+ def send(self, data, timeout=None):
if timeout is None:
timeout = -1
else:
@@ -104,7 +75,7 @@
self._msngr.settle(t)
raise
- def recvMessage(self, timeout=None):
+ def recv(self, timeout=None):
if timeout is None:
timeout = -1
else:
--
To view, visit http://gerrit.ovirt.org/11625
To unsubscribe, visit http://gerrit.ovirt.org/settings
Gerrit-MessageType: newchange
Gerrit-Change-Id: I6f98075c73000c1a1bea7228eb0590c0678ea900
Gerrit-PatchSet: 1
Gerrit-Project: vdsm
Gerrit-Branch: master
Gerrit-Owner: Saggi Mizrahi <smizrahi at redhat.com>
More information about the vdsm-patches
mailing list