build
dist
*.egg-info
+*.egg-info
+.eggs
+all: clean package
+
clean:
rm -rf dist deb_dist .build osm_common-*.tar.gz osm_common.egg-info eggs
cd deb_dist/osm-common*/ && dpkg-buildpackage -rfakeroot -uc -us
mkdir -p .build
cp deb_dist/python3-osm-common*.deb .build/
-
-
#!/bin/sh
-#tox
+tox -e flake8
+
def get_list(self, table, filter={}):
try:
- l = []
+ result = []
for _, row in self._find(table, self._format_filter(filter)):
- l.append(deepcopy(row))
- return l
+ result.append(deepcopy(row))
+ return result
except DbException:
raise
except Exception as e: # TODO refine
def get_one(self, table, filter={}, fail_on_empty=True, fail_on_more=True):
try:
- l = None
+ result = None
for _, row in self._find(table, self._format_filter(filter)):
if not fail_on_more:
return deepcopy(row)
- if l:
+ if result:
raise DbException("Found more than one entry with filter='{}'".format(filter),
HTTPStatus.CONFLICT.value)
- l = row
- if not l and fail_on_empty:
+ result = row
+ if not result and fail_on_empty:
raise DbException("Not found entry with filter='{}'".format(filter), HTTPStatus.NOT_FOUND)
- return deepcopy(l)
+ return deepcopy(result)
except Exception as e: # TODO refine
raise DbException(str(e))
dot_index = query_k.rfind(".")
if dot_index > 1 and query_k[dot_index+1:] in ("eq", "ne", "gt", "gte", "lt", "lte", "cont",
"ncont", "neq"):
- operator = "$" + query_k[dot_index+1:]
+ operator = "$" + query_k[dot_index + 1:]
if operator == "$neq":
operator = "$ne"
k = query_k[:dot_index]
def get_list(self, table, filter={}):
try:
- l = []
+ result = []
collection = self.db[table]
db_filter = self._format_filter(filter)
rows = collection.find(db_filter)
for row in rows:
- l.append(row)
- return l
+ result.append(row)
+ return result
except DbException:
raise
except Exception as e: # TODO refine
from osm_common.dbbase import DbBase, DbException
+
def exception_message(message):
return "database exception " + message
+
@pytest.fixture
def db_base():
return DbBase()
+
def test_constructor():
db_base = DbBase()
-
- assert db_base != None
+ assert db_base is not None
assert isinstance(db_base, DbBase)
+
def test_db_connect(db_base):
db_base.db_connect(None)
+
def test_db_disconnect(db_base):
db_base.db_disconnect()
+
def test_get_list(db_base):
with pytest.raises(DbException) as excinfo:
db_base.get_list(None, None)
assert str(excinfo.value).startswith(exception_message("Method 'get_list' not implemented"))
assert excinfo.value.http_code == http.HTTPStatus.NOT_FOUND
+
def test_get_one(db_base):
with pytest.raises(DbException) as excinfo:
db_base.get_one(None, None, None, None)
assert str(excinfo.value).startswith(exception_message("Method 'get_one' not implemented"))
assert excinfo.value.http_code == http.HTTPStatus.NOT_FOUND
+
def test_create(db_base):
with pytest.raises(DbException) as excinfo:
db_base.create(None, None)
assert str(excinfo.value).startswith(exception_message("Method 'create' not implemented"))
assert excinfo.value.http_code == http.HTTPStatus.NOT_FOUND
+
def test_del_list(db_base):
with pytest.raises(DbException) as excinfo:
db_base.del_list(None, None)
assert str(excinfo.value).startswith(exception_message("Method 'del_list' not implemented"))
assert excinfo.value.http_code == http.HTTPStatus.NOT_FOUND
+
def test_del_one(db_base):
with pytest.raises(DbException) as excinfo:
db_base.del_one(None, None, None)
__author__ = 'Eduardo Sousa <eduardosousa@av.it.pt>'
+
@pytest.fixture
def db_memory():
db = DbMemory()
return db
+
@pytest.fixture
def db_memory_with_data():
db = DbMemory()
return db
+
def empty_exception_message():
return 'database exception '
+
def get_one_exception_message(filter):
return "database exception Not found entry with filter='{}'".format(filter)
+
def get_one_multiple_exception_message(filter):
return "database exception Found more than one entry with filter='{}'".format(filter)
+
def del_one_exception_message(filter):
return "database exception Not found entry with filter='{}'".format(filter)
+
def replace_exception_message(filter):
return "database exception Not found entry with filter='{}'".format(filter)
+
def test_constructor():
db = DbMemory()
-
assert db.logger == logging.getLogger('db')
assert len(db.db) == 0
+
def test_constructor_with_logger():
logger_name = 'db_local'
-
db = DbMemory(logger_name=logger_name)
-
assert db.logger == logging.getLogger(logger_name)
assert len(db.db) == 0
+
def test_db_connect():
logger_name = 'db_local'
config = {'logger_name': logger_name}
-
db = DbMemory()
db.db_connect(config)
-
assert db.logger == logging.getLogger(logger_name)
assert len(db.db) == 0
+
def test_db_disconnect(db_memory):
db_memory.db_disconnect()
+
@pytest.mark.parametrize("table, filter", [
("test", {}),
("test", {"_id": 1}),
("test", {"_id": 1, "data": 1})])
def test_get_list_with_empty_db(db_memory, table, filter):
result = db_memory.get_list(table, filter)
-
assert len(result) == 0
+
@pytest.mark.parametrize("table, filter, expected_data", [
("test", {}, [{"_id": 1, "data": 1}, {"_id": 2, "data": 2}, {"_id": 3, "data": 3}]),
("test", {"_id": 1}, [{"_id": 1, "data": 1}]),
("test_table", {"_id": 1, "data": 1}, [])])
def test_get_list_with_non_empty_db(db_memory_with_data, table, filter, expected_data):
result = db_memory_with_data.get_list(table, filter)
-
assert len(result) == len(expected_data)
for data in expected_data:
assert data in result
+
def test_get_list_exception(db_memory_with_data):
table = 'test'
filter = {}
-
db_memory_with_data._find = MagicMock(side_effect=Exception())
-
with pytest.raises(DbException) as excinfo:
db_memory_with_data.get_list(table, filter)
assert str(excinfo.value) == empty_exception_message()
assert excinfo.value.http_code == http.HTTPStatus.NOT_FOUND
+
@pytest.mark.parametrize("table, filter, expected_data", [
("test", {"_id": 1}, {"_id": 1, "data": 1}),
("test", {"_id": 2}, {"_id": 2, "data": 2}),
("test", {"_id": 3, "data": 3}, {"_id": 3, "data": 3})])
def test_get_one(db_memory_with_data, table, filter, expected_data):
result = db_memory_with_data.get_one(table, filter)
-
assert result == expected_data
assert len(db_memory_with_data.db) == 1
assert table in db_memory_with_data.db
assert len(db_memory_with_data.db[table]) == 3
assert result in db_memory_with_data.db[table]
+
@pytest.mark.parametrize("table, filter, expected_data", [
("test", {}, {"_id": 1, "data": 1})])
def test_get_one_with_multiple_results(db_memory_with_data, table, filter, expected_data):
result = db_memory_with_data.get_one(table, filter, fail_on_more=False)
-
assert result == expected_data
assert len(db_memory_with_data.db) == 1
assert table in db_memory_with_data.db
assert len(db_memory_with_data.db[table]) == 3
assert result in db_memory_with_data.db[table]
+
def test_get_one_with_multiple_results_exception(db_memory_with_data):
table = "test"
filter = {}
-
with pytest.raises(DbException) as excinfo:
db_memory_with_data.get_one(table, filter)
-
assert str(excinfo.value) == (empty_exception_message() + get_one_multiple_exception_message(filter))
-# assert excinfo.value.http_code == http.HTTPStatus.CONFLICT
+ # assert excinfo.value.http_code == http.HTTPStatus.CONFLICT
+
@pytest.mark.parametrize("table, filter", [
("test", {"_id": 4}),
assert str(excinfo.value) == (empty_exception_message() + get_one_exception_message(filter))
assert excinfo.value.http_code == http.HTTPStatus.NOT_FOUND
+
@pytest.mark.parametrize("table, filter", [
("test", {"_id": 4}),
("test", {"data": 4}),
("test_table", {"_id": 4, "data": 4})])
def test_get_one_with_non_empty_db_none(db_memory_with_data, table, filter):
result = db_memory_with_data.get_one(table, filter, fail_on_empty=False)
-
- assert result == None
+ assert result is None
+
@pytest.mark.parametrize("table, filter", [
("test", {"_id": 4}),
assert str(excinfo.value) == (empty_exception_message() + get_one_exception_message(filter))
assert excinfo.value.http_code == http.HTTPStatus.NOT_FOUND
+
@pytest.mark.parametrize("table, filter", [
("test", {"_id": 4}),
("test", {"data": 4}),
("test_table", {"_id": 4, "data": 4})])
def test_get_one_with_empty_db_none(db_memory, table, filter):
result = db_memory.get_one(table, filter, fail_on_empty=False)
-
- assert result == None
+ assert result is None
+
def test_get_one_generic_exception(db_memory_with_data):
table = 'test'
filter = {}
-
db_memory_with_data._find = MagicMock(side_effect=Exception())
-
with pytest.raises(DbException) as excinfo:
db_memory_with_data.get_one(table, filter)
assert str(excinfo.value) == empty_exception_message()
assert excinfo.value.http_code == http.HTTPStatus.NOT_FOUND
+
@pytest.mark.parametrize("table, filter, expected_data", [
("test", {}, []),
("test", {"_id": 1}, [{"_id": 2, "data": 2}, {"_id": 3, "data": 3}]),
("test", {"_id": 2, "data": 2}, [{"_id": 1, "data": 1}, {"_id": 3, "data": 3}])])
def test_del_list_with_non_empty_db(db_memory_with_data, table, filter, expected_data):
result = db_memory_with_data.del_list(table, filter)
-
assert result["deleted"] == (3 - len(expected_data))
assert len(db_memory_with_data.db) == 1
assert table in db_memory_with_data.db
for data in expected_data:
assert data in db_memory_with_data.db[table]
+
@pytest.mark.parametrize("table, filter", [
("test", {}),
("test", {"_id": 1}),
result = db_memory.del_list(table, filter)
assert result['deleted'] == 0
+
def test_del_list_generic_exception(db_memory_with_data):
table = 'test'
filter = {}
-
db_memory_with_data._find = MagicMock(side_effect=Exception())
-
with pytest.raises(DbException) as excinfo:
db_memory_with_data.del_list(table, filter)
assert str(excinfo.value) == empty_exception_message()
assert excinfo.value.http_code == http.HTTPStatus.NOT_FOUND
+
@pytest.mark.parametrize("table, filter, data", [
("test", {}, {"_id": 1, "data": 1}),
("test", {"_id": 1}, {"_id": 1, "data": 1}),
("test", {"_id": 2, "data": 2}, {"_id": 2, "data": 2})])
def test_del_one(db_memory_with_data, table, filter, data):
result = db_memory_with_data.del_one(table, filter)
-
assert result == {"deleted": 1}
assert len(db_memory_with_data.db) == 1
assert table in db_memory_with_data.db
assert len(db_memory_with_data.db[table]) == 2
assert data not in db_memory_with_data.db[table]
+
@pytest.mark.parametrize("table, filter", [
("test", {}),
("test", {"_id": 1}),
assert str(excinfo.value) == (empty_exception_message() + del_one_exception_message(filter))
assert excinfo.value.http_code == http.HTTPStatus.NOT_FOUND
+
@pytest.mark.parametrize("table, filter", [
("test", {}),
("test", {"_id": 1}),
("test_table", {"_id": 2, "data": 2})])
def test_del_one_with_empty_db_none(db_memory, table, filter):
result = db_memory.del_one(table, filter, fail_on_empty=False)
+ assert result is None
- assert result == None
@pytest.mark.parametrize("table, filter", [
("test", {"_id": 4}),
assert str(excinfo.value) == (empty_exception_message() + del_one_exception_message(filter))
assert excinfo.value.http_code == http.HTTPStatus.NOT_FOUND
+
@pytest.mark.parametrize("table, filter", [
("test", {"_id": 4}),
("test", {"_id": 5}),
("test_table", {"_id": 2, "data": 2})])
def test_del_one_with_non_empty_db_none(db_memory_with_data, table, filter):
result = db_memory_with_data.del_one(table, filter, fail_on_empty=False)
+ assert result is None
- assert result == None
@pytest.mark.parametrize("fail_on_empty", [
(True),
def test_del_one_generic_exception(db_memory_with_data, fail_on_empty):
table = 'test'
filter = {}
-
db_memory_with_data._find = MagicMock(side_effect=Exception())
-
with pytest.raises(DbException) as excinfo:
db_memory_with_data.del_one(table, filter, fail_on_empty=fail_on_empty)
assert str(excinfo.value) == empty_exception_message()
assert excinfo.value.http_code == http.HTTPStatus.NOT_FOUND
+
@pytest.mark.parametrize("table, filter, indata", [
("test", {}, {"_id": 1, "data": 42}),
("test", {}, {"_id": 3, "data": 42}),
("test", {"_id": 3, "data": 3}, {"_id": 3, "data": 42})])
def test_replace(db_memory_with_data, table, filter, indata):
result = db_memory_with_data.replace(table, filter, indata)
-
assert result == {"updated": 1}
assert len(db_memory_with_data.db) == 1
assert table in db_memory_with_data.db
assert len(db_memory_with_data.db[table]) == 3
assert indata in db_memory_with_data.db[table]
+
@pytest.mark.parametrize("table, filter, indata", [
("test", {}, {'_id': 1, 'data': 1}),
("test", {}, {'_id': 2, 'data': 1}),
assert str(excinfo.value) == (empty_exception_message() + replace_exception_message(filter))
assert excinfo.value.http_code == http.HTTPStatus.NOT_FOUND
+
@pytest.mark.parametrize("table, filter, indata", [
("test", {}, {'_id': 1, 'data': 1}),
("test", {}, {'_id': 2, 'data': 1}),
("test_table", {'_id': 1, 'data': 1}, {'_id': 1, 'data': 1})])
def test_replace_without_data_none(db_memory, table, filter, indata):
result = db_memory.replace(table, filter, indata, fail_on_empty=False)
- assert result == None
+ assert result is None
+
@pytest.mark.parametrize("table, filter, indata", [
("test_table", {}, {'_id': 1, 'data': 1}),
assert str(excinfo.value) == (empty_exception_message() + replace_exception_message(filter))
assert excinfo.value.http_code == http.HTTPStatus.NOT_FOUND
+
@pytest.mark.parametrize("table, filter, indata", [
("test_table", {}, {'_id': 1, 'data': 1}),
("test_table", {}, {'_id': 2, 'data': 1}),
("test_table", {'_id': 1, 'data': 1}, {'_id': 1, 'data': 1})])
def test_replace_with_data_none(db_memory_with_data, table, filter, indata):
result = db_memory_with_data.replace(table, filter, indata, fail_on_empty=False)
- assert result == None
+ assert result is None
+
@pytest.mark.parametrize("fail_on_empty", [
- (True),
- (False)])
+ True,
+ False])
def test_replace_generic_exception(db_memory_with_data, fail_on_empty):
table = 'test'
filter = {}
indata = {'_id': 1, 'data': 1}
-
db_memory_with_data._find = MagicMock(side_effect=Exception())
-
with pytest.raises(DbException) as excinfo:
db_memory_with_data.replace(table, filter, indata, fail_on_empty=fail_on_empty)
assert str(excinfo.value) == empty_exception_message()
assert excinfo.value.http_code == http.HTTPStatus.NOT_FOUND
+
@pytest.mark.parametrize("table, id, data", [
("test", "1", {"data": 1}),
("test", "1", {"data": 2}),
def test_create_with_empty_db_with_id(db_memory, table, id, data):
data_to_insert = data
data_to_insert['_id'] = id
-
returned_id = db_memory.create(table, data_to_insert)
-
assert returned_id == id
assert len(db_memory.db) == 1
assert table in db_memory.db
assert len(db_memory.db[table]) == 1
assert data_to_insert in db_memory.db[table]
+
@pytest.mark.parametrize("table, id, data", [
("test", "4", {"data": 1}),
("test", "5", {"data": 2}),
def test_create_with_non_empty_db_with_id(db_memory_with_data, table, id, data):
data_to_insert = data
data_to_insert['_id'] = id
-
returned_id = db_memory_with_data.create(table, data_to_insert)
-
assert returned_id == id
assert len(db_memory_with_data.db) == (1 if table == 'test' else 2)
assert table in db_memory_with_data.db
assert len(db_memory_with_data.db[table]) == (4 if table == 'test' else 1)
assert data_to_insert in db_memory_with_data.db[table]
+
@pytest.mark.parametrize("table, data", [
("test", {"data": 1}),
("test", {"data": 2}),
("test_table", {"data_1": 2, "data_2": 1})])
def test_create_with_empty_db_without_id(db_memory, table, data):
returned_id = db_memory.create(table, data)
-
assert len(db_memory.db) == 1
assert table in db_memory.db
assert len(db_memory.db[table]) == 1
-
data_inserted = data
data_inserted['_id'] = returned_id
-
assert data_inserted in db_memory.db[table]
+
@pytest.mark.parametrize("table, data", [
("test", {"data": 1}),
("test", {"data": 2}),
("test_table", {"data_1": 2, "data_2": 1})])
def test_create_with_non_empty_db_without_id(db_memory_with_data, table, data):
returned_id = db_memory_with_data.create(table, data)
-
assert len(db_memory_with_data.db) == (1 if table == 'test' else 2)
assert table in db_memory_with_data.db
assert len(db_memory_with_data.db[table]) == (4 if table == 'test' else 1)
-
data_inserted = data
data_inserted['_id'] = returned_id
-
assert data_inserted in db_memory_with_data.db[table]
+
def test_create_with_exception(db_memory):
table = "test"
data = {"_id": 1, "data": 1}
-
db_memory.db = MagicMock()
db_memory.db.__contains__.side_effect = Exception()
-
with pytest.raises(DbException) as excinfo:
db_memory.create(table, data)
assert str(excinfo.value) == empty_exception_message()
from osm_common.fsbase import FsBase, FsException
+
def exception_message(message):
return "storage exception " + message
+
@pytest.fixture
def fs_base():
return FsBase()
+
def test_constructor():
fs_base = FsBase()
-
- assert fs_base != None
+ assert fs_base is not None
assert isinstance(fs_base, FsBase)
+
def test_get_params(fs_base):
params = fs_base.get_params()
-
assert isinstance(params, dict)
assert len(params) == 0
+
def test_fs_connect(fs_base):
fs_base.fs_connect(None)
+
def test_fs_disconnect(fs_base):
fs_base.fs_disconnect()
+
def test_mkdir(fs_base):
with pytest.raises(FsException) as excinfo:
fs_base.mkdir(None)
assert str(excinfo.value).startswith(exception_message("Method 'mkdir' not implemented"))
assert excinfo.value.http_code == http.HTTPStatus.INTERNAL_SERVER_ERROR
+
def test_file_exists(fs_base):
with pytest.raises(FsException) as excinfo:
fs_base.file_exists(None)
assert str(excinfo.value).startswith(exception_message("Method 'file_exists' not implemented"))
assert excinfo.value.http_code == http.HTTPStatus.INTERNAL_SERVER_ERROR
+
def test_file_size(fs_base):
with pytest.raises(FsException) as excinfo:
fs_base.file_size(None)
assert str(excinfo.value).startswith(exception_message("Method 'file_size' not implemented"))
assert excinfo.value.http_code == http.HTTPStatus.INTERNAL_SERVER_ERROR
+
def test_file_extract(fs_base):
with pytest.raises(FsException) as excinfo:
fs_base.file_extract(None, None)
assert str(excinfo.value).startswith(exception_message("Method 'file_extract' not implemented"))
assert excinfo.value.http_code == http.HTTPStatus.INTERNAL_SERVER_ERROR
+
def test_file_open(fs_base):
with pytest.raises(FsException) as excinfo:
fs_base.file_open(None, None)
assert str(excinfo.value).startswith(exception_message("Method 'file_open' not implemented"))
assert excinfo.value.http_code == http.HTTPStatus.INTERNAL_SERVER_ERROR
+
def test_file_delete(fs_base):
with pytest.raises(FsException) as excinfo:
fs_base.file_delete(None, None)
__author__ = "Eduardo Sousa <eduardosousa@av.it.pt>"
+
def valid_path():
return tempfile.gettempdir() + '/'
+
def invalid_path():
return '/#tweeter/'
+
@pytest.fixture
def fs_local():
fs = FsLocal()
fs.fs_connect({'path': valid_path()})
-
return fs
+
def fs_connect_exception_message(path):
return "storage exception Invalid configuration param at '[storage]': path '{}' does not exist".format(path)
+
def file_open_file_not_found_exception(storage):
f = storage if isinstance(storage, str) else '/'.join(storage)
return "storage exception File {} does not exist".format(f)
+
def file_open_io_exception(storage):
f = storage if isinstance(storage, str) else '/'.join(storage)
return "storage exception File {} cannot be opened".format(f)
+
def dir_ls_not_a_directory_exception(storage):
f = storage if isinstance(storage, str) else '/'.join(storage)
return "storage exception File {} does not exist".format(f)
+
def dir_ls_io_exception(storage):
f = storage if isinstance(storage, str) else '/'.join(storage)
return "storage exception File {} cannot be opened".format(f)
+
def file_delete_exception_message(storage):
return "storage exception File {} does not exist".format(storage)
+
def test_constructor_without_logger():
fs = FsLocal()
-
assert fs.logger == logging.getLogger('fs')
assert fs.path is None
+
def test_constructor_with_logger():
logger_name = 'fs_local'
-
fs = FsLocal(logger_name=logger_name)
-
assert fs.logger == logging.getLogger(logger_name)
assert fs.path is None
+
def test_get_params(fs_local):
params = fs_local.get_params()
-
assert len(params) == 2
assert "fs" in params
assert "path" in params
assert params["fs"] == "local"
assert params["path"] == valid_path()
+
@pytest.mark.parametrize("config, exp_logger, exp_path", [
({'logger_name': 'fs_local', 'path': valid_path()}, 'fs_local', valid_path()),
({'logger_name': 'fs_local', 'path': valid_path()[:-1]}, 'fs_local', valid_path()),
def test_fs_connect_with_valid_config(config, exp_logger, exp_path):
fs = FsLocal()
fs.fs_connect(config)
-
assert fs.logger == logging.getLogger(exp_logger)
assert fs.path == exp_path
+
@pytest.mark.parametrize("config, exp_exception_message", [
({'logger_name': 'fs_local', 'path': invalid_path()}, fs_connect_exception_message(invalid_path())),
({'logger_name': 'fs_local', 'path': invalid_path()[:-1]}, fs_connect_exception_message(invalid_path()[:-1])),
({'path': invalid_path()[:-1]}, fs_connect_exception_message(invalid_path()[:-1]))])
def test_fs_connect_with_invalid_path(config, exp_exception_message):
fs = FsLocal()
-
with pytest.raises(FsException) as excinfo:
fs.fs_connect(config)
assert str(excinfo.value) == exp_exception_message
+
def test_fs_disconnect(fs_local):
fs_local.fs_disconnect()
+
def test_mkdir_with_valid_path(fs_local):
folder_name = str(uuid.uuid4())
folder_path = valid_path() + folder_name
-
fs_local.mkdir(folder_name)
-
assert os.path.exists(folder_path)
-
os.rmdir(folder_path)
+
def test_mkdir_with_exception(fs_local):
folder_name = str(uuid.uuid4())
folder_path = valid_path() + folder_name
os.mkdir(folder_path)
-
with pytest.raises(FsException) as excinfo:
fs_local.mkdir(folder_name)
assert excinfo.value.http_code == http.HTTPStatus.INTERNAL_SERVER_ERROR
-
os.rmdir(folder_path)
+
@pytest.mark.parametrize("storage, mode, expected", [
(str(uuid.uuid4()), 'file', False),
([str(uuid.uuid4())], 'file', False),
def test_file_exists_returns_false(fs_local, storage, mode, expected):
assert fs_local.file_exists(storage, mode) == expected
+
@pytest.mark.parametrize("storage, mode, expected", [
(str(uuid.uuid4()), 'file', True),
([str(uuid.uuid4())], 'file', True),
([str(uuid.uuid4())], 'dir', True)])
def test_file_exists_returns_true(fs_local, storage, mode, expected):
path = valid_path() + storage if isinstance(storage, str) else valid_path() + storage[0]
-
if mode == 'file':
os.mknod(path)
elif mode == 'dir':
os.mkdir(path)
-
assert fs_local.file_exists(storage, mode) == expected
-
if mode == 'file':
os.remove(path)
elif mode == 'dir':
os.rmdir(path)
+
@pytest.mark.parametrize("storage, mode", [
(str(uuid.uuid4()), 'file'),
([str(uuid.uuid4())], 'file'),
([str(uuid.uuid4())], 'dir')])
def test_file_size(fs_local, storage, mode):
path = valid_path() + storage if isinstance(storage, str) else valid_path() + storage[0]
-
if mode == 'file':
os.mknod(path)
elif mode == 'dir':
os.mkdir(path)
-
size = os.path.getsize(path)
-
assert fs_local.file_size(storage) == size
-
if mode == 'file':
os.remove(path)
elif mode == 'dir':
os.rmdir(path)
+
@pytest.mark.parametrize("files, path", [
(['foo', 'bar', 'foobar'], str(uuid.uuid4())),
(['foo', 'bar', 'foobar'], [str(uuid.uuid4())])])
def test_file_extract(fs_local, files, path):
for f in files:
os.mknod(valid_path() + f)
-
tar_path = valid_path() + str(uuid.uuid4()) + '.tar'
with tarfile.open(tar_path, 'w') as tar:
for f in files:
tar.add(valid_path() + f, arcname=f)
-
with tarfile.open(tar_path, 'r') as tar:
fs_local.file_extract(tar, path)
-
extracted_path = valid_path() + (path if isinstance(path, str) else '/'.join(path))
ls_dir = os.listdir(extracted_path)
-
assert len(ls_dir) == len(files)
for f in files:
assert f in ls_dir
-
os.remove(tar_path)
-
for f in files:
os.remove(valid_path() + f)
-
shutil.rmtree(extracted_path)
+
@pytest.mark.parametrize("storage, mode", [
(str(uuid.uuid4()), 'r'),
(str(uuid.uuid4()), 'w'),
([str(uuid.uuid4())], 'ab')])
def test_file_open(fs_local, storage, mode):
path = valid_path() + storage if isinstance(storage, str) else valid_path() + storage[0]
-
os.mknod(path)
-
file_obj = fs_local.file_open(storage, mode)
-
assert isinstance(file_obj, io.IOBase)
- assert file_obj.closed == False
-
+ assert file_obj.closed is False
os.remove(path)
+
@pytest.mark.parametrize("storage, mode", [
(str(uuid.uuid4()), 'r'),
(str(uuid.uuid4()), 'rb'),
assert str(excinfo.value) == file_open_file_not_found_exception(storage)
assert excinfo.value.http_code == http.HTTPStatus.NOT_FOUND
+
@pytest.mark.parametrize("storage, mode", [
(str(uuid.uuid4()), 'r'),
(str(uuid.uuid4()), 'w'),
([str(uuid.uuid4())], 'ab')])
def test_file_open_io_error(fs_local, storage, mode):
path = valid_path() + storage if isinstance(storage, str) else valid_path() + storage[0]
-
os.mknod(path)
os.chmod(path, 0)
-
with pytest.raises(FsException) as excinfo:
fs_local.file_open(storage, mode)
assert str(excinfo.value) == file_open_io_exception(storage)
assert excinfo.value.http_code == http.HTTPStatus.BAD_REQUEST
-
os.remove(path)
+
@pytest.mark.parametrize("storage, with_files", [
(str(uuid.uuid4()), True),
(str(uuid.uuid4()), False),
([str(uuid.uuid4())], False)])
def test_dir_ls(fs_local, storage, with_files):
path = valid_path() + storage if isinstance(storage, str) else valid_path() + storage[0]
-
os.mkdir(path)
-
- if with_files == True:
+ if with_files is True:
file_name = str(uuid.uuid4())
file_path = path + '/' + file_name
os.mknod(file_path)
-
result = fs_local.dir_ls(storage)
- if with_files == True:
+ if with_files is True:
assert len(result) == 1
assert result[0] == file_name
else:
assert len(result) == 0
-
shutil.rmtree(path)
+
@pytest.mark.parametrize("storage", [
(str(uuid.uuid4())),
([str(uuid.uuid4())])])
def test_dir_ls_with_not_a_directory_error(fs_local, storage):
path = valid_path() + storage if isinstance(storage, str) else valid_path() + storage[0]
-
os.mknod(path)
-
with pytest.raises(FsException) as excinfo:
fs_local.dir_ls(storage)
assert str(excinfo.value) == dir_ls_not_a_directory_exception(storage)
assert excinfo.value.http_code == http.HTTPStatus.NOT_FOUND
-
os.remove(path)
+
@pytest.mark.parametrize("storage", [
(str(uuid.uuid4())),
([str(uuid.uuid4())])])
def test_dir_ls_with_io_error(fs_local, storage):
path = valid_path() + storage if isinstance(storage, str) else valid_path() + storage[0]
-
os.mkdir(path)
os.chmod(path, 0)
-
with pytest.raises(FsException) as excinfo:
fs_local.dir_ls(storage)
assert str(excinfo.value) == dir_ls_io_exception(storage)
assert excinfo.value.http_code == http.HTTPStatus.BAD_REQUEST
-
os.rmdir(path)
+
@pytest.mark.parametrize("storage, with_files, ignore_non_exist", [
(str(uuid.uuid4()), True, True),
(str(uuid.uuid4()), False, True),
([str(uuid.uuid4())], False, False)])
def test_file_delete_with_dir(fs_local, storage, with_files, ignore_non_exist):
path = valid_path() + storage if isinstance(storage, str) else valid_path() + storage[0]
-
os.mkdir(path)
-
- if with_files == True:
+ if with_files is True:
file_path = path + '/' + str(uuid.uuid4())
os.mknod(file_path)
-
fs_local.file_delete(storage, ignore_non_exist)
+ assert os.path.exists(path) is False
- assert os.path.exists(path) == False
@pytest.mark.parametrize("storage", [
(str(uuid.uuid4())),
assert str(excinfo.value) == file_delete_exception_message(storage)
assert excinfo.value.http_code == http.HTTPStatus.NOT_FOUND
+
@pytest.mark.parametrize("storage", [
(str(uuid.uuid4())),
([str(uuid.uuid4())])])
def test_file_delete_no_exception(fs_local, storage):
path = valid_path() + storage if isinstance(storage, str) else valid_path() + storage[0]
-
fs_local.file_delete(storage, ignore_non_exist=True)
-
- assert os.path.exists(path) == False
+ assert os.path.exists(path) is False
def exception_message(message):
return "messaging exception " + message
+
@pytest.fixture
def msg_base():
return MsgBase()
+
def test_constructor():
msgbase = MsgBase()
-
- assert msgbase != None
+ assert msgbase is not None
assert isinstance(msgbase, MsgBase)
+
def test_connect(msg_base):
msg_base.connect(None)
+
def test_disconnect(msg_base):
msg_base.disconnect()
+
def test_write(msg_base):
with pytest.raises(MsgException) as excinfo:
msg_base.write("test", "test", "test")
assert str(excinfo.value).startswith(exception_message("Method 'write' not implemented"))
assert excinfo.value.http_code == http.HTTPStatus.INTERNAL_SERVER_ERROR
+
def test_read(msg_base):
with pytest.raises(MsgException) as excinfo:
msg_base.read("test")
assert str(excinfo.value).startswith(exception_message("Method 'read' not implemented"))
assert excinfo.value.http_code == http.HTTPStatus.INTERNAL_SERVER_ERROR
+
def test_aiowrite(msg_base, event_loop):
with pytest.raises(MsgException) as excinfo:
event_loop.run_until_complete(msg_base.aiowrite("test", "test", "test", event_loop))
assert str(excinfo.value).startswith(exception_message("Method 'aiowrite' not implemented"))
assert excinfo.value.http_code == http.HTTPStatus.INTERNAL_SERVER_ERROR
+
def test_aioread(msg_base, event_loop):
with pytest.raises(MsgException) as excinfo:
event_loop.run_until_complete(msg_base.aioread("test", event_loop))
__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()
basepython = python3
deps = flake8
commands =
- flake8 setup.py
+ flake8 osm_common/ setup.py --max-line-length 120 --exclude .svn,CVS,.gz,.git,__pycache__,.tox,local,temp --ignore W291,W293,E226
[testenv:build]
basepython = python3
projects / osm / common.git / commitdiff