osm_common/tests/test_msglocal.py

   1 import http
   2 import logging
   3 import pytest
   4 import tempfile
   5 import shutil
   6 import uuid
   7 import os
   8 import yaml
   9 import time
  10 import threading
  11
  12 from unittest.mock import MagicMock
  13 from osm_common.msgbase import MsgException
  14 from osm_common.msglocal import MsgLocal
  15
  16 __author__ = "Eduardo Sousa <eduardosousa@av.it.pt>"
  17
  18
  19 def valid_path():
  20     return tempfile.gettempdir() + '/'
  21
  22
  23 def invalid_path():
  24     return '/#tweeter/'
  25
  26
  27 @pytest.fixture
  28 def msg_local():
  29     msg = MsgLocal()
  30     yield msg
  31
  32     if msg.path and msg.path != invalid_path() and msg.path != valid_path():
  33         msg.disconnect()
  34         shutil.rmtree(msg.path)
  35
  36
  37 @pytest.fixture
  38 def msg_local_config():
  39     msg = MsgLocal()
  40     msg.connect({"path": valid_path() + str(uuid.uuid4())})
  41     yield msg
  42
  43     msg.disconnect()
  44     if msg.path != invalid_path():
  45         shutil.rmtree(msg.path)
  46
  47
  48 @pytest.fixture
  49 def msg_local_with_data():
  50     msg = MsgLocal()
  51     msg.connect({"path": valid_path() + str(uuid.uuid4())})
  52
  53     msg.write("topic1", "key1", "msg1")
  54     msg.write("topic1", "key2", "msg1")
  55     msg.write("topic2", "key1", "msg1")
  56     msg.write("topic2", "key2", "msg1")
  57     msg.write("topic1", "key1", "msg2")
  58     msg.write("topic1", "key2", "msg2")
  59     msg.write("topic2", "key1", "msg2")
  60     msg.write("topic2", "key2", "msg2")
  61     yield msg
  62
  63     msg.disconnect()
  64     if msg.path != invalid_path():
  65         shutil.rmtree(msg.path)
  66
  67
  68 def empty_exception_message():
  69     return "messaging exception "
  70
  71
  72 def test_constructor():
  73     msg = MsgLocal()
  74     assert msg.logger == logging.getLogger('msg')
  75     assert msg.path is None
  76     assert len(msg.files_read) == 0
  77     assert len(msg.files_write) == 0
  78     assert len(msg.buffer) == 0
  79
  80
  81 def test_constructor_with_logger():
  82     logger_name = 'msg_local'
  83     msg = MsgLocal(logger_name=logger_name)
  84     assert msg.logger == logging.getLogger(logger_name)
  85     assert msg.path is None
  86     assert len(msg.files_read) == 0
  87     assert len(msg.files_write) == 0
  88     assert len(msg.buffer) == 0
  89
  90
  91 @pytest.mark.parametrize("config, logger_name, path", [
  92     ({"logger_name": "msg_local", "path": valid_path()}, "msg_local", valid_path()),
  93     ({"logger_name": "msg_local", "path": valid_path()[:-1]}, "msg_local", valid_path()),
  94     ({"logger_name": "msg_local", "path": valid_path() + "test_it/"}, "msg_local", valid_path() + "test_it/"),
  95     ({"logger_name": "msg_local", "path": valid_path() + "test_it"}, "msg_local", valid_path() + "test_it/"),
  96     ({"path": valid_path()}, "msg", valid_path()),
  97     ({"path": valid_path()[:-1]}, "msg", valid_path()),
  98     ({"path": valid_path() + "test_it/"}, "msg", valid_path() + "test_it/"),
  99     ({"path": valid_path() + "test_it"}, "msg", valid_path() + "test_it/")])
 100 def test_connect(msg_local, config, logger_name, path):
 101     msg_local.connect(config)
 102     assert msg_local.logger == logging.getLogger(logger_name)
 103     assert msg_local.path == path
 104     assert len(msg_local.files_read) == 0
 105     assert len(msg_local.files_write) == 0
 106     assert len(msg_local.buffer) == 0
 107
 108
 109 @pytest.mark.parametrize("config", [
 110     ({"logger_name": "msg_local", "path": invalid_path()}),
 111     ({"path": invalid_path()})])
 112 def test_connect_with_exception(msg_local, config):
 113     with pytest.raises(MsgException) as excinfo:
 114         msg_local.connect(config)
 115     assert str(excinfo.value).startswith(empty_exception_message())
 116     assert excinfo.value.http_code == http.HTTPStatus.INTERNAL_SERVER_ERROR
 117
 118
 119 def test_disconnect(msg_local_config):
 120     msg_local_config.disconnect()
 121     for f in msg_local_config.files_read.values():
 122         assert f.closed
 123     for f in msg_local_config.files_write.values():
 124         assert f.closed
 125
 126
 127 def test_disconnect_with_read(msg_local_config):
 128     msg_local_config.read('topic1', blocks=False)
 129     msg_local_config.read('topic2', blocks=False)
 130     msg_local_config.disconnect()
 131     for f in msg_local_config.files_read.values():
 132         assert f.closed
 133     for f in msg_local_config.files_write.values():
 134         assert f.closed
 135
 136
 137 def test_disconnect_with_write(msg_local_with_data):
 138     msg_local_with_data.disconnect()
 139
 140     for f in msg_local_with_data.files_read.values():
 141         assert f.closed
 142
 143     for f in msg_local_with_data.files_write.values():
 144         assert f.closed
 145
 146
 147 def test_disconnect_with_read_and_write(msg_local_with_data):
 148     msg_local_with_data.read('topic1', blocks=False)
 149     msg_local_with_data.read('topic2', blocks=False)
 150
 151     msg_local_with_data.disconnect()
 152     for f in msg_local_with_data.files_read.values():
 153         assert f.closed
 154     for f in msg_local_with_data.files_write.values():
 155         assert f.closed
 156
 157
 158 @pytest.mark.parametrize("topic, key, msg", [
 159     ("test_topic", "test_key", "test_msg"),
 160     ("test", "test_key", "test_msg"),
 161     ("test_topic", "test", "test_msg"),
 162     ("test_topic", "test_key", "test"),
 163     ("test_topic", "test_list", ["a", "b", "c"]),
 164     ("test_topic", "test_tuple", ("c", "b", "a")),
 165     ("test_topic", "test_dict", {"a": 1, "b": 2, "c": 3}),
 166     ("test_topic", "test_number", 123),
 167     ("test_topic", "test_float", 1.23),
 168     ("test_topic", "test_boolean", True),
 169     ("test_topic", "test_none", None)])
 170 def test_write(msg_local_config, topic, key, msg):
 171     file_path = msg_local_config.path + topic
 172     msg_local_config.write(topic, key, msg)
 173     assert os.path.exists(file_path)
 174
 175     with open(file_path, 'r') as stream:
 176         assert yaml.load(stream) == {key: msg if not isinstance(msg, tuple) else list(msg)}
 177
 178
 179 @pytest.mark.parametrize("topic, key, msg, times", [
 180     ("test_topic", "test_key", "test_msg", 2),
 181     ("test", "test_key", "test_msg", 3),
 182     ("test_topic", "test", "test_msg", 4),
 183     ("test_topic", "test_key", "test", 2),
 184     ("test_topic", "test_list", ["a", "b", "c"], 3),
 185     ("test_topic", "test_tuple", ("c", "b", "a"), 4),
 186     ("test_topic", "test_dict", {"a": 1, "b": 2, "c": 3}, 2),
 187     ("test_topic", "test_number", 123, 3),
 188     ("test_topic", "test_float", 1.23, 4),
 189     ("test_topic", "test_boolean", True, 2),
 190     ("test_topic", "test_none", None, 3)])
 191 def test_write_with_multiple_calls(msg_local_config, topic, key, msg, times):
 192     file_path = msg_local_config.path + topic
 193
 194     for _ in range(times):
 195         msg_local_config.write(topic, key, msg)
 196     assert os.path.exists(file_path)
 197
 198     with open(file_path, 'r') as stream:
 199         for _ in range(times):
 200             data = stream.readline()
 201             assert yaml.load(data) == {key: msg if not isinstance(msg, tuple) else list(msg)}
 202
 203
 204 def test_write_exception(msg_local_config):
 205     msg_local_config.files_write = MagicMock()
 206     msg_local_config.files_write.__contains__.side_effect = Exception()
 207
 208     with pytest.raises(MsgException) as excinfo:
 209         msg_local_config.write("test", "test", "test")
 210     assert str(excinfo.value).startswith(empty_exception_message())
 211     assert excinfo.value.http_code == http.HTTPStatus.INTERNAL_SERVER_ERROR
 212
 213
 214 @pytest.mark.parametrize("topics, datas", [
 215     (["topic"], [{"key": "value"}]),
 216     (["topic1"], [{"key": "value"}]),
 217     (["topic2"], [{"key": "value"}]),
 218     (["topic", "topic1"], [{"key": "value"}]),
 219     (["topic", "topic2"], [{"key": "value"}]),
 220     (["topic1", "topic2"], [{"key": "value"}]),
 221     (["topic", "topic1", "topic2"], [{"key": "value"}]),
 222     (["topic"], [{"key": "value"}, {"key1": "value1"}]),
 223     (["topic1"], [{"key": "value"}, {"key1": "value1"}]),
 224     (["topic2"], [{"key": "value"}, {"key1": "value1"}]),
 225     (["topic", "topic1"], [{"key": "value"}, {"key1": "value1"}]),
 226     (["topic", "topic2"], [{"key": "value"}, {"key1": "value1"}]),
 227     (["topic1", "topic2"], [{"key": "value"}, {"key1": "value1"}]),
 228     (["topic", "topic1", "topic2"], [{"key": "value"}, {"key1": "value1"}])])
 229 def test_read(msg_local_with_data, topics, datas):
 230     def write_to_topic(topics, datas):
 231         # Allow msglocal to block while waiting
 232         time.sleep(2)
 233         for topic in topics:
 234             for data in datas:
 235                 with open(msg_local_with_data.path + topic, "a+") as fp:
 236                     yaml.safe_dump(data, fp, default_flow_style=True, width=20000)
 237                     fp.flush()
 238
 239     # If file is not opened first, the messages written won't be seen
 240     for topic in topics:
 241         if topic not in msg_local_with_data.files_read:
 242             msg_local_with_data.read(topic, blocks=False)
 243
 244     t = threading.Thread(target=write_to_topic, args=(topics, datas))
 245     t.start()
 246
 247     for topic in topics:
 248         for data in datas:
 249             recv_topic, recv_key, recv_msg = msg_local_with_data.read(topic)
 250             key = list(data.keys())[0]
 251             val = data[key]
 252             assert recv_topic == topic
 253             assert recv_key == key
 254             assert recv_msg == val
 255     t.join()
 256
 257
 258 @pytest.mark.parametrize("topics, datas", [
 259     (["topic"], [{"key": "value"}]),
 260     (["topic1"], [{"key": "value"}]),
 261     (["topic2"], [{"key": "value"}]),
 262     (["topic", "topic1"], [{"key": "value"}]),
 263     (["topic", "topic2"], [{"key": "value"}]),
 264     (["topic1", "topic2"], [{"key": "value"}]),
 265     (["topic", "topic1", "topic2"], [{"key": "value"}]),
 266     (["topic"], [{"key": "value"}, {"key1": "value1"}]),
 267     (["topic1"], [{"key": "value"}, {"key1": "value1"}]),
 268     (["topic2"], [{"key": "value"}, {"key1": "value1"}]),
 269     (["topic", "topic1"], [{"key": "value"}, {"key1": "value1"}]),
 270     (["topic", "topic2"], [{"key": "value"}, {"key1": "value1"}]),
 271     (["topic1", "topic2"], [{"key": "value"}, {"key1": "value1"}]),
 272     (["topic", "topic1", "topic2"], [{"key": "value"}, {"key1": "value1"}])])
 273 def test_read_non_block(msg_local_with_data, topics, datas):
 274     def write_to_topic(topics, datas):
 275         for topic in topics:
 276             for data in datas:
 277                 with open(msg_local_with_data.path + topic, "a+") as fp:
 278                     yaml.safe_dump(data, fp, default_flow_style=True, width=20000)
 279                     fp.flush()
 280
 281     # If file is not opened first, the messages written won't be seen
 282     for topic in topics:
 283         if topic not in msg_local_with_data.files_read:
 284             msg_local_with_data.read(topic, blocks=False)
 285
 286     t = threading.Thread(target=write_to_topic, args=(topics, datas))
 287     t.start()
 288     t.join()
 289
 290     for topic in topics:
 291         for data in datas:
 292             recv_topic, recv_key, recv_msg = msg_local_with_data.read(topic, blocks=False)
 293             key = list(data.keys())[0]
 294             val = data[key]
 295             assert recv_topic == topic
 296             assert recv_key == key
 297             assert recv_msg == val
 298
 299
 300 @pytest.mark.parametrize("topics, datas", [
 301     (["topic"], [{"key": "value"}]),
 302     (["topic1"], [{"key": "value"}]),
 303     (["topic2"], [{"key": "value"}]),
 304     (["topic", "topic1"], [{"key": "value"}]),
 305     (["topic", "topic2"], [{"key": "value"}]),
 306     (["topic1", "topic2"], [{"key": "value"}]),
 307     (["topic", "topic1", "topic2"], [{"key": "value"}]),
 308     (["topic"], [{"key": "value"}, {"key1": "value1"}]),
 309     (["topic1"], [{"key": "value"}, {"key1": "value1"}]),
 310     (["topic2"], [{"key": "value"}, {"key1": "value1"}]),
 311     (["topic", "topic1"], [{"key": "value"}, {"key1": "value1"}]),
 312     (["topic", "topic2"], [{"key": "value"}, {"key1": "value1"}]),
 313     (["topic1", "topic2"], [{"key": "value"}, {"key1": "value1"}]),
 314     (["topic", "topic1", "topic2"], [{"key": "value"}, {"key1": "value1"}])])
 315 def test_read_non_block_none(msg_local_with_data, topics, datas):
 316     def write_to_topic(topics, datas):
 317         time.sleep(2)
 318         for topic in topics:
 319             for data in datas:
 320                 with open(msg_local_with_data.path + topic, "a+") as fp:
 321                     yaml.safe_dump(data, fp, default_flow_style=True, width=20000)
 322                     fp.flush()
 323     # If file is not opened first, the messages written won't be seen
 324     for topic in topics:
 325         if topic not in msg_local_with_data.files_read:
 326             msg_local_with_data.read(topic, blocks=False)
 327     t = threading.Thread(target=write_to_topic, args=(topics, datas))
 328     t.start()
 329
 330     for topic in topics:
 331         recv_data = msg_local_with_data.read(topic, blocks=False)
 332         assert recv_data is None
 333     t.join()
 334
 335
 336 @pytest.mark.parametrize("blocks", [
 337     (True),
 338     (False)])
 339 def test_read_exception(msg_local_with_data, blocks):
 340     msg_local_with_data.files_read = MagicMock()
 341     msg_local_with_data.files_read.__contains__.side_effect = Exception()
 342
 343     with pytest.raises(MsgException) as excinfo:
 344         msg_local_with_data.read("topic1", blocks=blocks)
 345     assert str(excinfo.value).startswith(empty_exception_message())
 346     assert excinfo.value.http_code == http.HTTPStatus.INTERNAL_SERVER_ERROR
 347
 348
 349 @pytest.mark.parametrize("topics, datas", [
 350     (["topic"], [{"key": "value"}]),
 351     (["topic1"], [{"key": "value"}]),
 352     (["topic2"], [{"key": "value"}]),
 353     (["topic", "topic1"], [{"key": "value"}]),
 354     (["topic", "topic2"], [{"key": "value"}]),
 355     (["topic1", "topic2"], [{"key": "value"}]),
 356     (["topic", "topic1", "topic2"], [{"key": "value"}]),
 357     (["topic"], [{"key": "value"}, {"key1": "value1"}]),
 358     (["topic1"], [{"key": "value"}, {"key1": "value1"}]),
 359     (["topic2"], [{"key": "value"}, {"key1": "value1"}]),
 360     (["topic", "topic1"], [{"key": "value"}, {"key1": "value1"}]),
 361     (["topic", "topic2"], [{"key": "value"}, {"key1": "value1"}]),
 362     (["topic1", "topic2"], [{"key": "value"}, {"key1": "value1"}]),
 363     (["topic", "topic1", "topic2"], [{"key": "value"}, {"key1": "value1"}])])
 364 def test_aioread(msg_local_with_data, event_loop, topics, datas):
 365     def write_to_topic(topics, datas):
 366         time.sleep(2)
 367         for topic in topics:
 368             for data in datas:
 369                 with open(msg_local_with_data.path + topic, "a+") as fp:
 370                     yaml.safe_dump(data, fp, default_flow_style=True, width=20000)
 371                     fp.flush()
 372     # If file is not opened first, the messages written won't be seen
 373     for topic in topics:
 374         if topic not in msg_local_with_data.files_read:
 375             msg_local_with_data.read(topic, blocks=False)
 376
 377     t = threading.Thread(target=write_to_topic, args=(topics, datas))
 378     t.start()
 379     for topic in topics:
 380         for data in datas:
 381             recv = event_loop.run_until_complete(msg_local_with_data.aioread(topic, event_loop))
 382             recv_topic, recv_key, recv_msg = recv
 383             key = list(data.keys())[0]
 384             val = data[key]
 385             assert recv_topic == topic
 386             assert recv_key == key
 387             assert recv_msg == val
 388     t.join()
 389
 390
 391 def test_aioread_exception(msg_local_with_data, event_loop):
 392     msg_local_with_data.files_read = MagicMock()
 393     msg_local_with_data.files_read.__contains__.side_effect = Exception()
 394
 395     with pytest.raises(MsgException) as excinfo:
 396         event_loop.run_until_complete(msg_local_with_data.aioread("topic1", event_loop))
 397     assert str(excinfo.value).startswith(empty_exception_message())
 398     assert excinfo.value.http_code == http.HTTPStatus.INTERNAL_SERVER_ERROR
 399
 400
 401 def test_aioread_general_exception(msg_local_with_data, event_loop):
 402     msg_local_with_data.read = MagicMock()
 403     msg_local_with_data.read.side_effect = Exception()
 404
 405     with pytest.raises(MsgException) as excinfo:
 406         event_loop.run_until_complete(msg_local_with_data.aioread("topic1", event_loop))
 407     assert str(excinfo.value).startswith(empty_exception_message())
 408     assert excinfo.value.http_code == http.HTTPStatus.INTERNAL_SERVER_ERROR
 409
 410
 411 @pytest.mark.parametrize("topic, key, msg", [
 412     ("test_topic", "test_key", "test_msg"),
 413     ("test", "test_key", "test_msg"),
 414     ("test_topic", "test", "test_msg"),
 415     ("test_topic", "test_key", "test"),
 416     ("test_topic", "test_list", ["a", "b", "c"]),
 417     ("test_topic", "test_tuple", ("c", "b", "a")),
 418     ("test_topic", "test_dict", {"a": 1, "b": 2, "c": 3}),
 419     ("test_topic", "test_number", 123),
 420     ("test_topic", "test_float", 1.23),
 421     ("test_topic", "test_boolean", True),
 422     ("test_topic", "test_none", None)])
 423 def test_aiowrite(msg_local_config, event_loop, topic, key, msg):
 424     file_path = msg_local_config.path + topic
 425     event_loop.run_until_complete(msg_local_config.aiowrite(topic, key, msg))
 426     assert os.path.exists(file_path)
 427
 428     with open(file_path, 'r') as stream:
 429         assert yaml.load(stream) == {key: msg if not isinstance(msg, tuple) else list(msg)}
 430
 431
 432 @pytest.mark.parametrize("topic, key, msg, times", [
 433     ("test_topic", "test_key", "test_msg", 2),
 434     ("test", "test_key", "test_msg", 3),
 435     ("test_topic", "test", "test_msg", 4),
 436     ("test_topic", "test_key", "test", 2),
 437     ("test_topic", "test_list", ["a", "b", "c"], 3),
 438     ("test_topic", "test_tuple", ("c", "b", "a"), 4),
 439     ("test_topic", "test_dict", {"a": 1, "b": 2, "c": 3}, 2),
 440     ("test_topic", "test_number", 123, 3),
 441     ("test_topic", "test_float", 1.23, 4),
 442     ("test_topic", "test_boolean", True, 2),
 443     ("test_topic", "test_none", None, 3)])
 444 def test_aiowrite_with_multiple_calls(msg_local_config, event_loop, topic, key, msg, times):
 445     file_path = msg_local_config.path + topic
 446     for _ in range(times):
 447         event_loop.run_until_complete(msg_local_config.aiowrite(topic, key, msg))
 448     assert os.path.exists(file_path)
 449
 450     with open(file_path, 'r') as stream:
 451         for _ in range(times):
 452             data = stream.readline()
 453             assert yaml.load(data) == {key: msg if not isinstance(msg, tuple) else list(msg)}
 454
 455
 456 def test_aiowrite_exception(msg_local_config, event_loop):
 457     msg_local_config.files_write = MagicMock()
 458     msg_local_config.files_write.__contains__.side_effect = Exception()
 459
 460     with pytest.raises(MsgException) as excinfo:
 461         event_loop.run_until_complete(msg_local_config.aiowrite("test", "test", "test"))
 462     assert str(excinfo.value).startswith(empty_exception_message())
 463     assert excinfo.value.http_code == http.HTTPStatus.INTERNAL_SERVER_ERROR