Fix tests (#63)

* Fix tests.

* Revert parameter.

Author: doru1004

rayvens/__init__.py

@@ -14,7 +14,7 @@
 # limitations under the License.
 #
 
-from rayvens.api import init, Stream
+from rayvens.api import init, meta, Stream
 from rayvens.core.common import OutputEvent
 
-__all__ = ['init', 'Stream', 'OutputEvent']
+__all__ = ['init', 'meta', 'Stream', 'OutputEvent']

rayvens/api.py

@@ -268,8 +268,22 @@ def flush_batch(self):
         self.context.latest_sent_event_timestamp = time.time()
         self.context.event_counter += 1
 
-    def _meta(self, action, *args, **kwargs):
-        return verify_do(self, _global_camel, action, *args, **kwargs)
+    def _get_integration(self, source_sink_name):
+        if source_sink_name in self._sinks:
+            integration = self._sinks[source_sink_name]
+        if source_sink_name in self._sources:
+            integration = self._sources[source_sink_name]
+        if integration is None:
+            raise RuntimeError(
+                f'{source_sink_name} not found on stream {self.name}')
+        return integration
+
+    def _integration_invoke(self, source_sink_name, message):
+        integration = self._get_integration(source_sink_name)
+        return integration.invocation.invoke(message)
+
+    def _get_integration_name(self, source_sink_name):
+        return self._get_integration(source_sink_name).integration_name
 
     def _get_latest_timestamp(self):
         return self.context.latest_sent_event_timestamp
@@ -341,3 +355,7 @@ def init(mode=os.getenv('RAYVENS_MODE', 'auto'),
                 f'{transport} transport unsupported for mode {mode}.')
     else:
         raise RuntimeError(f'Unsupported mode {mode}.')
+
+
+def meta(stream, action, *args, **kwargs):
+    return verify_do(stream.actor, _global_camel, action, *args, **kwargs)

rayvens/core/verify.py

@@ -14,57 +14,56 @@
 # limitations under the License.
 #
 
+import ray
 import time
 from rayvens.core import kamel
 
 
-def verify_do(stream, _global_camel, action, *args, **kwargs):
+def verify_do(handle, _global_camel, action, *args, **kwargs):
     if action == 'verify_log':
-        return _verify_log(stream, _global_camel, *args, **kwargs)
+        return _verify_log(handle, _global_camel, *args, **kwargs)
     raise RuntimeError('invalid meta action')
 
 
-def _verify_log(stream,
+def wait_for_event(handle):
+    event_count = 0
+    countdown = 20
+    while event_count == 0:
+        event_count = ray.get(handle.event_count.remote())
+        time.sleep(1)
+        countdown -= 1
+        if countdown == 0:
+            break
+    if event_count == 0:
+        return False
+    return True
+
+
+def _verify_log(handle,
                 _global_camel,
                 sink_source_name,
                 message,
                 wait_for_events=False):
-    # Get integration:
-    integration = None
-    if sink_source_name in stream._sinks:
-        integration = stream._sinks[sink_source_name]
-    if sink_source_name in stream._sources:
-        integration = stream._sources[sink_source_name]
-    if integration is None:
-        raise RuntimeError(
-            f'{sink_source_name} not found on stream {stream.name}')
-
     log = "FAIL"
 
     # Wait for at least one event to happen.
     if wait_for_events:
-        event_count = 0
-        countdown = 20
-        while event_count == 0:
-            event_count = stream.event_count()
-            time.sleep(1)
-            countdown -= 1
-            if countdown == 0:
-                break
-        if event_count == 0:
-            print("[LOG CHECK]:", log)
+        if not wait_for_event(handle):
+            print("[LOG CHECK]:", "NO EVENTS RECEIVED")
             return False
 
     if _global_camel.mode.is_local():
         # In the local case the integration run is ongoing and we can
         # access the logs directly.
-        outcome = integration.invocation.invoke(message)
+        outcome = ray.get(
+            handle._integration_invoke.remote(sink_source_name, message))
     else:
         # When running using the operator then the integration run command
         # is non-blocking and returns immediately. The logs can be queried
         # using the kamel log command.
-        invocation = kamel.log(_global_camel.mode,
-                               integration.integration_name, message)
+        integration_name = ray.get(
+            handle._get_integration_name.remote(sink_source_name))
+        invocation = kamel.log(_global_camel.mode, integration_name, message)
         outcome = invocation is not None
         invocation.kill()
 

tests/generic_sink.py

@@ -59,7 +59,7 @@
 stream << output_message
 
 # Verify outcome.
-stream._meta('verify_log', sink, output_message, wait_for_events=True)
+rayvens.meta(stream, 'verify_log', sink, output_message, wait_for_events=True)
 
 # Delete all integrations from stream.
 stream.disconnect_all(after_idle_for=5)

tests/kafka_scaling_transport.py

@@ -31,7 +31,7 @@
     ray.init(address='auto')
 
 # The Kafka topic used for communication.
-topic = "testTopic"
+topic = "testTopicPartitioned"
 
 rayvens.init(mode=run_mode, transport='kafka')
 
@@ -53,11 +53,13 @@
     kafka_transport_partitions=3)
 source = source_stream.add_source(source_config)
 
-# Verify outcome.
-source_stream._meta('verify_log',
-                    test_sink,
-                    "quoteResponse",
-                    wait_for_events=True)
+# Verify outcome. Since events are going through the Kafka infrastructure
+# we need to disable checks based on event counts.
+rayvens.meta(source_stream,
+             'verify_log',
+             test_sink,
+             "quoteResponse",
+             wait_for_events=False)
 
 # Disconnect source and sink.
 source_stream.disconnect_all(after=10)

tests/kafka_transport.py

@@ -52,10 +52,11 @@
 source = source_stream.add_source(source_config)
 
 # Verify outcome.
-source_stream._meta('verify_log',
-                    test_sink,
-                    "quoteResponse",
-                    wait_for_events=True)
+rayvens.meta(source_stream,
+             'verify_log',
+             test_sink,
+             "quoteResponse",
+             wait_for_events=True)
 
 # Disconnect source and sink.
 source_stream.disconnect_all(after=5)

tests/sink.py

@@ -52,7 +52,7 @@
 stream << output_message
 
 # Verify outcome.
-stream._meta('verify_log', sink, output_message, wait_for_events=True)
+rayvens.meta(stream, 'verify_log', sink, output_message, wait_for_events=True)
 
 # Delete all integrations from stream.
 stream.disconnect_all(after_idle_for=5)