__author__ = "Eduardo Sousa <eduardosousa@av.it.pt>"
+
def valid_path():
return tempfile.gettempdir() + '/'
+
def invalid_path():
return '/#tweeter/'
+
@pytest.fixture
def msg_local():
msg = MsgLocal()
-
yield msg
if msg.path and msg.path != invalid_path() and msg.path != valid_path():
msg.disconnect()
shutil.rmtree(msg.path)
+
@pytest.fixture
def msg_local_config():
msg = MsgLocal()
msg.connect({"path": valid_path() + str(uuid.uuid4())})
-
yield msg
-
+
msg.disconnect()
if msg.path != invalid_path():
shutil.rmtree(msg.path)
+
@pytest.fixture
def msg_local_with_data():
msg = MsgLocal()
msg.connect({"path": valid_path() + str(uuid.uuid4())})
-
+
msg.write("topic1", "key1", "msg1")
msg.write("topic1", "key2", "msg1")
msg.write("topic2", "key1", "msg1")
msg.write("topic1", "key2", "msg2")
msg.write("topic2", "key1", "msg2")
msg.write("topic2", "key2", "msg2")
-
yield msg
-
+
msg.disconnect()
if msg.path != invalid_path():
shutil.rmtree(msg.path)
+
def empty_exception_message():
return "messaging exception "
+
def test_constructor():
msg = MsgLocal()
-
assert msg.logger == logging.getLogger('msg')
- assert msg.path == None
+ assert msg.path is None
assert len(msg.files_read) == 0
assert len(msg.files_write) == 0
assert len(msg.buffer) == 0
+
def test_constructor_with_logger():
logger_name = 'msg_local'
-
msg = MsgLocal(logger_name=logger_name)
-
assert msg.logger == logging.getLogger(logger_name)
- assert msg.path == None
+ assert msg.path is None
assert len(msg.files_read) == 0
assert len(msg.files_write) == 0
assert len(msg.buffer) == 0
+
@pytest.mark.parametrize("config, logger_name, path", [
({"logger_name": "msg_local", "path": valid_path()}, "msg_local", valid_path()),
({"logger_name": "msg_local", "path": valid_path()[:-1]}, "msg_local", valid_path()),
({"path": valid_path() + "test_it"}, "msg", valid_path() + "test_it/")])
def test_connect(msg_local, config, logger_name, path):
msg_local.connect(config)
-
assert msg_local.logger == logging.getLogger(logger_name)
assert msg_local.path == path
assert len(msg_local.files_read) == 0
assert len(msg_local.files_write) == 0
assert len(msg_local.buffer) == 0
+
@pytest.mark.parametrize("config", [
({"logger_name": "msg_local", "path": invalid_path()}),
({"path": invalid_path()})])
assert str(excinfo.value).startswith(empty_exception_message())
assert excinfo.value.http_code == http.HTTPStatus.INTERNAL_SERVER_ERROR
+
def test_disconnect(msg_local_config):
msg_local_config.disconnect()
-
for f in msg_local_config.files_read.values():
assert f.closed
-
for f in msg_local_config.files_write.values():
assert f.closed
+
def test_disconnect_with_read(msg_local_config):
msg_local_config.read('topic1', blocks=False)
msg_local_config.read('topic2', blocks=False)
-
msg_local_config.disconnect()
-
for f in msg_local_config.files_read.values():
assert f.closed
-
for f in msg_local_config.files_write.values():
assert f.closed
+
def test_disconnect_with_write(msg_local_with_data):
msg_local_with_data.disconnect()
for f in msg_local_with_data.files_write.values():
assert f.closed
+
def test_disconnect_with_read_and_write(msg_local_with_data):
msg_local_with_data.read('topic1', blocks=False)
msg_local_with_data.read('topic2', blocks=False)
msg_local_with_data.disconnect()
-
for f in msg_local_with_data.files_read.values():
assert f.closed
-
for f in msg_local_with_data.files_write.values():
assert f.closed
+
@pytest.mark.parametrize("topic, key, msg", [
("test_topic", "test_key", "test_msg"),
("test", "test_key", "test_msg"),
("test_topic", "test_none", None)])
def test_write(msg_local_config, topic, key, msg):
file_path = msg_local_config.path + topic
-
msg_local_config.write(topic, key, msg)
-
assert os.path.exists(file_path)
with open(file_path, 'r') as stream:
assert yaml.load(stream) == {key: msg if not isinstance(msg, tuple) else list(msg)}
+
@pytest.mark.parametrize("topic, key, msg, times", [
("test_topic", "test_key", "test_msg", 2),
("test", "test_key", "test_msg", 3),
for _ in range(times):
msg_local_config.write(topic, key, msg)
-
assert os.path.exists(file_path)
with open(file_path, 'r') as stream:
data = stream.readline()
assert yaml.load(data) == {key: msg if not isinstance(msg, tuple) else list(msg)}
+
def test_write_exception(msg_local_config):
msg_local_config.files_write = MagicMock()
msg_local_config.files_write.__contains__.side_effect = Exception()
assert str(excinfo.value).startswith(empty_exception_message())
assert excinfo.value.http_code == http.HTTPStatus.INTERNAL_SERVER_ERROR
+
@pytest.mark.parametrize("topics, datas", [
(["topic"], [{"key": "value"}]),
(["topic1"], [{"key": "value"}]),
for topic in topics:
for data in datas:
recv_topic, recv_key, recv_msg = msg_local_with_data.read(topic)
-
key = list(data.keys())[0]
val = data[key]
-
assert recv_topic == topic
assert recv_key == key
assert recv_msg == val
-
t.join()
+
@pytest.mark.parametrize("topics, datas", [
(["topic"], [{"key": "value"}]),
(["topic1"], [{"key": "value"}]),
for topic in topics:
for data in datas:
recv_topic, recv_key, recv_msg = msg_local_with_data.read(topic, blocks=False)
-
key = list(data.keys())[0]
val = data[key]
-
assert recv_topic == topic
assert recv_key == key
assert recv_msg == val
+
@pytest.mark.parametrize("topics, datas", [
(["topic"], [{"key": "value"}]),
(["topic1"], [{"key": "value"}]),
with open(msg_local_with_data.path + topic, "a+") as fp:
yaml.safe_dump(data, fp, default_flow_style=True, width=20000)
fp.flush()
-
# If file is not opened first, the messages written won't be seen
for topic in topics:
if topic not in msg_local_with_data.files_read:
msg_local_with_data.read(topic, blocks=False)
-
t = threading.Thread(target=write_to_topic, args=(topics, datas))
t.start()
for topic in topics:
recv_data = msg_local_with_data.read(topic, blocks=False)
-
- assert recv_data == None
-
+ assert recv_data is None
t.join()
+
@pytest.mark.parametrize("blocks", [
(True),
(False)])
assert str(excinfo.value).startswith(empty_exception_message())
assert excinfo.value.http_code == http.HTTPStatus.INTERNAL_SERVER_ERROR
+
@pytest.mark.parametrize("topics, datas", [
(["topic"], [{"key": "value"}]),
(["topic1"], [{"key": "value"}]),
with open(msg_local_with_data.path + topic, "a+") as fp:
yaml.safe_dump(data, fp, default_flow_style=True, width=20000)
fp.flush()
-
# If file is not opened first, the messages written won't be seen
for topic in topics:
if topic not in msg_local_with_data.files_read:
t = threading.Thread(target=write_to_topic, args=(topics, datas))
t.start()
-
for topic in topics:
for data in datas:
recv = event_loop.run_until_complete(msg_local_with_data.aioread(topic, event_loop))
recv_topic, recv_key, recv_msg = recv
-
key = list(data.keys())[0]
val = data[key]
-
assert recv_topic == topic
assert recv_key == key
assert recv_msg == val
-
t.join()
+
def test_aioread_exception(msg_local_with_data, event_loop):
msg_local_with_data.files_read = MagicMock()
msg_local_with_data.files_read.__contains__.side_effect = Exception()
assert str(excinfo.value).startswith(empty_exception_message())
assert excinfo.value.http_code == http.HTTPStatus.INTERNAL_SERVER_ERROR
+
def test_aioread_general_exception(msg_local_with_data, event_loop):
msg_local_with_data.read = MagicMock()
msg_local_with_data.read.side_effect = Exception()
assert str(excinfo.value).startswith(empty_exception_message())
assert excinfo.value.http_code == http.HTTPStatus.INTERNAL_SERVER_ERROR
+
@pytest.mark.parametrize("topic, key, msg", [
("test_topic", "test_key", "test_msg"),
("test", "test_key", "test_msg"),
("test_topic", "test_none", None)])
def test_aiowrite(msg_local_config, event_loop, topic, key, msg):
file_path = msg_local_config.path + topic
-
event_loop.run_until_complete(msg_local_config.aiowrite(topic, key, msg))
-
assert os.path.exists(file_path)
with open(file_path, 'r') as stream:
assert yaml.load(stream) == {key: msg if not isinstance(msg, tuple) else list(msg)}
+
@pytest.mark.parametrize("topic, key, msg, times", [
("test_topic", "test_key", "test_msg", 2),
("test", "test_key", "test_msg", 3),
("test_topic", "test_none", None, 3)])
def test_aiowrite_with_multiple_calls(msg_local_config, event_loop, topic, key, msg, times):
file_path = msg_local_config.path + topic
-
for _ in range(times):
event_loop.run_until_complete(msg_local_config.aiowrite(topic, key, msg))
-
assert os.path.exists(file_path)
with open(file_path, 'r') as stream:
data = stream.readline()
assert yaml.load(data) == {key: msg if not isinstance(msg, tuple) else list(msg)}
+
def test_aiowrite_exception(msg_local_config, event_loop):
msg_local_config.files_write = MagicMock()
msg_local_config.files_write.__contains__.side_effect = Exception()
projects / osm / common.git / blobdiff