Merge pull request #33 from TheGhouls/oct-0.4.6

Oct 0.4.6 resolves #28 and #30

.coveragerc

@@ -1,2 +1,8 @@
+[run]
+concurrency=multiprocessing
+
 [report]
-omit = oct/utilities/templates/scripts/v_user.j2
+omit =
+    oct/utilities/templates/scripts/v_user.j2
+    oct/core/devices.py
+    oct/utilities/run_device.py

.travis.yml

@@ -11,7 +11,8 @@ script:
 - python setup.py install
 - nosetests -vdx tests/ --with-coverage --cover-package=oct
 after_success:
-  coveralls
+  - coverage combine
+  - coveralls
 deploy:
   provider: pypi
   user: Emmanuel.Valette

changelog.txt

@@ -1,3 +1,20 @@
+0.4.5 to 0.4.6
+==============
+
+* allow database configuration in project configuration, you can now use any peewee supported database backend
+* update rebuild command to take database configuration in consideration
+* devices has been added to project, allowing you to easily start a zmq forwarder or streamer for more complex tests
+* `oct run` command has been reworked. Some useless and undocumented arguments has been removed and new arguments has been added :
+  * ``-o`` allowing to choose output directory for tests results
+  * ``--with-forwarder`` start HQ with external publisher forwarder, allowing you to connect to forwarder device instead of simply binding a PUB socket
+  * ``--forwarder-address`` allow you to set ip and port for external forwarder with format <ip>:<port>. If not set but ``--with-forwarder`` is present default will be used (localhost:5002). Default port is based on your configuration and use the ``external_publisher`` configuration key
+
+Backward compatibility break
+----------------------------
+
+* `oct rebuild-results` commands arguments has changed, results_file is now optionnal and the command will by default look
+at the configuration file. If no results_database key is present, it will use the patern `results_directory + results.sqlite`
+
 0.4.4 to 0.4.5
 ==============
 
@@ -165,7 +182,7 @@ Backward compatibility breaks
 as been removed. Transaction now inherit by default from the BaseTransaction class located in the oct-turrets package
 * A new command has been added `oct-pack-turrets` that will create a tar file from a test folder containing all files and configuration
 required for running a turret. This command will generate a tar archive per turret in your project `config.json` file
-* More informations are stored in the results for the turrets and the report display the last known status of a turret and the associated last update time
+## Last update Thu Jul  7 17:30:27 2016 VALETTE Emmanuel
 * You can hotplug a turret while a test is running. This turret will appear in the final report
 * The HQ will sleep for 1 second after the creation of the sockets to wait for them
 * Remove the GenericTransaction class in oct.core.generic

doc/collecting_results.rst

@@ -28,9 +28,8 @@ You can simply use the ``oct rebuild-results`` like this for example:
 
 .. code-block:: bash
 
-    oct rebuild-results . results.sqlite config.json
+    oct rebuild-results . config.json
 
 .. note::
 
-    The oct rebuild-results command will only work on an already created results folder that contains only the sqlite results
-    and optionnaly the configuration.
+    `rebuild-results` command will use database configuration in `config.json` file

doc/commands_reference.rst

@@ -128,17 +128,17 @@ usage :
 
 .. code-block:: sh
 
-    oct rebuild-results <path-to-results> <path-to-sqlite> <path-to-config>
+    oct rebuild-results <path-to-results> <path-to-config> [-f] <path-to-sqlite-results>
 
 Arguments :
 
-================ ====  ========== =============================================
-name             type  mandatory  description
-================ ====  ========== =============================================
-results_dir      str   yes        results directory to rebuild
-results_file     str   yes        sqlite result file to use
-config_file      str   yes        json config file of the project
-================ ====  ========== =============================================
+================== ====  ========== =============================================
+name               type  mandatory  description
+================== ====  ========== =============================================
+results_dir        str   yes        results directory to rebuild
+config_file        str   yes        json config file of the project
+-f, --results-file str   no         sqlite results file
+================== ====  ========== =============================================
 
 Results to csv
 ---------------

doc/first_project.rst

@@ -63,6 +63,10 @@ The main and more important file here is the `config.json` file, let's take a lo
         "turrets_requirements": [],
         "extra_turret_config": {
             // put global turrets config key / values here
+        },
+        "results_database": {
+            "db_uri": "default",
+            "params": {}
         }
     }
 
@@ -88,6 +92,8 @@ Every key here is useful, but some keys are not required to run a test. Let's ta
 
 * ``extra_turret_config``: A nested object containing all extra turrets parameters. Each value in it will be set in each turret configuration
 
+* ``results_database``: Nested object that allows results database configuration, for exemple if you don't want to use the default sqlite results database
+
 This configuration is simple but will allow you to run simple tests in a local environment.
 
 

oct/__init__.py

@@ -1 +1 @@
-__version__ = '0.4.5'
+__version__ = '0.4.6'

