Repository: vert-x3/vertx-eventbus-bridge-clients (language: python)

Issue:
[Issue-39] Python improvement
Motivation: Fixes #39 

Explain here the context, and why you're making that change, what is the problem you're trying to solve.

Conformance:

Your commits should be signed and you should have signed the Eclipse Contributor Agreement as explained in https://github.com/eclipse/vert.x/blob/master/CONTRIBUTING.md
Please also make sure you adhere to the code style guidelines: https://github.com/vert-x3/wiki/wiki/Vert.x-code-style-guidelines


This PR tries to improve Python client of eventbus client by:
* renamed to follow PEP conventions
* removed unnecessary DelveryOption, it can be defined as headers on sending messages to bridge
* added integration tests via project https://github.com/gaol/test-eventbus-bridge
* re connect(5 times by default) if current connection is broken
* introduce logging module for the logging messages
* REAEME update


Discussion / hints:
@punkstarman I updated according to your feedback, thank you very much on the reviews. 

The update includes:
* review feedback on README and logging messages
* make the EventBus context manager
* format the code using [Black](https://pypi.org/project/black/)
* leave the test codes unchanged, focusing on implementation itself. current test codes actually test nothing IMO :)
* checking reply timeout when specifying reply_handler on send method
* removed uneeded DeliveryOption

The update does not include:
* test codes against new implementation

Hope it can make the review easier, :) once this gets approved, I will send another PR to improve the tests. :)
Thank you @punkstarman for the review. I am merging this and start with the tests part :)

