add Integration Tests using mosquitto as MQTT broker

.drone.yml

@@ -2,7 +2,21 @@
 kind: pipeline
 name: default
 
+services:
+- name: broker
+  image: eclipse-mosquitto
+
 steps:
+- name: run tests
+  image: alpine:latest
+  commands:
+  - apk add --no-cache python3
+  - cp -r /drone/src /tmp/src
+  - pip3 install --no-cache-dir -r /tmp/src/requirements.txt
+  - pip3 install --no-cache-dir behave
+  - mv /tmp/src/config.tests.toml /tmp/src/config.toml
+  - cd /tmp/src && python3 -m behave
+
 - name: build docker image
   image: plugins/docker
   settings:
@@ -18,6 +32,6 @@ steps:
       - pull_request
 ---
 kind: signature
-hmac: dca1de31b29a4f2d489969ddd585c5eb52d1dfeea4fddaa21e80872625ee61b0
+hmac: 08431f201762d04937c91712bd562f1ac689ddab046de1cc06f74fdfbd3f2853
 
 ...

.gitignore

@@ -1 +1,2 @@
 config.toml
+__pycache__/

config.tests.toml

@@ -0,0 +1,37 @@
+# Copy this file to config.toml to use it.
+
+[exporter]
+listen = "0.0.0.0"
+port   = 9466
+
+# Configure your MQTT broker's address and credentials here.
+[mqtt]
+client_id = "mqtt_pushgw-%(hostname)s"
+broker    = "broker"
+port      = 1883
+username  = ""
+password  = ""
+
+# List some keys we should subscribe to. Use '#' to subscribe to *everything*.
+# We will ignore any messages that can't be directly parsed as floats.
+subscribe = [
+    'sensor/#',
+    "otherTopic"
+]
+
+# Expire metrics after 5 Minutes. Can be overridden in the [[topic]] sections.
+# Comment to disable.
+expire = 300
+
+
+[[topic]]
+match  = 'sensor/(?P<sensor_name>\w+)/(?P<__metric__>\w+)'
+
+[[topic]]
+match  = 'sensor/slow_sensor/super_slow_metric'
+expire = 900
+metric = "just_a_metric"
+
+[[topic]]
+match  = 'sensor/sucky_sensor/.*'
+ignore = true

features/environment.py

@@ -0,0 +1,12 @@
+from time import time
+
+import paho.mqtt.client as mqtt
+
+def before_all(context):
+    client = mqtt.Client(client_id="IT")
+    client.connect("broker", 1883)
+
+    def publish(topic, payload):
+        client.publish(topic, payload)
+
+    context.publish = publish

features/mqtt.feature

@@ -0,0 +1,26 @@
+Feature: MQTT stuff.
+
+  Scenario: Float values
+
+    Given app is running
+     when Topic sensor/bathroom/temperature receives message with payload "15.2"
+     then Metric 'temperature{sensor_name="bathroom",mqtt_topic="sensor/bathroom/temperature"}' exists
+      and its value is equal to 15.2
+      and Metric 'mqtt_data_age{sensor_name="bathroom",mqtt_topic="sensor/bathroom/temperature",metric="temperature"}' exists
+      and its value is less than 0.5
+
+  Scenario: String values
+
+    Given app is running
+     when Topic sensor/bathroom/window receives message with payload "OPEN"
+     then Metric 'window{sensor_name="bathroom",mqtt_topic="sensor/bathroom/window",window="OPEN"}' exists
+      and its value is equal to 1.0
+      and Metric 'mqtt_data_age{sensor_name="bathroom",mqtt_topic="sensor/bathroom/window",metric="window"}' exists
+      and its value is less than 0.5
+     when Topic sensor/bathroom/window receives message with payload "CLOSED"
+     then Metric 'window{sensor_name="bathroom",mqtt_topic="sensor/bathroom/window",window="OPEN"}' exists
+      and its value is equal to 0.0
+      and Metric 'window{sensor_name="bathroom",mqtt_topic="sensor/bathroom/window",window="CLOSED"}' exists
+      and its value is equal to 1.0
+      and Metric 'mqtt_data_age{sensor_name="bathroom",mqtt_topic="sensor/bathroom/window",metric="window"}' exists
+      and its value is less than 0.5

features/steps/mqtt_steps.py

@@ -0,0 +1,62 @@
+# encoding: utf-8
+
+import threading
+
+from time import sleep
+from behave import given, when, then
+
+import mqtt_pushgateway
+
+def get_metrics():
+    prom_data = mqtt_pushgateway.http_metrics().data.decode("UTF-8")
+
+    def floatify(line):
+        key, val = line.split()
+        return key, float(val)
+
+    return dict([
+        floatify(line)
+        for line in prom_data.split("\n")
+        if line
+    ])
+
+@given('App is running')
+def step(context):
+    context.mqtt_pushgateway = threading.Thread(target=mqtt_pushgateway.main)
+    context.mqtt_pushgateway.daemon = True
+    context.mqtt_pushgateway.start()
+    context.dirty = False
+    context.metrics = {}
+    sleep(0.1)
+
+@when('Topic {topic} receives message with payload "{payload}"')
+def step(context, topic, payload):
+    context.dirty = True
+    context.publish(topic, payload.encode("UTF-8"))
+
+@then("Metric '{metric}' exists")
+def step(context, metric):
+    # Refresh context.metrics if necessary
+    if context.dirty or not context.metrics:
+        sleep(0.1)
+        context.metrics = get_metrics()
+
+    if metric not in context.metrics:
+        print("####\nMetric does not exist:\n   %s\nThese metrics do exist:\n%s\n####\n" % (
+            metric, "\n".join([ " * %s" % m for m in context.metrics])
+        ))
+        assert metric in context.metrics, "metric does not exist, see stdout for details"
+
+    context.metric_value = context.metrics[metric]
+
+@then("its value is equal to {value:f}")
+def step(context, value):
+    assert context.metric_value == value, "Value %f is not == %f" % (context.metric_value, value)
+
+@then("its value is less than {value:f}")
+def step(context, value):
+    assert context.metric_value < value, "Value %f is not < %f" % (context.metric_value, value)
+
+@then("its value is more than {value:f}")
+def step(context, value):
+    assert context.metric_value > value, "Value %f is not > %f" % (context.metric_value, value)

mqtt_pushgateway.py

@@ -9,7 +9,7 @@
 import time
 import json
 
-import paho.mqtt.client as mqttClient
+import paho.mqtt.client as mqtt
 
 from collections import defaultdict
 from datetime    import datetime, timedelta
@@ -153,7 +153,7 @@ def on_message(client, userdata, message):
         logging.warning("Metric update for '%s' failed" % topic, exc_info=True)
 
 def main():
-    client = mqttClient.Client(config["mqtt"]["client_id"] % dict(
+    client = mqtt.Client(config["mqtt"]["client_id"] % dict(
         hostname=socket.gethostname()
     ))
     client.username_pw_set(config["mqtt"]["username"], config["mqtt"]["password"])