oct/core/devices.py

@@ -0,0 +1,57 @@
+"""Contain all zeromq devices needed by OCT
+"""
+from __future__ import print_function, unicode_literals
+import zmq
+
+
+def forwarder(frontend, backend):
+    """Simple pub/sub forwarder
+
+    :param int frontend: fontend zeromq port
+    :param int backend: backend zeromq port
+    """
+    try:
+        context = zmq.Context()
+
+        front_sub = context.socket(zmq.SUB)
+        front_sub.bind("tcp://*:%d" % frontend)
+
+        front_sub.setsockopt_string(zmq.SUBSCRIBE, "")
+
+        back_pub = context.socket(zmq.PUB)
+        back_pub.bind("tcp://*:%d" % backend)
+
+        print("forwarder started, backend on port : %d\tfrontend on port: %d" % (backend, frontend))
+        zmq.proxy(front_sub, back_pub)
+    except Exception as e:
+        print(e)
+    finally:
+        front_sub.close()
+        back_pub.close()
+        context.term()
+
+
+def streamer(frontend, backend):
+    """Simple push/pull streamer
+
+    :param int frontend: fontend zeromq port
+    :param int backend: backend zeromq port
+    """
+    try:
+        context = zmq.Context()
+
+        front_pull = context.socket(zmq.PULL)
+        front_pull.set_hwm(0)
+        front_pull.bind("tcp://*:%d" % frontend)
+
+        back_push = context.socket(zmq.PUSH)
+        back_push.bind("tcp://*:%d" % backend)
+
+        print("streamer started, backend on port : %d\tfrontend on port: %d" % (backend, frontend))
+        zmq.proxy(front_pull, back_push)
+    except Exception as e:
+        print(e)
+    finally:
+        front_pull.close()
+        back_push.close()
+        context.term()

oct/core/hq.py

@@ -13,36 +13,66 @@ class HightQuarter(object):
     """The main hight quarter that will receive informations from the turrets
     and send the start message
 
-    :param int publish_port: the port for publishing information to turrets
-    :param int rc_port: the result collector port for collecting results from the turrets
-    :param StatsHandler stats_handler: the stats handler writer
+    :param str output_dir: output directory for results
     :param dict config: the configuration of the test
+    :param str topic: topic for external publishing socket
+    :param bool with_forwarder: tell HQ if it should connects to forwarder, default False
+    :param bool with_streamer: tell HQ if ti should connects to streamer, default False
+    :param str streamer_address: streamer address to connect with form : <ip>:<port>
     """
-    def __init__(self, publish_port, rc_port, output_dir, config, topic):
+
+    def __init__(self, output_dir, config, topic, master=True, *args, **kwargs):
         self.context = zmq.Context()
         self.poller = zmq.Poller()
         self.topic = topic
+        self.master = master
 
         self.result_collector = self.context.socket(zmq.PULL)
-        self.result_collector.set_hwm(0)
-        self.result_collector.bind("tcp://*:{}".format(rc_port))
-
         self.external_publisher = self.context.socket(zmq.PUB)
-        self.external_publisher.bind("tcp://*:{}".format(config.get('external_publisher', 5002)))
-
-        self.stats_handler = StatsHandler(output_dir, config)
+        self.stats_handler = StatsHandler()
 
-        self.poller.register(self.result_collector, zmq.POLLIN)
-
-        self.turrets_manager = TurretsManager(publish_port)
+        self._configure_sockets(config)
+        self.turrets_manager = TurretsManager(config.get('publish_port', 5000), master)
         self.config = config
         self.started = False
         self.messages = 0
 
+        with_forwarder = kwargs.get('with_forwarder', False)
+        forwarder_address = None
+        if with_forwarder is True:
+            forwarder_address = kwargs.get('forwarder_address', None)
+            if forwarder_address is None:
+                forwarder_address = "127.0.0.1:{}".format(config.get('external_publisher', 5002))
+
+        self._configure_external_publisher(config, with_forwarder, forwarder_address)
+
         # waiting for init sockets
         print("Warmup")
         time.sleep(1)
 
+    def _configure_external_publisher(self, config, with_forwarder=False, forwarder_address=None):
+        external_publisher = config.get('external_publisher', 5002) if not forwarder_address else forwarder_address
+
+        print(external_publisher)
+        if with_forwarder:
+            self.external_publisher.connect("tcp://{}".format(external_publisher))
+        else:
+            self.external_publisher.bind("tcp://*:{}".format(external_publisher))
+
+    def _configure_sockets(self, config, with_streamer=False, with_forwarder=False):
+        """Configure sockets for HQ
+
+        :param dict config: test configuration
+        :param bool with_streamer: tell if we need to connect to streamer or simply bind
+        :param bool with_forwarder: tell if we need to connect to forwarder or simply bind
+        """
+        rc_port = config.get('rc_port', 5001)
+
+        self.result_collector.set_hwm(0)
+        self.result_collector.bind("tcp://*:{}".format(rc_port))
+
+        self.poller.register(self.result_collector, zmq.POLLIN)
+
     def _process_socks(self, socks):
         if self.result_collector in socks:
             data = self.result_collector.recv_string()