Fix patch:
diff --git a/python/build.sh b/python/build.sh
index 424cdad..dd7f1b5 100755
--- a/python/build.sh
+++ b/python/build.sh
@@ -26,6 +26,10 @@ echo -e "Build and test the project"
 echo -e "Run unit tests"
 python -m unittest -v test/unittesting/*.py
 
+echo -e "Run system tests"
+[ "$(pip show requests)" ] || pip install requests
+python -m unittest -v test/systemtesting/*.py
+
 if [ $dist -eq 1 -o $publish -eq 1 ]; then
   echo -e "Package the project"
   python setup.py sdist


Test patch:
diff --git a/python/test/systemtesting/__init__.py b/python/test/systemtesting/__init__.py
index e69de29..567dc18 100644
--- a/python/test/systemtesting/__init__.py
+++ b/python/test/systemtesting/__init__.py
@@ -0,0 +1,3 @@
+from .testutils import EventBusBridgeStarter, CountDownLatch
+
+__all__ = ['EventBusBridgeStarter', 'CountDownLatch']
diff --git a/python/test/systemtesting/server/pom.xml b/python/test/systemtesting/server/pom.xml
deleted file mode 100644
index c06f3c5..0000000
--- a/python/test/systemtesting/server/pom.xml
+++ /dev/null
@@ -1,57 +0,0 @@
-<project xmlns="http://maven.apache.org/POM/4.0.0"
-         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
-         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0
-                      http://maven.apache.org/xsd/maven-4.0.0.xsd">
-  <modelVersion>4.0.0</modelVersion>
-  <groupId>io.vertx.blog</groupId>
-  <artifactId>Server</artifactId>
-  <version>1.0-SNAPSHOT</version>
-
-  <dependencies>
-    <dependency>
-      <groupId>io.vertx</groupId>
-      <artifactId>vertx-tcp-eventbus-bridge</artifactId>
-      <version>3.3.0</version>
-    </dependency>
-  </dependencies>
-
-  <build>
-    <plugins>
-      <plugin>
-        <artifactId>maven-compiler-plugin</artifactId>
-        <version>3.3</version>
-        <configuration>
-          <source>1.8</source>
-          <target>1.8</target>
-        </configuration>
-      </plugin>
-
-      <plugin>
-        <groupId>org.apache.maven.plugins</groupId>
-        <artifactId>maven-shade-plugin</artifactId>
-        <version>2.3</version>
-        <executions>
-          <execution>
-            <phase>package</phase>
-            <goals>
-              <goal>shade</goal>
-            </goals>
-            <configuration>
-              <transformers>
-                <transformer implementation="org.apache.maven.plugins.shade.resource.ManifestResourceTransformer">
-                  <manifestEntries>
-                    <Main-Class>io.vertx.core.Starter</Main-Class>
-                    <Main-Verticle>io.first.Server</Main-Verticle>
-                  </manifestEntries>
-                </transformer>
-              </transformers>
-              <artifactSet/>
-              <outputFile>${project.build.directory}/${project.artifactId}-${project.version}-fat.jar</outputFile>
-            </configuration>
-          </execution>
-        </executions>
-      </plugin>
-    </plugins>
-  </build>
-
-</project>
diff --git a/python/test/systemtesting/server/src/main/java/io/first/Server.java b/python/test/systemtesting/server/src/main/java/io/first/Server.java
deleted file mode 100644
index 2260e89..0000000
--- a/python/test/systemtesting/server/src/main/java/io/first/Server.java
+++ /dev/null
@@ -1,44 +0,0 @@
-package io.first;
-
-import io.vertx.core.*;
-import io.vertx.core.Vertx;
-import io.vertx.core.AbstractVerticle;
-import io.vertx.core.Future;
-import io.vertx.core.eventbus.Message;
-import io.vertx.core.json.JsonObject;
-import io.vertx.core.Handler;
-import io.vertx.ext.bridge.BridgeOptions;
-import io.vertx.ext.bridge.PermittedOptions;
-import io.vertx.ext.eventbus.bridge.tcp.TcpEventBusBridge;
-import io.vertx.core.eventbus.MessageConsumer;
-import io.vertx.core.eventbus.EventBus;
-/**
- *
- * @author jay
- */
-public class Server extends AbstractVerticle{
-
-  public void start(Future<Void> fut){
-
-	TcpEventBusBridge bridge = TcpEventBusBridge.create(
-    vertx,
-    new BridgeOptions()
-        .addInboundPermitted(new PermittedOptions().setAddress("echo"))
-        .addOutboundPermitted(new PermittedOptions().setAddress("echo")));
-
-	bridge.listen(7001, res -> {
-		if (res.succeeded()) {
-		
-		} else {
-			System.exit(0);
-		}
-	});
-	EventBus eb = vertx.eventBus();
-	
-	MessageConsumer< JsonObject > consumer=eb.consumer("echo", message -> {
-		message.reply(message.body());
-	});
-	
-	
-  }
-}
diff --git a/python/test/systemtesting/test-client/test_publish.py b/python/test/systemtesting/test-client/test_publish.py
deleted file mode 100644
index 415be83..0000000
--- a/python/test/systemtesting/test-client/test_publish.py
+++ /dev/null
@@ -1,39 +0,0 @@
-import Vertx.Eventbus as Eventbus
-import Vertx.DeliveryOption as DeliveryOption
-import json
-import unittest
-import time
-
-#handlers (self,message)
-
-
-class Client(unittest.TestCase):
-    ('System Testing')
-    result = {'msg': 'test', }
-
-    def test_publish(self):
-        c = Client()
-        eb = Eventbus.Eventbus(self, 'localhost', 7001)
-
-        #jsonObject -body
-        body = {'msg': 'test1', }
-
-        # DeliveryOption
-        do = DeliveryOption.DeliveryOption()
-        do.addHeader('type', 'text')
-        do.addHeader('size', 'small')
-        do.addReplyAddress('echo')
-        do.setTimeInterval(5)
-
-        # publish
-        eb.publish('echo', body, do)
-
-        # publish without do
-        eb.publish('echo', body)
-
-        # close after 2 seconds
-        eb.closeConnection(2)
-
-
-if __name__ == '__main__':
-    unittest.main()
diff --git a/python/test/systemtesting/test-client/test_register.py b/python/test/systemtesting/test-client/test_register.py
deleted file mode 100644
index 49b387f..0000000
--- a/python/test/systemtesting/test-client/test_register.py
+++ /dev/null
@@ -1,36 +0,0 @@
-import Vertx.Eventbus as Eventbus
-import Vertx.DeliveryOption as DeliveryOption
-import json
-import unittest
-import time
-
-#handlers (self,message)
-
-
-class Client(unittest.TestCase):
-    ('System Testing')
-    result = {'msg': 'test', }
-
-    # Handler
-    def Handler(self, message):
-        if message != None:
-            Client.result = message['body']
-            print(Client.result)
-
-    def test_publish(self):
-        c = Client()
-        eb = Eventbus.Eventbus(self, 'localhost', 7001)
-
-        eb.registerHandler('echo', Client.Handler)
-        assert(eb.Handlers['echo'] != None)
-
-        eb.unregisterHandler('echo')
-        if eb.Handlers != None:
-            assert(1)
-
-        # close after 2 seconds
-        eb.closeConnection(2)
-
-
-if __name__ == '__main__':
-    unittest.main()
diff --git a/python/test/systemtesting/test-client/test_send.py b/python/test/systemtesting/test-client/test_send.py
deleted file mode 100644
index 562ba01..0000000
--- a/python/test/systemtesting/test-client/test_send.py
+++ /dev/null
@@ -1,57 +0,0 @@
-import Vertx.Eventbus as Eventbus
-import Vertx.DeliveryOption as DeliveryOption
-import json
-import unittest
-import time
-
-#replyHandler (self,error,message)
-#handlers (self,message)
-
-
-class Client(unittest.TestCase):
-    ('System Testing')
-    result = {'msg': 'test', }
-
-    # Handler
-    def Handler(self, message):
-        if message != None:
-            Client.result = message['body']
-
-    # replyHandler
-    def replyHandler(self, error, message):
-        pass
-
-    def test_send(self):
-        c = Client()
-        eb = Eventbus.Eventbus(c, 'localhost', 7001)
-
-        #jsonObject -body
-        body = {'msg': 'test1', }
-
-        # DeliveryOption
-        do = DeliveryOption.DeliveryOption()
-        do.addHeader('type', 'text')
-        do.addHeader('size', 'small')
-        do.addReplyAddress('echo')
-        do.setTimeInterval(5)
-
-        # register handler
-        eb.registerHandler('echo', Client.Handler)
-        # send
-        eb.send('echo', body, do)
-
-        # without deliveryoption
-        eb.send('echo', body, Client.replyHandler)
-        assert(Client.result == {'msg': 'test1', })
-
-        # without replyHandler
-        body = {'msg': 'test2', }
-
-        eb.send('echo', body, do)
-
-        # close after 1 seconds
-        eb.closeConnection(1)
-        assert(Client.result == {'msg': 'test2', })
-
-if __name__ == '__main__':
-    unittest.main()
diff --git a/python/test/systemtesting/test_eventbus.py b/python/test/systemtesting/test_eventbus.py
new file mode 100644
index 0000000..0b437f5
--- /dev/null
+++ b/python/test/systemtesting/test_eventbus.py
@@ -0,0 +1,144 @@
+import unittest
+from . import EventBusBridgeStarter, CountDownLatch
+from vertx import EventBus
+
+
+class EventBusClientTests(unittest.TestCase):
+    """
+    This is the tests against a local test eventbus bridge
+    """
+    
+    starter = None
+    
+    def __init__(self, *args, **kwargs):
+        super(EventBusClientTests, self).__init__(*args, **kwargs)
+    
+    @classmethod
+    def setUpClass(cls):
+        cls.starter = EventBusBridgeStarter(debug=True)
+        cls.starter.start()
+        cls.starter.wait_started()
+    
+    @classmethod
+    def tearDownClass(cls):
+        cls.starter.stop()
+    
+    def test_register_unregister(self):
+        latch = CountDownLatch(2)
+        ebus = EventBus(options={'connect': True})
+        
+        def list_handlers(message):
+            handler_list = message['body']['list']
+            addresses = [e['address'] for e in handler_list]
+            if "first" in message['headers'] and message['headers']['first']:
+                self.assertIn("list-handler", addresses)
+                self.assertIn("test-handler", addresses)
+                ebus.unregister_handler("test-handler")
+                self.assertIn("list-handler", ebus.handlers())
+                self.assertNotIn("test-handler", ebus.handlers())
+                ebus.send("list", reply_address="list-handler")
+                latch.count_down()
+            else:
+                self.assertIn("list-handler", addresses)
+                self.assertNotIn("test-handler", addresses)
+                ebus.close()
+                latch.count_down()
+            # check list handler again, no test-handler anymore !!
+        
+        ebus.register_handler("list-handler", list_handlers)
+        ebus.register_handler("test-handler", lambda x: print(x))
+        self.assertIn("list-handler", ebus.handlers())
+        self.assertIn("test-handler", ebus.handlers())
+        ebus.send("list", reply_address="list-handler", headers={'first': True})
+        latch.awaits(5)
+    
+    def test_send(self):
+        latch = CountDownLatch()
+        ebus = EventBus()
+        ebus.connect()
+        
+        def handler(message):
+            self.assertEqual(message['body']['hello'], 'world')
+            ebus.close()
+            latch.count_down()
+        ebus.register_handler("echo-back", handler)
+        ebus.send("echo", reply_address="echo-back", body={'hello': 'world'})
+        latch.awaits(5)
+    
+    def test_send_with_handler(self):
+        latch = CountDownLatch()
+        ebus = EventBus()
+        ebus.connect()
+        
+        def handler(message):
+            self.assertEqual(message['body']['hello'], 'world')
+            latch.count_down()
+        ebus.send("echo", body={'hello': 'world'}, reply_handler=handler)
+        latch.awaits(5)
+        ebus.close()
+    
+    def test_publish(self):
+        latch = CountDownLatch()
+        ebus = EventBus()
+        ebus.connect()
+        
+        def handler(message):
+            print("got publish messages back")
+            self.assertEqual(message['body']['hello'], 'world')
+            self.assertEqual(message['headers']['name'], 'vertx-python')
+            ebus.close()
+            latch.count_down()
+        ebus.register_handler("publish-back", handler)
+        ebus.publish("publish-back", headers={'name': 'vertx-python'}, body={'hello': 'world'})
+        latch.awaits(5)
+    
+    def test_send_auto_connect(self):
+        latch = CountDownLatch()
+        ebus = EventBus(options={'auto_connect': True})
+        
+        def handler(message):
+            self.assertEqual(message['body']['hello'], 'world')
+            ebus.close()
+            latch.count_down()
+        ebus.register_handler("echo-back", handler)
+        ebus.send("echo", reply_address="echo-back", body={'hello': 'world'})
+        latch.awaits(5)
+    
+    def test_send_no_connect(self):
+        ebus = EventBus(options={'auto_connect': False})
+        try:
+            ebus.register_handler("echo-back", lambda x: print(x))
+            self.fail("should not here")
+        except Exception as e:
+            self.assertEqual(str(e.args[0]), "socket has been closed.")
+    
+    def test_err_handler_bad_address(self):
+        latch = CountDownLatch()
+        
+        def err_handler(message):
+            self.assertEqual(message['type'], 'err')
+            self.assertEqual(message['message'], 'access_denied')
+            ebus.close()
+            latch.count_down()
+        ebus = EventBus(err_handler=err_handler)
+        ebus.send('#-Bad_Address$', body={'name': 'python'})
+        latch.awaits(5)
+    
+    def test_re_connect_with_registered_handlers(self):
+        latch = CountDownLatch()
+        
+        def local_handler(message):
+            self.assertEqual(message['type'], 'message')
+            self.assertEqual(message['body']['name'], 'python')
+            ebus.close()
+            latch.count_down()
+        ebus = EventBus()
+        # register local before getting connect
+        ebus._register_local("local-handler", local_handler)
+        ebus.connect()
+        ebus.send('local-handler', body={'name': 'python'})
+        latch.awaits(5)
+
+
+if __name__ == "__main__":
+    unittest.main()
diff --git a/python/test/systemtesting/testutils.py b/python/test/systemtesting/testutils.py
new file mode 100644
index 0000000..1943bd6
--- /dev/null
+++ b/python/test/systemtesting/testutils.py
@@ -0,0 +1,156 @@
+#!/usr/bin/python
+# Authors:
+# 2020: Wolfgang Fahl https://github.com/WolfgangFahl
+# 2021: Lin Gao https://github.com/gaol
+#
+# This test starter borrows lots from https://github.com/rc-dukes/vertx-eventbus-python, thanks to Wolfgang Fahl
+#
+
+from datetime import datetime
+from hashlib import md5
+import os
+import time
+from subprocess import Popen, PIPE
+import tempfile
+from threading import Thread, Condition
+import requests
+
+STARTER_FAIL_INDICATOR = "Failed to start the TCP EventBus Bridge"
+DEFAULT_WAIT_FOR = "Welcome to use EventBus Starter"
+JAR_URL_TEMPLATE = "https://github.com/gaol/test-eventbus-bridge/releases/download/%s/test-ebridge-%s-fat.jar"
+
+
+_FAT_JARS_ = {'1.0.0': '05cd3e187bf516db4685abb15c9bf983',
+              '1.0.1': 'a5c837430e98357c5ee5e533cf22b5e9'}
+
+
+class EventBusBridgeStarter:
+    """
+    An Vertx EventBus Bridge Starter, used to start a bridge server for integration testing
+    """
+    
+    def __init__(self, jar_version='1.0.1', port=7000, wait_for=DEFAULT_WAIT_FOR, debug=False, conf=None):
+        """
+        construct me
+
+        Args:
+           jar_version(str: the bridge jar version to use
+           port(int): the port to listen to, defaults to 7000
+           wait_for(str): the output string on stderr of the java process to wait for
+           debug(bool): True if debugging output should be shown else False - default: False
+        """
+        self.port = port
+        self.wait_for = wait_for
+        self.process = None
+        self.started = False
+        self.debug = debug
+        self.conf = conf
+        if jar_version not in _FAT_JARS_:
+            print("%s is not a known version" % jar_version)
+            exit(1)
+        self.jar_version = jar_version
+        temp_dir = tempfile.gettempdir().lower()
+        self.jar_file = "%s/test-ebridge-%s-fat.jar" % (temp_dir, jar_version)
+        self.jar_url = JAR_URL_TEMPLATE % (jar_version, jar_version)
+        self.failed = False
+    
+    def start(self):
+        try:
+            if not os.path.exists(self.jar_file):
+                if self.debug:
+                    print("Downloading bridge jar from %s" % self.jar_url)
+                req = requests.get(self.jar_url)
+                with open(self.jar_file, 'wb') as f:
+                    f.write(req.content)
+                    f.close()
+            with open(self.jar_file, 'rb') as f:
+                jar_md5 = md5(f.read()).hexdigest()
+            if _FAT_JARS_[self.jar_version] != jar_md5:
+                print("%s is not a valid test eventbus bridge jar file" % self.jar_file)
+                exit(1)
+            if self.conf is None:
+                self.process = Popen(['java', '-jar', self.jar_file], stderr=PIPE)
+            elif type(self.conf) is dict or os.path.exists(self.conf):
+                self.process = Popen(['java', '-jar', '-conf', self.conf, self.jar_file], stderr=PIPE)
+            t = Thread(target=self._handle_output)
+            t.daemon = True  # thread dies with the program
+            t.start()
+            # you need to wait for started with expected output
+        except IOError as e:
+            print(e)
+            raise e
+    
+    def wait_started(self, time_out=30.0, time_step=0.1):
+        """ wait for the java server to be started
+
+        Args:
+          time_out(float): the timeout in secs after which the wait fails with an Exception
+          time_step(float): the time step in secs in which the state should be regularly checked
+
+        :raise:
+           :exception: wait timed out
+           :exception: if bridge failed to start
+        """
+        time_left = time_out
+        while not self.started and not self.failed and time_left > 0:
+            time.sleep(time_step)
+            time_left = time_left - time_step
+        if time_left <= 0:
+            raise Exception("wait for start time_out after %.3f secs" % time_out)
+        if self.failed:
+            raise Exception("failed to start vertx eventbus bridge")
+        if self.debug:
+            print("wait for start successful after %.3f secs" % (time_out - time_left))
+    
+    def _handle_output(self):
+        """ handle the output of the java program"""
+        out = self.process.stderr
+        for bline in iter(out.readline, b''):
+            line = bline.decode('utf8')
+            if self.debug:
+                print("Java: %s" % line)
+            if self.wait_for in line:
+                self.started = True
+            if STARTER_FAIL_INDICATOR in line:
+                self.failed = True
+                self.stop()
+                break
+        out.close()
+    
+    def stop(self):
+        if self.started:
+            self.process.kill()
+            self.started = False
+
+
+class CountDownLatch:
+    """
+    CountDownLatch can be used for async testing
+    """
+    def __init__(self, count=1):
+        self.count = count
+        self.condition = Condition()
+    
+    def awaits(self, timeout=None):
+        try:
+            self.condition.acquire()
+            start = datetime.now()
+            while self.count > 0:
+                self.condition.wait(timeout / 10)  # divides each step by 10
+                if timeout is not None:
+                    spent = (datetime.now() - start).seconds
+                    if spent > timeout:
+                        raise Exception("timeout after waiting for %d seconds!" % spent)
+        finally:
+            self.condition.release()
+    
+    def count_down(self):
+        try:
+            self.condition.acquire()
+            self.count -= 1
+            self.condition.notifyAll()
+        finally:
+            self.condition.release()
+    
+    def get_count(self):
+        return self.count