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