oct/core/turrets_manager.py

@@ -9,17 +9,22 @@
 class TurretsManager(object):
     """Turrets management while runing test. This class is in charge to send
     message to turrets and to store informations about active turrets
+
+    :param int publish_port: pub socket port
+    :param bool master: tell if current HQ is a master. Only master can send messages to turrets
     """
     STATUS_REQUEST = {'command': 'status_request', 'msg': None}
     START = {'command': 'start', 'msg': 'open fire'}
     STOP = {'command': 'stop', 'msg': 'premature stop'}
 
-    def __init__(self, publish_port):
+    def __init__(self, publish_port, master=True):
         self.turrets = {}
+        self.master = master
 
-        context = zmq.Context()
-        self.publisher = context.socket(zmq.PUB)
-        self.publisher.bind("tcp://*:{}".format(publish_port))
+        if master:
+            context = zmq.Context()
+            self.publisher = context.socket(zmq.PUB)
+            self.publisher.bind("tcp://*:{}".format(publish_port))
 
     def clean(self):
         self.publisher.close()
@@ -45,6 +50,9 @@ def process_message(self, message, is_started=False):
         :param dict message: incomming message
         :param bool is_started: test started indicator
         """
+        if not self.master:
+            return False
+
         if 'status' not in message:
             return False
         message['name'] = message['turret']
@@ -97,6 +105,8 @@ def publish(self, message, channel=None):
         :param dict message: message to send to turrets
         :pram str channel: channel to send message, default to empty string
         """
+        if not self.master:
+            return
         channel = channel or ''
         data = json.dumps(message)
         self.publisher.send_string("%s %s" % (channel, data))

oct/results/models.py

@@ -1,6 +1,9 @@
+import os
 import json
 import datetime
-from peewee import Proxy, TextField, FloatField, CharField, IntegerField, Model, DateTimeField, SqliteDatabase
+
+from playhouse.db_url import connect
+from peewee import Proxy, TextField, FloatField, CharField, IntegerField, Model, DateTimeField
 
 db = Proxy()
 
@@ -59,15 +62,22 @@ class Meta:
         database = db
 
 
-def set_database(db_path, proxy, config):
+def set_database(db_url, proxy, config):
     """Initialize the peewee database with the given configuration
 
-    :param str db_path: the path of the sqlite database
+    If the given db_url is a regular file, it will be used as sqlite database
+
+    :param str db_url: the connection string for database or path if sqlite file
     :param peewee.Proxy proxy: the peewee proxy to initialise
     :param dict config: the configuration dictionnary
     """
+    db_config = config.get('results_database', {}).get('params', {})
+
     if 'testing' in config and config['testing'] is True:
-        database = SqliteDatabase('/tmp/results.sqlite', check_same_thread=False, threadlocals=True)
+        database = connect('sqlite:////tmp/results.sqlite', check_same_thread=False, threadlocals=True)
     else:
-        database = SqliteDatabase(db_path, check_same_thread=False, threadlocals=True)
+        if os.path.isfile(db_url) or os.path.isdir(os.path.dirname(db_url)):
+            db_url = "sqlite:///" + db_url
+            db_config.update(check_same_thread=False, threadlocals=True)
+        database = connect(db_url, **db_config)
     proxy.initialize(database)

oct/results/stats_handler.py

@@ -1,6 +1,7 @@
 import os
 import ujson
 
+from oct.utilities.configuration import get_db_uri
 from oct.results.models import Result, Turret, set_database, db
 
 
@@ -12,22 +13,23 @@ def init_stats(output_dir, config):
 
     try:
         os.makedirs(output_dir, 0o755)
-    except OSError:
-        print("ERROR: Can not create output directory\n")
+    except OSError as e:
+        print("ERROR: Can not create output directory: %s\n" % e)
         raise
 
-    set_database(output_dir + "results.sqlite", db, config)
+    db_uri = get_db_uri(config, output_dir)
+
+    set_database(db_uri, db, config)
+
+    tables_to_create = [t for t in [Result, Turret] if not t.table_exists()]
+
     db.connect()
-    db.create_tables([Result, Turret])
+    db.create_tables(tables_to_create)
 
 
 class StatsHandler(object):
-    """This class will handle results and stats comming from the turrets
-    :param str output_dir: the output directory for the results
-    """
-    def __init__(self, output_dir, config, context=None):
-        self.output_dir = output_dir
-        self.turret_name = 'Turret'
+    """This class will handle results and stats comming from the turrets"""
+    def __init__(self):
         self.results = []
 
     def write_result(self, data):