1# -*- coding: utf-8 -*- 2# (c) 2018-2020 The mqttwarn developers 3import io 4import os 5import json 6 7from builtins import str 8import logging 9 10import pytest 11 12from mqttwarn.core import make_service, decode_payload 13from tests import configfile, configfile_v2 14from tests.util import core_bootstrap, send_message 15 16 17def test_make_service(): 18 service = make_service(name='foo') 19 assert '<mqttwarn.core.Service object at' in str(service) 20 21 22def test_bootstrap(caplog): 23 """ 24 Bootstrap the core machinery without MQTT 25 """ 26 27 with caplog.at_level(logging.DEBUG): 28 29 # Bootstrap the core machinery without MQTT 30 core_bootstrap(configfile=configfile) 31 32 # Proof that mqttwarn loaded all services properly 33 assert 'Successfully loaded service "file"' in caplog.text, caplog.text 34 assert 'Successfully loaded service "log"' in caplog.text, caplog.text 35 36 assert 'Starting 1 worker threads' in caplog.text, caplog.text 37 38 # Capturing the last message does not work. Why? 39 #assert 'Job queue has 0 items to process' in caplog.text, caplog.text 40 41 42def test_decode_payload_foo(caplog): 43 44 with caplog.at_level(logging.DEBUG): 45 46 # Bootstrap the core machinery without MQTT. 47 core_bootstrap(configfile=configfile) 48 49 # Proof that decoding an unconfigured thing yields nothing sensible. 50 outcome = decode_payload(section='foo', topic='bar', payload='baz') 51 assert outcome['topic'] == 'bar' 52 assert outcome['payload'] == 'baz' 53 assert 'Cannot decode JSON object, payload=baz' in caplog.text, caplog.text 54 55 56def test_decode_payload_json(caplog): 57 58 with caplog.at_level(logging.DEBUG): 59 60 # Bootstrap the core machinery without MQTT. 61 core_bootstrap(configfile=configfile) 62 63 # Proof that decoding a valid JSON payload decodes it appropriately. 64 outcome = decode_payload(section='foo', topic='bar', payload='{"baz": "qux"}') 65 assert outcome['topic'] == 'bar' 66 assert outcome['payload'] == '{"baz": "qux"}' 67 assert outcome['baz'] == 'qux' 68 69 70def test_decode_payload_datamap(caplog): 71 72 with caplog.at_level(logging.DEBUG): 73 74 # Bootstrap the core machinery without MQTT. 75 core_bootstrap(configfile=configfile) 76 77 # Proof that decoding a valid JSON payload decodes it appropriately. 78 outcome = decode_payload(section='test/datamap', topic='bar', payload='{"baz": "qux"}') 79 assert outcome['topic'] == 'bar' 80 assert outcome['baz'] == 'qux' 81 assert outcome['datamap-key'] == 'datamap-value' 82 83 84def test_decode_payload_alldata(caplog): 85 86 with caplog.at_level(logging.DEBUG): 87 88 # Bootstrap the core machinery without MQTT. 89 core_bootstrap(configfile=configfile) 90 91 # Proof that decoding a valid JSON payload decodes it appropriately. 92 outcome = decode_payload(section='test/alldata', topic='bar', payload='{"baz": "qux"}') 93 assert outcome['topic'] == 'bar' 94 assert outcome['baz'] == 'qux' 95 assert outcome['alldata-key'] == 'alldata-value' 96 97 98def test_message_log(caplog): 99 """ 100 Submit a message to the "log" plugin and proof 101 everything gets dispatched properly. 102 """ 103 104 with caplog.at_level(logging.DEBUG): 105 106 # Bootstrap the core machinery without MQTT 107 core_bootstrap(configfile=configfile) 108 109 # Signal mocked MQTT message to the core machinery for processing 110 send_message(topic='test/log-1', payload='{"name": "temperature", "value": 42.42}') 111 112 # Proof that the message has been routed to the "log" plugin properly 113 assert "temperature: 42.42" in caplog.text, caplog.text 114 115 116def test_message_file(): 117 """ 118 Submit a message to the "file" plugin and proof 119 everything gets dispatched properly. 120 """ 121 122 data = { 123 'name': 'temperature', 124 'value': 42.42, 125 } 126 127 outputfile = '/tmp/mqttwarn-test.01' 128 if os.path.exists(outputfile): 129 os.unlink(outputfile) 130 131 # Bootstrap the core machinery without MQTT. 132 core_bootstrap(configfile=configfile) 133 134 # Signal mocked MQTT message to the core machinery for processing. 135 send_message(topic='test/file-1', payload=json.dumps(data)) 136 137 # Proof that the message has been written to the designated file properly. 138 with open(outputfile) as f: 139 content = f.read() 140 assert "temperature: 42.42" in content, content 141 142 143def test_message_file_unicode(): 144 """ 145 Submit a message to the "file" plugin and proof 146 everything gets dispatched properly. 147 148 This time, we use special characters (umlauts) 149 to proof charset encoding is also handled properly. 150 """ 151 152 data = { 153 'item': 'Räuber Hotzenplotz' 154 } 155 156 outputfile = '/tmp/mqttwarn-test.02' 157 if os.path.exists(outputfile): 158 os.unlink(outputfile) 159 160 # Bootstrap the core machinery without MQTT. 161 core_bootstrap(configfile=configfile) 162 163 # Signal mocked MQTT message to the core machinery for processing. 164 send_message(topic='test/file-2', payload=json.dumps(data)) 165 166 # Proof that the message has been written to the designated file properly. 167 with io.open(outputfile, mode='rt', encoding='utf-8') as f: 168 content = f.read() 169 assert u'Räuber Hotzenplotz' in content, content 170 171 172@pytest.mark.parametrize("configfile", [configfile, configfile_v2]) 173def test_plugin_module(caplog, configfile): 174 """ 175 Check if loading a service module with dotted name works. 176 """ 177 178 with caplog.at_level(logging.DEBUG): 179 180 # Bootstrap the core machinery without MQTT 181 core_bootstrap(configfile=configfile) 182 183 # Signal mocked MQTT message to the core machinery for processing 184 send_message(topic='test/plugin-module', payload='{"name": "temperature", "value": 42.42}') 185 186 # Proof that the message has been routed to the "log" plugin properly 187 assert 'Plugin invoked' in caplog.text, caplog.text 188 189 190@pytest.mark.parametrize("configfile", [configfile, configfile_v2]) 191def test_plugin_file(caplog, configfile): 192 """ 193 Check if loading a service module from a file works. 194 """ 195 196 with caplog.at_level(logging.DEBUG): 197 198 # Bootstrap the core machinery without MQTT 199 core_bootstrap(configfile=configfile) 200 201 # Signal mocked MQTT message to the core machinery for processing 202 send_message(topic='test/plugin-file', payload='{"name": "temperature", "value": 42.42}') 203 204 # Proof that the message has been routed to the "log" plugin properly 205 assert 'Plugin invoked' in caplog.text, caplog.text 206 207 208def test_xform_func(caplog): 209 """ 210 Submit a message to the "log" plugin and proof 211 everything gets dispatched properly. 212 213 This time, it validates the "xform" function in the context of invoking 214 a user-defined function defined through the "format" setting. 215 """ 216 with caplog.at_level(logging.DEBUG): 217 218 # Bootstrap the core machinery without MQTT 219 core_bootstrap(configfile=configfile) 220 221 # Signal mocked MQTT message to the core machinery for processing 222 send_message(topic='test/log-2', payload='{"name": "temperature", "value": 42.42}') 223 224 # Proof that the message has been routed to the "log" plugin properly 225 assert "'value': 42.42" in caplog.text, caplog.text 226 assert "'datamap-key': 'datamap-value'" in caplog.text, caplog.text 227