tests and default

Author: golanha

hkube_python_wrapper/config/config.py

@@ -31,7 +31,7 @@ def getBoolEnv(name, defaultValue):
     "num_ping_threads": getIntEnv('DISCOVERY_SERVER_NUM_PING_THREADS', 5),
     "servingReportInterval": getIntEnv('DISCOVERY_SERVING_REPORT_INTERVAL', 5000),
     "streaming": {
-        "messagesMemoryBuff": getIntEnv('STREAMING_MAX_BUFFER_MB', 1500),
+        "messagesMemoryBuff": getIntEnv('STREAMING_MAX_BUFFER_MB', 10),
         "port": os.environ.get('STREAMING_DISCOVERY_PORT', 9022),
         "statisticsInterval": os.environ.get('STREAMING_STATISTICS_INTERVAL', 2),
         "stateful": getBoolEnv('STREAMING_STATEFUL', 'True')

tests/test_streaming.py

@@ -4,10 +4,11 @@
 import time
 
 parsedFlows = {
-    'analyze': [{'source': 'A', 'next': ['B']}, {'source': 'B', 'next': ['C']}, {'source': 'C', 'next': ['D']}], 
+    'analyze': [{'source': 'A', 'next': ['B']}, {'source': 'B', 'next': ['C']}, {'source': 'C', 'next': ['D']}],
     'master': [{'source': 'B', 'next': ['A', 'C']}, {'source': 'C', 'next': ['D']}, {'source': 'D', 'next': ['E']}]
 }
 
+
 def test_streaming_flow():
     messages = []
 
@@ -21,7 +22,7 @@ def statsInvoked(args):
     nodeName = 'B'
     parents = [{'nodeName': 'A', 'address': {'host': '127.0.0.1', 'port': port}, 'type': 'Add'}]
     producer_config = {'port': port, 'messagesMemoryBuff': 5000, 'encoding': 'msgpack', 'statisticsInterval': 1}
-    listen_config = {'encoding': 'msgpack'}
+    listen_config = {'encoding': 'msgpack', 'delay': 10}
     streamingManager = StreamingManager()
     streamingManager.setParsedFlows(parsedFlows, 'analyze')
     streamingManager.setupStreamingProducer(statsInvoked, producer_config, [nodeName], 'A')
@@ -43,6 +44,7 @@ def statsInvoked(args):
     assert messages[3] == {'msg': '4'}
     streamingManager.stopStreaming()
 
+
 def test_streaming_manager():
     resultsAtB = {}
     resultsAtC = {}
@@ -65,7 +67,7 @@ def statsInvoked(args):
     parents2 = [{'nodeName': 'B', 'address': {'host': '127.0.0.1', 'port': port2}, 'type': 'Add'}]
     producer_configA = {'port': port1, 'messagesMemoryBuff': 5000, 'encoding': 'msgpack', 'statisticsInterval': 1}
     producer_configB = {'port': port2, 'messagesMemoryBuff': 5000, 'encoding': 'msgpack', 'statisticsInterval': 1}
-    listen_config = {'encoding': 'msgpack'}
+    listen_config = {'encoding': 'msgpack', 'delay': 10}
 
     streamingManagerA = StreamingManager()
     streamingManagerA.setParsedFlows(parsedFlows, 'analyze')
@@ -92,8 +94,8 @@ def statsInvoked(args):
     streamingManagerA.stopStreaming()
     streamingManagerB.stopStreaming()
 
-def xtest_messaging():
 
+def xtest_messaging():
     producer_config = {'port': 9536, 'messagesMemoryBuff': 5000, 'encoding': 'msgpack', 'statisticsInterval': 1}
     listenr_config = {'remoteAddress': 'tcp://localhost:9536', 'encoding': 'msgpack', 'messageOriginNodeName': 'b'}
     asserts = {}
@@ -132,7 +134,7 @@ def onMessage(envelope, msg, origin):
     assert asserts['stats'][0]['sent'] == 0
     messageListener = MessageListener(listenr_config, receiverNode='a')
     messageListener.registerMessageListener(onMessage)
-    for _ in range(1,5):
+    for _ in range(1, 5):
         if (asserts.get('field1')):
             break
         time.sleep(1)
@@ -154,12 +156,13 @@ def getHello(envelope, msg, origin):
     assert len(asserts['envelope']) == 1
     assert asserts['envelope'][0]['source'] == 'a'
 
+
 def xtest_messaging_split():
     producer_config = {'port': 9536, 'messagesMemoryBuff': 5000, 'encoding': 'msgpack', 'statisticsInterval': 1}
     listenr_config_c = {'remoteAddress': 'tcp://localhost:9536', 'encoding': 'msgpack', 'messageOriginNodeName': 'c'}
     asserts = {}
     asserts['responses'] = 0
-    messageProducer = MessageProducer(producer_config, ['a','b'], 'c')
+    messageProducer = MessageProducer(producer_config, ['a', 'b'], 'c')
 
     def onStatistics(statistics):
         asserts['stats'] = statistics
@@ -199,8 +202,8 @@ def onMessage(envelope, msg, origin):
 
     messageListener = MessageListener(listenr_config_c, receiverNode='a')
     messageListener.registerMessageListener(onMessage)
-    for _ in range(1,5):
-        if ( asserts.get('field1')):
+    for _ in range(1, 5):
+        if (asserts.get('field1')):
             break
         time.sleep(1)
     time.sleep(2)