Reformat common to standardized format
Change-Id: I9ba54f3510e17de285e1e774d18ee5c7205818ad
Signed-off-by: garciadeblas <gerardo.garciadeblas@telefonica.com>
diff --git a/osm_common/tests/packages/invalid_package_vnf/Scripts/charms/simple/src/charm.py b/osm_common/tests/packages/invalid_package_vnf/Scripts/charms/simple/src/charm.py
index 409f286..54cefc4 100755
--- a/osm_common/tests/packages/invalid_package_vnf/Scripts/charms/simple/src/charm.py
+++ b/osm_common/tests/packages/invalid_package_vnf/Scripts/charms/simple/src/charm.py
@@ -27,7 +27,6 @@
class MyNativeCharm(CharmBase):
-
def __init__(self, framework, key):
super().__init__(framework, key)
diff --git a/osm_common/tests/packages/invalid_package_vnf/manifest.mf b/osm_common/tests/packages/invalid_package_vnf/manifest.mf
index dacf77f..007dde7 100644
--- a/osm_common/tests/packages/invalid_package_vnf/manifest.mf
+++ b/osm_common/tests/packages/invalid_package_vnf/manifest.mf
@@ -45,7 +45,7 @@
# Invalid hash algorithm
Source: Scripts/charms/simple/src/charm.py
Algorithm: SHA-733
-Hash: 7895f7b9e1b7ed5b5bcd64398950ca95b456d7fc973334351474eed466c2f480
+Hash: ea72f897a966e6174ed9164fabc3c500df5a2f712eb6b22ab2408afb07d04d14
# Wrong hash
Source: Scripts/charms/simple/hooks/start
diff --git a/osm_common/tests/packages/native_charm_with_metadata_dir_vnf/Scripts/charms/simple/src/charm.py b/osm_common/tests/packages/native_charm_with_metadata_dir_vnf/Scripts/charms/simple/src/charm.py
index 409f286..54cefc4 100755
--- a/osm_common/tests/packages/native_charm_with_metadata_dir_vnf/Scripts/charms/simple/src/charm.py
+++ b/osm_common/tests/packages/native_charm_with_metadata_dir_vnf/Scripts/charms/simple/src/charm.py
@@ -27,7 +27,6 @@
class MyNativeCharm(CharmBase):
-
def __init__(self, framework, key):
super().__init__(framework, key)
diff --git a/osm_common/tests/packages/native_charm_with_metadata_dir_vnf/manifest.mf b/osm_common/tests/packages/native_charm_with_metadata_dir_vnf/manifest.mf
index b42c240..ab907a8 100644
--- a/osm_common/tests/packages/native_charm_with_metadata_dir_vnf/manifest.mf
+++ b/osm_common/tests/packages/native_charm_with_metadata_dir_vnf/manifest.mf
@@ -46,7 +46,7 @@
Source: Scripts/charms/simple/src/charm.py
Algorithm: SHA-256
-Hash: 7895f7b9e1b7ed5b5bcd64398950ca95b456d7fc973334351474eed466c2f480
+Hash: ea72f897a966e6174ed9164fabc3c500df5a2f712eb6b22ab2408afb07d04d14
Source: Scripts/charms/simple/hooks/start
Algorithm: SHA-256
diff --git a/osm_common/tests/packages/native_charm_without_metadata_dir_vnf/Scripts/charms/simple/src/charm.py b/osm_common/tests/packages/native_charm_without_metadata_dir_vnf/Scripts/charms/simple/src/charm.py
index 409f286..54cefc4 100755
--- a/osm_common/tests/packages/native_charm_without_metadata_dir_vnf/Scripts/charms/simple/src/charm.py
+++ b/osm_common/tests/packages/native_charm_without_metadata_dir_vnf/Scripts/charms/simple/src/charm.py
@@ -27,7 +27,6 @@
class MyNativeCharm(CharmBase):
-
def __init__(self, framework, key):
super().__init__(framework, key)
diff --git a/osm_common/tests/packages/native_charm_without_metadata_dir_vnf/native_charm_vnfd.mf b/osm_common/tests/packages/native_charm_without_metadata_dir_vnf/native_charm_vnfd.mf
index d948858..fbe1a70 100644
--- a/osm_common/tests/packages/native_charm_without_metadata_dir_vnf/native_charm_vnfd.mf
+++ b/osm_common/tests/packages/native_charm_without_metadata_dir_vnf/native_charm_vnfd.mf
@@ -46,7 +46,7 @@
Source: Scripts/charms/simple/src/charm.py
Algorithm: SHA-256
-Hash: 7895f7b9e1b7ed5b5bcd64398950ca95b456d7fc973334351474eed466c2f480
+Hash: ea72f897a966e6174ed9164fabc3c500df5a2f712eb6b22ab2408afb07d04d14
Source: Scripts/charms/simple/hooks/start
Algorithm: SHA-256
diff --git a/osm_common/tests/test_dbbase.py b/osm_common/tests/test_dbbase.py
index 1abd1c7..117350e 100644
--- a/osm_common/tests/test_dbbase.py
+++ b/osm_common/tests/test_dbbase.py
@@ -43,7 +43,9 @@
def test_db_connect(db_base):
with pytest.raises(DbException) as excinfo:
db_base.db_connect(None)
- assert str(excinfo.value).startswith(exception_message("Method 'db_connect' not implemented"))
+ assert str(excinfo.value).startswith(
+ exception_message("Method 'db_connect' not implemented")
+ )
def test_db_disconnect(db_base):
@@ -53,42 +55,54 @@
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 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 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 str(excinfo.value).startswith(
+ exception_message("Method 'create' not implemented")
+ )
assert excinfo.value.http_code == http.HTTPStatus.NOT_FOUND
def test_create_list(db_base):
with pytest.raises(DbException) as excinfo:
db_base.create_list(None, None)
- assert str(excinfo.value).startswith(exception_message("Method 'create_list' not implemented"))
+ assert str(excinfo.value).startswith(
+ exception_message("Method 'create_list' 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 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)
- assert str(excinfo.value).startswith(exception_message("Method 'del_one' not implemented"))
+ assert str(excinfo.value).startswith(
+ exception_message("Method 'del_one' not implemented")
+ )
assert excinfo.value.http_code == http.HTTPStatus.NOT_FOUND
@@ -118,17 +132,25 @@
for db_base in self.db_bases:
for value, salt in TEST:
# no encryption
- encrypted = db_base.encrypt(value, schema_version='1.0', salt=salt)
- self.assertEqual(encrypted, value, "value '{}' has been encrypted".format(value))
- decrypted = db_base.decrypt(encrypted, schema_version='1.0', salt=salt)
- self.assertEqual(decrypted, value, "value '{}' has been decrypted".format(value))
+ encrypted = db_base.encrypt(value, schema_version="1.0", salt=salt)
+ self.assertEqual(
+ encrypted, value, "value '{}' has been encrypted".format(value)
+ )
+ decrypted = db_base.decrypt(encrypted, schema_version="1.0", salt=salt)
+ self.assertEqual(
+ decrypted, value, "value '{}' has been decrypted".format(value)
+ )
# encrypt/decrypt
- encrypted = db_base.encrypt(value, schema_version='1.1', salt=salt)
- self.assertNotEqual(encrypted, value, "value '{}' has not been encrypted".format(value))
+ encrypted = db_base.encrypt(value, schema_version="1.1", salt=salt)
+ self.assertNotEqual(
+ encrypted, value, "value '{}' has not been encrypted".format(value)
+ )
self.assertIsInstance(encrypted, str, "Encrypted is not ascii text")
- decrypted = db_base.decrypt(encrypted, schema_version='1.1', salt=salt)
- self.assertEqual(decrypted, value, "value is not equal after encryption/decryption")
+ decrypted = db_base.decrypt(encrypted, schema_version="1.1", salt=salt)
+ self.assertEqual(
+ decrypted, value, "value is not equal after encryption/decryption"
+ )
def test_encrypt_decrypt_salt(self):
value = "value to be encrypted!"
@@ -136,22 +158,44 @@
for db_base in self.db_bases:
for salt in (None, "salt 1", "1afd5d1a-4a7e-4d9c-8c65-251290183106"):
# encrypt/decrypt
- encrypted.append(db_base.encrypt(value, schema_version='1.1', salt=salt))
- self.assertNotEqual(encrypted[-1], value, "value '{}' has not been encrypted".format(value))
+ encrypted.append(
+ db_base.encrypt(value, schema_version="1.1", salt=salt)
+ )
+ self.assertNotEqual(
+ encrypted[-1],
+ value,
+ "value '{}' has not been encrypted".format(value),
+ )
self.assertIsInstance(encrypted[-1], str, "Encrypted is not ascii text")
- decrypted = db_base.decrypt(encrypted[-1], schema_version='1.1', salt=salt)
- self.assertEqual(decrypted, value, "value is not equal after encryption/decryption")
+ decrypted = db_base.decrypt(
+ encrypted[-1], schema_version="1.1", salt=salt
+ )
+ self.assertEqual(
+ decrypted, value, "value is not equal after encryption/decryption"
+ )
for i in range(0, len(encrypted)):
- for j in range(i+1, len(encrypted)):
- self.assertNotEqual(encrypted[i], encrypted[j],
- "encryption with different salt must contain different result")
+ for j in range(i + 1, len(encrypted)):
+ self.assertNotEqual(
+ encrypted[i],
+ encrypted[j],
+ "encryption with different salt must contain different result",
+ )
# decrypt with a different master key
try:
- decrypted = self.db_bases[-1].decrypt(encrypted[0], schema_version='1.1', salt=None)
- self.assertNotEqual(encrypted[0], decrypted, "Decryption with different KEY must generate different result")
+ decrypted = self.db_bases[-1].decrypt(
+ encrypted[0], schema_version="1.1", salt=None
+ )
+ self.assertNotEqual(
+ encrypted[0],
+ decrypted,
+ "Decryption with different KEY must generate different result",
+ )
except DbException as e:
- self.assertEqual(e.http_code, HTTPStatus.INTERNAL_SERVER_ERROR,
- "Decryption with different KEY does not provide expected http_code")
+ self.assertEqual(
+ e.http_code,
+ HTTPStatus.INTERNAL_SERVER_ERROR,
+ "Decryption with different KEY does not provide expected http_code",
+ )
class TestDeepUpdate(unittest.TestCase):
@@ -193,30 +237,62 @@
({"A": ["a", "b", "a"]}, {"A": {"$a": None}}, {"A": ["b"]}),
({"A": ["a", "b", "a"]}, {"A": {"$d": None}}, {"A": ["a", "b", "a"]}),
# delete and insert at 0
- ({"A": ["a", "b", "c"]}, {"A": {"$b": None, "$+[0]": "b"}}, {"A": ["b", "a", "c"]}),
+ (
+ {"A": ["a", "b", "c"]},
+ {"A": {"$b": None, "$+[0]": "b"}},
+ {"A": ["b", "a", "c"]},
+ ),
# delete and edit
- ({"A": ["a", "b", "a"]}, {"A": {"$a": None, "$[1]": {"c": "d"}}}, {"A": [{"c": "d"}]}),
+ (
+ {"A": ["a", "b", "a"]},
+ {"A": {"$a": None, "$[1]": {"c": "d"}}},
+ {"A": [{"c": "d"}]},
+ ),
# insert if not exist
({"A": ["a", "b", "c"]}, {"A": {"$+b": "b"}}, {"A": ["a", "b", "c"]}),
({"A": ["a", "b", "c"]}, {"A": {"$+d": "f"}}, {"A": ["a", "b", "c", "f"]}),
# edit by filter
- ({"A": ["a", "b", "a"]}, {"A": {"$b": {"c": "d"}}}, {"A": ["a", {"c": "d"}, "a"]}),
- ({"A": ["a", "b", "a"]}, {"A": {"$b": None, "$+[0]": "b", "$+": "c"}}, {"A": ["b", "a", "a", "c"]}),
+ (
+ {"A": ["a", "b", "a"]},
+ {"A": {"$b": {"c": "d"}}},
+ {"A": ["a", {"c": "d"}, "a"]},
+ ),
+ (
+ {"A": ["a", "b", "a"]},
+ {"A": {"$b": None, "$+[0]": "b", "$+": "c"}},
+ {"A": ["b", "a", "a", "c"]},
+ ),
({"A": ["a", "b", "a"]}, {"A": {"$c": None}}, {"A": ["a", "b", "a"]}),
# index deletion out of range
({"A": ["a", "b", "a"]}, {"A": {"$[5]": None}}, {"A": ["a", "b", "a"]}),
# nested array->dict
- ({"A": ["a", "b", {"id": "1", "c": {"d": 2}}]}, {"A": {"$id: '1'": {"h": None, "c": {"d": "e", "f": "g"}}}},
- {"A": ["a", "b", {"id": "1", "c": {"d": "e", "f": "g"}}]}),
- ({"A": [{"id": 1, "c": {"d": 2}}, {"id": 1, "c": {"f": []}}]},
- {"A": {"$id: 1": {"h": None, "c": {"d": "e", "f": "g"}}}},
- {"A": [{"id": 1, "c": {"d": "e", "f": "g"}}, {"id": 1, "c": {"d": "e", "f": "g"}}]}),
+ (
+ {"A": ["a", "b", {"id": "1", "c": {"d": 2}}]},
+ {"A": {"$id: '1'": {"h": None, "c": {"d": "e", "f": "g"}}}},
+ {"A": ["a", "b", {"id": "1", "c": {"d": "e", "f": "g"}}]},
+ ),
+ (
+ {"A": [{"id": 1, "c": {"d": 2}}, {"id": 1, "c": {"f": []}}]},
+ {"A": {"$id: 1": {"h": None, "c": {"d": "e", "f": "g"}}}},
+ {
+ "A": [
+ {"id": 1, "c": {"d": "e", "f": "g"}},
+ {"id": 1, "c": {"d": "e", "f": "g"}},
+ ]
+ },
+ ),
# nested array->array
- ({"A": ["a", "b", ["a", "b"]]}, {"A": {"$b": None, "$[2]": {"$b": {}, "$+": "c"}}},
- {"A": ["a", ["a", {}, "c"]]}),
+ (
+ {"A": ["a", "b", ["a", "b"]]},
+ {"A": {"$b": None, "$[2]": {"$b": {}, "$+": "c"}}},
+ {"A": ["a", ["a", {}, "c"]]},
+ ),
# types str and int different, so not found
- ({"A": ["a", {"id": "1", "c": "d"}]}, {"A": {"$id: 1": {"c": "e"}}}, {"A": ["a", {"id": "1", "c": "d"}]}),
-
+ (
+ {"A": ["a", {"id": "1", "c": "d"}]},
+ {"A": {"$id: 1": {"c": "e"}}},
+ {"A": ["a", {"id": "1", "c": "d"}]},
+ ),
)
for t in TEST:
print(t)
@@ -244,7 +320,10 @@
# index edition out of range
({"A": ["a", "b", "a"]}, {"A": {"$[5]": 6}}),
# conflict, two editions on index 2
- ({"A": ["a", {"id": "1", "c": "d"}]}, {"A": {"$id: '1'": {"c": "e"}, "$c: d": {"c": "f"}}}),
+ (
+ {"A": ["a", {"id": "1", "c": "d"}]},
+ {"A": {"$id: '1'": {"c": "e"}, "$c: d": {"c": "f"}}},
+ ),
)
for t in TEST:
print(t)
@@ -255,5 +334,5 @@
print(e)
-if __name__ == '__main__':
+if __name__ == "__main__":
unittest.main()
diff --git a/osm_common/tests/test_dbmemory.py b/osm_common/tests/test_dbmemory.py
index 344d3dd..3c45527 100644
--- a/osm_common/tests/test_dbmemory.py
+++ b/osm_common/tests/test_dbmemory.py
@@ -28,7 +28,7 @@
from osm_common.dbmemory import DbMemory
from copy import deepcopy
-__author__ = 'Eduardo Sousa <eduardosousa@av.it.pt>'
+__author__ = "Eduardo Sousa <eduardosousa@av.it.pt>"
@pytest.fixture(scope="function", params=[True, False])
@@ -52,21 +52,40 @@
def db_memory_with_many_data(request):
db = DbMemory(lock=False)
- db.create_list("test", [
- {"_id": 1, "data": {"data2": {"data3": 1}}, "list": [{"a": 1}], "text": "sometext"},
- {"_id": 2, "data": {"data2": {"data3": 2}}, "list": [{"a": 2}], "list2": [1, 2, 3]},
- {"_id": 3, "data": {"data2": {"data3": 3}}, "list": [{"a": 3}]},
- {"_id": 4, "data": {"data2": {"data3": 4}}, "list": [{"a": 4}, {"a": 0}]},
- {"_id": 5, "data": {"data2": {"data3": 5}}, "list": [{"a": 5}]},
- {"_id": 6, "data": {"data2": {"data3": 6}}, "list": [{"0": {"a": 1}}]},
- {"_id": 7, "data": {"data2": {"data3": 7}}, "0": {"a": 0}},
- {"_id": 8, "list": [{"a": 3, "b": 0, "c": [{"a": 3, "b": 1}, {"a": 0, "b": "v"}]}, {"a": 0, "b": 1}]},
- ])
+ db.create_list(
+ "test",
+ [
+ {
+ "_id": 1,
+ "data": {"data2": {"data3": 1}},
+ "list": [{"a": 1}],
+ "text": "sometext",
+ },
+ {
+ "_id": 2,
+ "data": {"data2": {"data3": 2}},
+ "list": [{"a": 2}],
+ "list2": [1, 2, 3],
+ },
+ {"_id": 3, "data": {"data2": {"data3": 3}}, "list": [{"a": 3}]},
+ {"_id": 4, "data": {"data2": {"data3": 4}}, "list": [{"a": 4}, {"a": 0}]},
+ {"_id": 5, "data": {"data2": {"data3": 5}}, "list": [{"a": 5}]},
+ {"_id": 6, "data": {"data2": {"data3": 6}}, "list": [{"0": {"a": 1}}]},
+ {"_id": 7, "data": {"data2": {"data3": 7}}, "0": {"a": 0}},
+ {
+ "_id": 8,
+ "list": [
+ {"a": 3, "b": 0, "c": [{"a": 3, "b": 1}, {"a": 0, "b": "v"}]},
+ {"a": 0, "b": 1},
+ ],
+ },
+ ],
+ )
return db
def empty_exception_message():
- return 'database exception '
+ return "database exception "
def get_one_exception_message(db_filter):
@@ -74,7 +93,9 @@
def get_one_multiple_exception_message(db_filter):
- return "database exception Found more than one entry with filter='{}'".format(db_filter)
+ return "database exception Found more than one entry with filter='{}'".format(
+ db_filter
+ )
def del_one_exception_message(db_filter):
@@ -87,20 +108,20 @@
def test_constructor():
db = DbMemory()
- assert db.logger == logging.getLogger('db')
+ assert db.logger == logging.getLogger("db")
assert db.db == {}
def test_constructor_with_logger():
- logger_name = 'db_local'
+ logger_name = "db_local"
db = DbMemory(logger_name=logger_name)
assert db.logger == logging.getLogger(logger_name)
assert db.db == {}
def test_db_connect():
- logger_name = 'db_local'
- config = {'logger_name': logger_name}
+ logger_name = "db_local"
+ config = {"logger_name": logger_name}
db = DbMemory()
db.db_connect(config)
assert db.logger == logging.getLogger(logger_name)
@@ -111,32 +132,46 @@
db_memory.db_disconnect()
-@pytest.mark.parametrize("table, db_filter", [
- ("test", {}),
- ("test", {"_id": 1}),
- ("test", {"data": 1}),
- ("test", {"_id": 1, "data": 1})])
+@pytest.mark.parametrize(
+ "table, db_filter",
+ [
+ ("test", {}),
+ ("test", {"_id": 1}),
+ ("test", {"data": 1}),
+ ("test", {"_id": 1, "data": 1}),
+ ],
+)
def test_get_list_with_empty_db(db_memory, table, db_filter):
result = db_memory.get_list(table, db_filter)
assert len(result) == 0
-@pytest.mark.parametrize("table, db_filter, expected_data", [
- ("test", {}, [{"_id": 1, "data": 1}, {"_id": 2, "data": 2}, {"_id": 3, "data": 3}]),
- ("test", {"_id": 1}, [{"_id": 1, "data": 1}]),
- ("test", {"data": 1}, [{"_id": 1, "data": 1}]),
- ("test", {"_id": 1, "data": 1}, [{"_id": 1, "data": 1}]),
- ("test", {"_id": 2}, [{"_id": 2, "data": 2}]),
- ("test", {"data": 2}, [{"_id": 2, "data": 2}]),
- ("test", {"_id": 2, "data": 2}, [{"_id": 2, "data": 2}]),
- ("test", {"_id": 4}, []),
- ("test", {"data": 4}, []),
- ("test", {"_id": 4, "data": 4}, []),
- ("test_table", {}, []),
- ("test_table", {"_id": 1}, []),
- ("test_table", {"data": 1}, []),
- ("test_table", {"_id": 1, "data": 1}, [])])
-def test_get_list_with_non_empty_db(db_memory_with_data, table, db_filter, expected_data):
+@pytest.mark.parametrize(
+ "table, db_filter, expected_data",
+ [
+ (
+ "test",
+ {},
+ [{"_id": 1, "data": 1}, {"_id": 2, "data": 2}, {"_id": 3, "data": 3}],
+ ),
+ ("test", {"_id": 1}, [{"_id": 1, "data": 1}]),
+ ("test", {"data": 1}, [{"_id": 1, "data": 1}]),
+ ("test", {"_id": 1, "data": 1}, [{"_id": 1, "data": 1}]),
+ ("test", {"_id": 2}, [{"_id": 2, "data": 2}]),
+ ("test", {"data": 2}, [{"_id": 2, "data": 2}]),
+ ("test", {"_id": 2, "data": 2}, [{"_id": 2, "data": 2}]),
+ ("test", {"_id": 4}, []),
+ ("test", {"data": 4}, []),
+ ("test", {"_id": 4, "data": 4}, []),
+ ("test_table", {}, []),
+ ("test_table", {"_id": 1}, []),
+ ("test_table", {"data": 1}, []),
+ ("test_table", {"_id": 1, "data": 1}, []),
+ ],
+)
+def test_get_list_with_non_empty_db(
+ db_memory_with_data, table, db_filter, expected_data
+):
result = db_memory_with_data.get_list(table, db_filter)
assert len(result) == len(expected_data)
for data in expected_data:
@@ -144,7 +179,7 @@
def test_get_list_exception(db_memory_with_data):
- table = 'test'
+ table = "test"
db_filter = {}
db_memory_with_data._find = MagicMock(side_effect=Exception())
with pytest.raises(DbException) as excinfo:
@@ -153,16 +188,20 @@
assert excinfo.value.http_code == http.HTTPStatus.NOT_FOUND
-@pytest.mark.parametrize("table, db_filter, expected_data", [
- ("test", {"_id": 1}, {"_id": 1, "data": 1}),
- ("test", {"_id": 2}, {"_id": 2, "data": 2}),
- ("test", {"_id": 3}, {"_id": 3, "data": 3}),
- ("test", {"data": 1}, {"_id": 1, "data": 1}),
- ("test", {"data": 2}, {"_id": 2, "data": 2}),
- ("test", {"data": 3}, {"_id": 3, "data": 3}),
- ("test", {"_id": 1, "data": 1}, {"_id": 1, "data": 1}),
- ("test", {"_id": 2, "data": 2}, {"_id": 2, "data": 2}),
- ("test", {"_id": 3, "data": 3}, {"_id": 3, "data": 3})])
+@pytest.mark.parametrize(
+ "table, db_filter, expected_data",
+ [
+ ("test", {"_id": 1}, {"_id": 1, "data": 1}),
+ ("test", {"_id": 2}, {"_id": 2, "data": 2}),
+ ("test", {"_id": 3}, {"_id": 3, "data": 3}),
+ ("test", {"data": 1}, {"_id": 1, "data": 1}),
+ ("test", {"data": 2}, {"_id": 2, "data": 2}),
+ ("test", {"data": 3}, {"_id": 3, "data": 3}),
+ ("test", {"_id": 1, "data": 1}, {"_id": 1, "data": 1}),
+ ("test", {"_id": 2, "data": 2}, {"_id": 2, "data": 2}),
+ ("test", {"_id": 3, "data": 3}, {"_id": 3, "data": 3}),
+ ],
+)
def test_get_one(db_memory_with_data, table, db_filter, expected_data):
result = db_memory_with_data.get_one(table, db_filter)
assert result == expected_data
@@ -172,63 +211,83 @@
assert result in db_memory_with_data.db[table]
-@pytest.mark.parametrize("db_filter, expected_ids", [
- ({}, [1, 2, 3, 4, 5, 6, 7, 8]),
- ({"_id": 1}, [1]),
- ({"data.data2.data3": 2}, [2]),
- ({"data.data2.data3.eq": 2}, [2]),
- ({"data.data2.data3": [2]}, [2]),
- ({"data.data2.data3.cont": [2]}, [2]),
- ({"data.data2.data3.neq": 2}, [1, 3, 4, 5, 6, 7, 8]),
- ({"data.data2.data3.neq": [2]}, [1, 3, 4, 5, 6, 7, 8]),
- ({"data.data2.data3.ncont": [2]}, [1, 3, 4, 5, 6, 7, 8]),
- ({"data.data2.data3": [2, 3]}, [2, 3]),
- ({"data.data2.data3.gt": 4}, [5, 6, 7]),
- ({"data.data2.data3.gte": 4}, [4, 5, 6, 7]),
- ({"data.data2.data3.lt": 4}, [1, 2, 3]),
- ({"data.data2.data3.lte": 4}, [1, 2, 3, 4]),
- ({"data.data2.data3.lte": 4.5}, [1, 2, 3, 4]),
- ({"data.data2.data3.gt": "text"}, []),
- ({"nonexist.nonexist": "4"}, []),
- ({"nonexist.nonexist": None}, [1, 2, 3, 4, 5, 6, 7, 8]),
- ({"nonexist.nonexist.neq": "4"}, [1, 2, 3, 4, 5, 6, 7, 8]),
- ({"nonexist.nonexist.neq": None}, []),
- ({"text.eq": "sometext"}, [1]),
- ({"text.neq": "sometext"}, [2, 3, 4, 5, 6, 7, 8]),
- ({"text.eq": "somet"}, []),
- ({"text.gte": "a"}, [1]),
- ({"text.gte": "somet"}, [1]),
- ({"text.gte": "sometext"}, [1]),
- ({"text.lt": "somet"}, []),
- ({"data.data2.data3": 2, "data.data2.data4": None}, [2]),
- ({"data.data2.data3": 2, "data.data2.data4": 5}, []),
- ({"data.data2.data3": 4}, [4]),
- ({"data.data2.data3": [3, 4, "e"]}, [3, 4]),
- ({"data.data2.data3": None}, [8]),
- ({"data.data2": "4"}, []),
- ({"list.0.a": 1}, [1, 6]),
- ({"list2": 1}, [2]),
- ({"list2": [1, 5]}, [2]),
- ({"list2": [1, 2]}, [2]),
- ({"list2": [5, 7]}, []),
- ({"list.ANYINDEX.a": 1}, [1]),
- ({"list.a": 3, "list.b": 1}, [8]),
- ({"list.ANYINDEX.a": 3, "list.ANYINDEX.b": 1}, []),
- ({"list.ANYINDEX.a": 3, "list.ANYINDEX.c.a": 3}, [8]),
- ({"list.ANYINDEX.a": 3, "list.ANYINDEX.b": 0}, [8]),
- ({"list.ANYINDEX.a": 3, "list.ANYINDEX.c.ANYINDEX.a": 0, "list.ANYINDEX.c.ANYINDEX.b": "v"}, [8]),
- ({"list.ANYINDEX.a": 3, "list.ANYINDEX.c.ANYINDEX.a": 0, "list.ANYINDEX.c.ANYINDEX.b": 1}, []),
- ({"list.c.b": 1}, [8]),
- ({"list.c.b": None}, [1, 2, 3, 4, 5, 6, 7]),
- # ({"data.data2.data3": 4}, []),
- # ({"data.data2.data3": 4}, []),
-])
+@pytest.mark.parametrize(
+ "db_filter, expected_ids",
+ [
+ ({}, [1, 2, 3, 4, 5, 6, 7, 8]),
+ ({"_id": 1}, [1]),
+ ({"data.data2.data3": 2}, [2]),
+ ({"data.data2.data3.eq": 2}, [2]),
+ ({"data.data2.data3": [2]}, [2]),
+ ({"data.data2.data3.cont": [2]}, [2]),
+ ({"data.data2.data3.neq": 2}, [1, 3, 4, 5, 6, 7, 8]),
+ ({"data.data2.data3.neq": [2]}, [1, 3, 4, 5, 6, 7, 8]),
+ ({"data.data2.data3.ncont": [2]}, [1, 3, 4, 5, 6, 7, 8]),
+ ({"data.data2.data3": [2, 3]}, [2, 3]),
+ ({"data.data2.data3.gt": 4}, [5, 6, 7]),
+ ({"data.data2.data3.gte": 4}, [4, 5, 6, 7]),
+ ({"data.data2.data3.lt": 4}, [1, 2, 3]),
+ ({"data.data2.data3.lte": 4}, [1, 2, 3, 4]),
+ ({"data.data2.data3.lte": 4.5}, [1, 2, 3, 4]),
+ ({"data.data2.data3.gt": "text"}, []),
+ ({"nonexist.nonexist": "4"}, []),
+ ({"nonexist.nonexist": None}, [1, 2, 3, 4, 5, 6, 7, 8]),
+ ({"nonexist.nonexist.neq": "4"}, [1, 2, 3, 4, 5, 6, 7, 8]),
+ ({"nonexist.nonexist.neq": None}, []),
+ ({"text.eq": "sometext"}, [1]),
+ ({"text.neq": "sometext"}, [2, 3, 4, 5, 6, 7, 8]),
+ ({"text.eq": "somet"}, []),
+ ({"text.gte": "a"}, [1]),
+ ({"text.gte": "somet"}, [1]),
+ ({"text.gte": "sometext"}, [1]),
+ ({"text.lt": "somet"}, []),
+ ({"data.data2.data3": 2, "data.data2.data4": None}, [2]),
+ ({"data.data2.data3": 2, "data.data2.data4": 5}, []),
+ ({"data.data2.data3": 4}, [4]),
+ ({"data.data2.data3": [3, 4, "e"]}, [3, 4]),
+ ({"data.data2.data3": None}, [8]),
+ ({"data.data2": "4"}, []),
+ ({"list.0.a": 1}, [1, 6]),
+ ({"list2": 1}, [2]),
+ ({"list2": [1, 5]}, [2]),
+ ({"list2": [1, 2]}, [2]),
+ ({"list2": [5, 7]}, []),
+ ({"list.ANYINDEX.a": 1}, [1]),
+ ({"list.a": 3, "list.b": 1}, [8]),
+ ({"list.ANYINDEX.a": 3, "list.ANYINDEX.b": 1}, []),
+ ({"list.ANYINDEX.a": 3, "list.ANYINDEX.c.a": 3}, [8]),
+ ({"list.ANYINDEX.a": 3, "list.ANYINDEX.b": 0}, [8]),
+ (
+ {
+ "list.ANYINDEX.a": 3,
+ "list.ANYINDEX.c.ANYINDEX.a": 0,
+ "list.ANYINDEX.c.ANYINDEX.b": "v",
+ },
+ [8],
+ ),
+ (
+ {
+ "list.ANYINDEX.a": 3,
+ "list.ANYINDEX.c.ANYINDEX.a": 0,
+ "list.ANYINDEX.c.ANYINDEX.b": 1,
+ },
+ [],
+ ),
+ ({"list.c.b": 1}, [8]),
+ ({"list.c.b": None}, [1, 2, 3, 4, 5, 6, 7]),
+ # ({"data.data2.data3": 4}, []),
+ # ({"data.data2.data3": 4}, []),
+ ],
+)
def test_get_list(db_memory_with_many_data, db_filter, expected_ids):
result = db_memory_with_many_data.get_list("test", db_filter)
assert isinstance(result, list)
result_ids = [item["_id"] for item in result]
- assert len(result) == len(expected_ids), "for db_filter={} result={} expected_ids={}".format(db_filter, result,
- result_ids)
+ assert len(result) == len(
+ expected_ids
+ ), "for db_filter={} result={} expected_ids={}".format(
+ db_filter, result, result_ids
+ )
assert result_ids == expected_ids
for i in range(len(result)):
assert result[i] in db_memory_with_many_data.db["test"]
@@ -240,9 +299,12 @@
assert result == len(expected_ids)
-@pytest.mark.parametrize("table, db_filter, expected_data", [
- ("test", {}, {"_id": 1, "data": 1})])
-def test_get_one_with_multiple_results(db_memory_with_data, table, db_filter, expected_data):
+@pytest.mark.parametrize(
+ "table, db_filter, expected_data", [("test", {}, {"_id": 1, "data": 1})]
+)
+def test_get_one_with_multiple_results(
+ db_memory_with_data, table, db_filter, expected_data
+):
result = db_memory_with_data.get_one(table, db_filter, fail_on_more=False)
assert result == expected_data
assert len(db_memory_with_data.db) == 1
@@ -256,64 +318,86 @@
db_filter = {}
with pytest.raises(DbException) as excinfo:
db_memory_with_data.get_one(table, db_filter)
- assert str(excinfo.value) == (empty_exception_message() + get_one_multiple_exception_message(db_filter))
+ assert str(excinfo.value) == (
+ empty_exception_message() + get_one_multiple_exception_message(db_filter)
+ )
# assert excinfo.value.http_code == http.HTTPStatus.CONFLICT
-@pytest.mark.parametrize("table, db_filter", [
- ("test", {"_id": 4}),
- ("test", {"data": 4}),
- ("test", {"_id": 4, "data": 4}),
- ("test_table", {"_id": 4}),
- ("test_table", {"data": 4}),
- ("test_table", {"_id": 4, "data": 4})])
+@pytest.mark.parametrize(
+ "table, db_filter",
+ [
+ ("test", {"_id": 4}),
+ ("test", {"data": 4}),
+ ("test", {"_id": 4, "data": 4}),
+ ("test_table", {"_id": 4}),
+ ("test_table", {"data": 4}),
+ ("test_table", {"_id": 4, "data": 4}),
+ ],
+)
def test_get_one_with_non_empty_db_exception(db_memory_with_data, table, db_filter):
with pytest.raises(DbException) as excinfo:
db_memory_with_data.get_one(table, db_filter)
- assert str(excinfo.value) == (empty_exception_message() + get_one_exception_message(db_filter))
+ assert str(excinfo.value) == (
+ empty_exception_message() + get_one_exception_message(db_filter)
+ )
assert excinfo.value.http_code == http.HTTPStatus.NOT_FOUND
-@pytest.mark.parametrize("table, db_filter", [
- ("test", {"_id": 4}),
- ("test", {"data": 4}),
- ("test", {"_id": 4, "data": 4}),
- ("test_table", {"_id": 4}),
- ("test_table", {"data": 4}),
- ("test_table", {"_id": 4, "data": 4})])
+@pytest.mark.parametrize(
+ "table, db_filter",
+ [
+ ("test", {"_id": 4}),
+ ("test", {"data": 4}),
+ ("test", {"_id": 4, "data": 4}),
+ ("test_table", {"_id": 4}),
+ ("test_table", {"data": 4}),
+ ("test_table", {"_id": 4, "data": 4}),
+ ],
+)
def test_get_one_with_non_empty_db_none(db_memory_with_data, table, db_filter):
result = db_memory_with_data.get_one(table, db_filter, fail_on_empty=False)
assert result is None
-@pytest.mark.parametrize("table, db_filter", [
- ("test", {"_id": 4}),
- ("test", {"data": 4}),
- ("test", {"_id": 4, "data": 4}),
- ("test_table", {"_id": 4}),
- ("test_table", {"data": 4}),
- ("test_table", {"_id": 4, "data": 4})])
+@pytest.mark.parametrize(
+ "table, db_filter",
+ [
+ ("test", {"_id": 4}),
+ ("test", {"data": 4}),
+ ("test", {"_id": 4, "data": 4}),
+ ("test_table", {"_id": 4}),
+ ("test_table", {"data": 4}),
+ ("test_table", {"_id": 4, "data": 4}),
+ ],
+)
def test_get_one_with_empty_db_exception(db_memory, table, db_filter):
with pytest.raises(DbException) as excinfo:
db_memory.get_one(table, db_filter)
- assert str(excinfo.value) == (empty_exception_message() + get_one_exception_message(db_filter))
+ assert str(excinfo.value) == (
+ empty_exception_message() + get_one_exception_message(db_filter)
+ )
assert excinfo.value.http_code == http.HTTPStatus.NOT_FOUND
-@pytest.mark.parametrize("table, db_filter", [
- ("test", {"_id": 4}),
- ("test", {"data": 4}),
- ("test", {"_id": 4, "data": 4}),
- ("test_table", {"_id": 4}),
- ("test_table", {"data": 4}),
- ("test_table", {"_id": 4, "data": 4})])
+@pytest.mark.parametrize(
+ "table, db_filter",
+ [
+ ("test", {"_id": 4}),
+ ("test", {"data": 4}),
+ ("test", {"_id": 4, "data": 4}),
+ ("test_table", {"_id": 4}),
+ ("test_table", {"data": 4}),
+ ("test_table", {"_id": 4, "data": 4}),
+ ],
+)
def test_get_one_with_empty_db_none(db_memory, table, db_filter):
result = db_memory.get_one(table, db_filter, fail_on_empty=False)
assert result is None
def test_get_one_generic_exception(db_memory_with_data):
- table = 'test'
+ table = "test"
db_filter = {}
db_memory_with_data._find = MagicMock(side_effect=Exception())
with pytest.raises(DbException) as excinfo:
@@ -322,13 +406,19 @@
assert excinfo.value.http_code == http.HTTPStatus.NOT_FOUND
-@pytest.mark.parametrize("table, db_filter, expected_data", [
- ("test", {}, []),
- ("test", {"_id": 1}, [{"_id": 2, "data": 2}, {"_id": 3, "data": 3}]),
- ("test", {"_id": 2}, [{"_id": 1, "data": 1}, {"_id": 3, "data": 3}]),
- ("test", {"_id": 1, "data": 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, db_filter, expected_data):
+@pytest.mark.parametrize(
+ "table, db_filter, expected_data",
+ [
+ ("test", {}, []),
+ ("test", {"_id": 1}, [{"_id": 2, "data": 2}, {"_id": 3, "data": 3}]),
+ ("test", {"_id": 2}, [{"_id": 1, "data": 1}, {"_id": 3, "data": 3}]),
+ ("test", {"_id": 1, "data": 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, db_filter, expected_data
+):
result = db_memory_with_data.del_list(table, db_filter)
assert result["deleted"] == (3 - len(expected_data))
assert len(db_memory_with_data.db) == 1
@@ -338,21 +428,25 @@
assert data in db_memory_with_data.db[table]
-@pytest.mark.parametrize("table, db_filter", [
- ("test", {}),
- ("test", {"_id": 1}),
- ("test", {"_id": 2}),
- ("test", {"data": 1}),
- ("test", {"data": 2}),
- ("test", {"_id": 1, "data": 1}),
- ("test", {"_id": 2, "data": 2})])
+@pytest.mark.parametrize(
+ "table, db_filter",
+ [
+ ("test", {}),
+ ("test", {"_id": 1}),
+ ("test", {"_id": 2}),
+ ("test", {"data": 1}),
+ ("test", {"data": 2}),
+ ("test", {"_id": 1, "data": 1}),
+ ("test", {"_id": 2, "data": 2}),
+ ],
+)
def test_del_list_with_empty_db(db_memory, table, db_filter):
result = db_memory.del_list(table, db_filter)
- assert result['deleted'] == 0
+ assert result["deleted"] == 0
def test_del_list_generic_exception(db_memory_with_data):
- table = 'test'
+ table = "test"
db_filter = {}
db_memory_with_data._find = MagicMock(side_effect=Exception())
with pytest.raises(DbException) as excinfo:
@@ -361,14 +455,18 @@
assert excinfo.value.http_code == http.HTTPStatus.NOT_FOUND
-@pytest.mark.parametrize("table, db_filter, data", [
- ("test", {}, {"_id": 1, "data": 1}),
- ("test", {"_id": 1}, {"_id": 1, "data": 1}),
- ("test", {"data": 1}, {"_id": 1, "data": 1}),
- ("test", {"_id": 1, "data": 1}, {"_id": 1, "data": 1}),
- ("test", {"_id": 2}, {"_id": 2, "data": 2}),
- ("test", {"data": 2}, {"_id": 2, "data": 2}),
- ("test", {"_id": 2, "data": 2}, {"_id": 2, "data": 2})])
+@pytest.mark.parametrize(
+ "table, db_filter, data",
+ [
+ ("test", {}, {"_id": 1, "data": 1}),
+ ("test", {"_id": 1}, {"_id": 1, "data": 1}),
+ ("test", {"data": 1}, {"_id": 1, "data": 1}),
+ ("test", {"_id": 1, "data": 1}, {"_id": 1, "data": 1}),
+ ("test", {"_id": 2}, {"_id": 2, "data": 2}),
+ ("test", {"data": 2}, {"_id": 2, "data": 2}),
+ ("test", {"_id": 2, "data": 2}, {"_id": 2, "data": 2}),
+ ],
+)
def test_del_one(db_memory_with_data, table, db_filter, data):
result = db_memory_with_data.del_one(table, db_filter)
assert result == {"deleted": 1}
@@ -378,93 +476,111 @@
assert data not in db_memory_with_data.db[table]
-@pytest.mark.parametrize("table, db_filter", [
- ("test", {}),
- ("test", {"_id": 1}),
- ("test", {"_id": 2}),
- ("test", {"data": 1}),
- ("test", {"data": 2}),
- ("test", {"_id": 1, "data": 1}),
- ("test", {"_id": 2, "data": 2}),
- ("test_table", {}),
- ("test_table", {"_id": 1}),
- ("test_table", {"_id": 2}),
- ("test_table", {"data": 1}),
- ("test_table", {"data": 2}),
- ("test_table", {"_id": 1, "data": 1}),
- ("test_table", {"_id": 2, "data": 2})])
+@pytest.mark.parametrize(
+ "table, db_filter",
+ [
+ ("test", {}),
+ ("test", {"_id": 1}),
+ ("test", {"_id": 2}),
+ ("test", {"data": 1}),
+ ("test", {"data": 2}),
+ ("test", {"_id": 1, "data": 1}),
+ ("test", {"_id": 2, "data": 2}),
+ ("test_table", {}),
+ ("test_table", {"_id": 1}),
+ ("test_table", {"_id": 2}),
+ ("test_table", {"data": 1}),
+ ("test_table", {"data": 2}),
+ ("test_table", {"_id": 1, "data": 1}),
+ ("test_table", {"_id": 2, "data": 2}),
+ ],
+)
def test_del_one_with_empty_db_exception(db_memory, table, db_filter):
with pytest.raises(DbException) as excinfo:
db_memory.del_one(table, db_filter)
- assert str(excinfo.value) == (empty_exception_message() + del_one_exception_message(db_filter))
+ assert str(excinfo.value) == (
+ empty_exception_message() + del_one_exception_message(db_filter)
+ )
assert excinfo.value.http_code == http.HTTPStatus.NOT_FOUND
-@pytest.mark.parametrize("table, db_filter", [
- ("test", {}),
- ("test", {"_id": 1}),
- ("test", {"_id": 2}),
- ("test", {"data": 1}),
- ("test", {"data": 2}),
- ("test", {"_id": 1, "data": 1}),
- ("test", {"_id": 2, "data": 2}),
- ("test_table", {}),
- ("test_table", {"_id": 1}),
- ("test_table", {"_id": 2}),
- ("test_table", {"data": 1}),
- ("test_table", {"data": 2}),
- ("test_table", {"_id": 1, "data": 1}),
- ("test_table", {"_id": 2, "data": 2})])
+@pytest.mark.parametrize(
+ "table, db_filter",
+ [
+ ("test", {}),
+ ("test", {"_id": 1}),
+ ("test", {"_id": 2}),
+ ("test", {"data": 1}),
+ ("test", {"data": 2}),
+ ("test", {"_id": 1, "data": 1}),
+ ("test", {"_id": 2, "data": 2}),
+ ("test_table", {}),
+ ("test_table", {"_id": 1}),
+ ("test_table", {"_id": 2}),
+ ("test_table", {"data": 1}),
+ ("test_table", {"data": 2}),
+ ("test_table", {"_id": 1, "data": 1}),
+ ("test_table", {"_id": 2, "data": 2}),
+ ],
+)
def test_del_one_with_empty_db_none(db_memory, table, db_filter):
result = db_memory.del_one(table, db_filter, fail_on_empty=False)
assert result is None
-@pytest.mark.parametrize("table, db_filter", [
- ("test", {"_id": 4}),
- ("test", {"_id": 5}),
- ("test", {"data": 4}),
- ("test", {"data": 5}),
- ("test", {"_id": 1, "data": 2}),
- ("test", {"_id": 2, "data": 3}),
- ("test_table", {}),
- ("test_table", {"_id": 1}),
- ("test_table", {"_id": 2}),
- ("test_table", {"data": 1}),
- ("test_table", {"data": 2}),
- ("test_table", {"_id": 1, "data": 1}),
- ("test_table", {"_id": 2, "data": 2})])
+@pytest.mark.parametrize(
+ "table, db_filter",
+ [
+ ("test", {"_id": 4}),
+ ("test", {"_id": 5}),
+ ("test", {"data": 4}),
+ ("test", {"data": 5}),
+ ("test", {"_id": 1, "data": 2}),
+ ("test", {"_id": 2, "data": 3}),
+ ("test_table", {}),
+ ("test_table", {"_id": 1}),
+ ("test_table", {"_id": 2}),
+ ("test_table", {"data": 1}),
+ ("test_table", {"data": 2}),
+ ("test_table", {"_id": 1, "data": 1}),
+ ("test_table", {"_id": 2, "data": 2}),
+ ],
+)
def test_del_one_with_non_empty_db_exception(db_memory_with_data, table, db_filter):
with pytest.raises(DbException) as excinfo:
db_memory_with_data.del_one(table, db_filter)
- assert str(excinfo.value) == (empty_exception_message() + del_one_exception_message(db_filter))
+ assert str(excinfo.value) == (
+ empty_exception_message() + del_one_exception_message(db_filter)
+ )
assert excinfo.value.http_code == http.HTTPStatus.NOT_FOUND
-@pytest.mark.parametrize("table, db_filter", [
- ("test", {"_id": 4}),
- ("test", {"_id": 5}),
- ("test", {"data": 4}),
- ("test", {"data": 5}),
- ("test", {"_id": 1, "data": 2}),
- ("test", {"_id": 2, "data": 3}),
- ("test_table", {}),
- ("test_table", {"_id": 1}),
- ("test_table", {"_id": 2}),
- ("test_table", {"data": 1}),
- ("test_table", {"data": 2}),
- ("test_table", {"_id": 1, "data": 1}),
- ("test_table", {"_id": 2, "data": 2})])
+@pytest.mark.parametrize(
+ "table, db_filter",
+ [
+ ("test", {"_id": 4}),
+ ("test", {"_id": 5}),
+ ("test", {"data": 4}),
+ ("test", {"data": 5}),
+ ("test", {"_id": 1, "data": 2}),
+ ("test", {"_id": 2, "data": 3}),
+ ("test_table", {}),
+ ("test_table", {"_id": 1}),
+ ("test_table", {"_id": 2}),
+ ("test_table", {"data": 1}),
+ ("test_table", {"data": 2}),
+ ("test_table", {"_id": 1, "data": 1}),
+ ("test_table", {"_id": 2, "data": 2}),
+ ],
+)
def test_del_one_with_non_empty_db_none(db_memory_with_data, table, db_filter):
result = db_memory_with_data.del_one(table, db_filter, fail_on_empty=False)
assert result is None
-@pytest.mark.parametrize("fail_on_empty", [
- (True),
- (False)])
+@pytest.mark.parametrize("fail_on_empty", [(True), (False)])
def test_del_one_generic_exception(db_memory_with_data, fail_on_empty):
- table = 'test'
+ table = "test"
db_filter = {}
db_memory_with_data._find = MagicMock(side_effect=Exception())
with pytest.raises(DbException) as excinfo:
@@ -473,16 +589,20 @@
assert excinfo.value.http_code == http.HTTPStatus.NOT_FOUND
-@pytest.mark.parametrize("table, _id, indata", [
- ("test", 1, {"_id": 1, "data": 42}),
- ("test", 1, {"_id": 1, "data": 42, "kk": 34}),
- ("test", 1, {"_id": 1}),
- ("test", 2, {"_id": 2, "data": 42}),
- ("test", 2, {"_id": 2, "data": 42, "kk": 34}),
- ("test", 2, {"_id": 2}),
- ("test", 3, {"_id": 3, "data": 42}),
- ("test", 3, {"_id": 3, "data": 42, "kk": 34}),
- ("test", 3, {"_id": 3})])
+@pytest.mark.parametrize(
+ "table, _id, indata",
+ [
+ ("test", 1, {"_id": 1, "data": 42}),
+ ("test", 1, {"_id": 1, "data": 42, "kk": 34}),
+ ("test", 1, {"_id": 1}),
+ ("test", 2, {"_id": 2, "data": 42}),
+ ("test", 2, {"_id": 2, "data": 42, "kk": 34}),
+ ("test", 2, {"_id": 2}),
+ ("test", 3, {"_id": 3, "data": 42}),
+ ("test", 3, {"_id": 3, "data": 42, "kk": 34}),
+ ("test", 3, {"_id": 3}),
+ ],
+)
def test_replace(db_memory_with_data, table, _id, indata):
result = db_memory_with_data.replace(table, _id, indata)
assert result == {"updated": 1}
@@ -492,10 +612,14 @@
assert indata in db_memory_with_data.db[table]
-@pytest.mark.parametrize("table, _id, indata", [
- ("test", 1, {"_id": 1, "data": 42}),
- ("test", 2, {"_id": 2}),
- ("test", 3, {"_id": 3})])
+@pytest.mark.parametrize(
+ "table, _id, indata",
+ [
+ ("test", 1, {"_id": 1, "data": 42}),
+ ("test", 2, {"_id": 2}),
+ ("test", 3, {"_id": 3}),
+ ],
+)
def test_replace_without_data_exception(db_memory, table, _id, indata):
with pytest.raises(DbException) as excinfo:
db_memory.replace(table, _id, indata, fail_on_empty=True)
@@ -503,19 +627,27 @@
assert excinfo.value.http_code == http.HTTPStatus.NOT_FOUND
-@pytest.mark.parametrize("table, _id, indata", [
- ("test", 1, {"_id": 1, "data": 42}),
- ("test", 2, {"_id": 2}),
- ("test", 3, {"_id": 3})])
+@pytest.mark.parametrize(
+ "table, _id, indata",
+ [
+ ("test", 1, {"_id": 1, "data": 42}),
+ ("test", 2, {"_id": 2}),
+ ("test", 3, {"_id": 3}),
+ ],
+)
def test_replace_without_data_none(db_memory, table, _id, indata):
result = db_memory.replace(table, _id, indata, fail_on_empty=False)
assert result is None
-@pytest.mark.parametrize("table, _id, indata", [
- ("test", 11, {"_id": 11, "data": 42}),
- ("test", 12, {"_id": 12}),
- ("test", 33, {"_id": 33})])
+@pytest.mark.parametrize(
+ "table, _id, indata",
+ [
+ ("test", 11, {"_id": 11, "data": 42}),
+ ("test", 12, {"_id": 12}),
+ ("test", 33, {"_id": 33}),
+ ],
+)
def test_replace_with_data_exception(db_memory_with_data, table, _id, indata):
with pytest.raises(DbException) as excinfo:
db_memory_with_data.replace(table, _id, indata, fail_on_empty=True)
@@ -523,22 +655,24 @@
assert excinfo.value.http_code == http.HTTPStatus.NOT_FOUND
-@pytest.mark.parametrize("table, _id, indata", [
- ("test", 11, {"_id": 11, "data": 42}),
- ("test", 12, {"_id": 12}),
- ("test", 33, {"_id": 33})])
+@pytest.mark.parametrize(
+ "table, _id, indata",
+ [
+ ("test", 11, {"_id": 11, "data": 42}),
+ ("test", 12, {"_id": 12}),
+ ("test", 33, {"_id": 33}),
+ ],
+)
def test_replace_with_data_none(db_memory_with_data, table, _id, indata):
result = db_memory_with_data.replace(table, _id, indata, fail_on_empty=False)
assert result is None
-@pytest.mark.parametrize("fail_on_empty", [
- True,
- False])
+@pytest.mark.parametrize("fail_on_empty", [True, False])
def test_replace_generic_exception(db_memory_with_data, fail_on_empty):
- table = 'test'
+ table = "test"
_id = {}
- indata = {'_id': 1, 'data': 1}
+ 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, _id, indata, fail_on_empty=fail_on_empty)
@@ -546,26 +680,30 @@
assert excinfo.value.http_code == http.HTTPStatus.NOT_FOUND
-@pytest.mark.parametrize("table, id, data", [
- ("test", "1", {"data": 1}),
- ("test", "1", {"data": 2}),
- ("test", "2", {"data": 1}),
- ("test", "2", {"data": 2}),
- ("test_table", "1", {"data": 1}),
- ("test_table", "1", {"data": 2}),
- ("test_table", "2", {"data": 1}),
- ("test_table", "2", {"data": 2}),
- ("test", "1", {"data_1": 1, "data_2": 2}),
- ("test", "1", {"data_1": 2, "data_2": 1}),
- ("test", "2", {"data_1": 1, "data_2": 2}),
- ("test", "2", {"data_1": 2, "data_2": 1}),
- ("test_table", "1", {"data_1": 1, "data_2": 2}),
- ("test_table", "1", {"data_1": 2, "data_2": 1}),
- ("test_table", "2", {"data_1": 1, "data_2": 2}),
- ("test_table", "2", {"data_1": 2, "data_2": 1})])
+@pytest.mark.parametrize(
+ "table, id, data",
+ [
+ ("test", "1", {"data": 1}),
+ ("test", "1", {"data": 2}),
+ ("test", "2", {"data": 1}),
+ ("test", "2", {"data": 2}),
+ ("test_table", "1", {"data": 1}),
+ ("test_table", "1", {"data": 2}),
+ ("test_table", "2", {"data": 1}),
+ ("test_table", "2", {"data": 2}),
+ ("test", "1", {"data_1": 1, "data_2": 2}),
+ ("test", "1", {"data_1": 2, "data_2": 1}),
+ ("test", "2", {"data_1": 1, "data_2": 2}),
+ ("test", "2", {"data_1": 2, "data_2": 1}),
+ ("test_table", "1", {"data_1": 1, "data_2": 2}),
+ ("test_table", "1", {"data_1": 2, "data_2": 1}),
+ ("test_table", "2", {"data_1": 1, "data_2": 2}),
+ ("test_table", "2", {"data_1": 2, "data_2": 1}),
+ ],
+)
def test_create_with_empty_db_with_id(db_memory, table, id, data):
data_to_insert = data
- data_to_insert['_id'] = id
+ data_to_insert["_id"] = id
returned_id = db_memory.create(table, data_to_insert)
assert returned_id == id
assert len(db_memory.db) == 1
@@ -574,85 +712,97 @@
assert data_to_insert in db_memory.db[table]
-@pytest.mark.parametrize("table, id, data", [
- ("test", "4", {"data": 1}),
- ("test", "5", {"data": 2}),
- ("test", "4", {"data": 1}),
- ("test", "5", {"data": 2}),
- ("test_table", "4", {"data": 1}),
- ("test_table", "5", {"data": 2}),
- ("test_table", "4", {"data": 1}),
- ("test_table", "5", {"data": 2}),
- ("test", "4", {"data_1": 1, "data_2": 2}),
- ("test", "5", {"data_1": 2, "data_2": 1}),
- ("test", "4", {"data_1": 1, "data_2": 2}),
- ("test", "5", {"data_1": 2, "data_2": 1}),
- ("test_table", "4", {"data_1": 1, "data_2": 2}),
- ("test_table", "5", {"data_1": 2, "data_2": 1}),
- ("test_table", "4", {"data_1": 1, "data_2": 2}),
- ("test_table", "5", {"data_1": 2, "data_2": 1})])
+@pytest.mark.parametrize(
+ "table, id, data",
+ [
+ ("test", "4", {"data": 1}),
+ ("test", "5", {"data": 2}),
+ ("test", "4", {"data": 1}),
+ ("test", "5", {"data": 2}),
+ ("test_table", "4", {"data": 1}),
+ ("test_table", "5", {"data": 2}),
+ ("test_table", "4", {"data": 1}),
+ ("test_table", "5", {"data": 2}),
+ ("test", "4", {"data_1": 1, "data_2": 2}),
+ ("test", "5", {"data_1": 2, "data_2": 1}),
+ ("test", "4", {"data_1": 1, "data_2": 2}),
+ ("test", "5", {"data_1": 2, "data_2": 1}),
+ ("test_table", "4", {"data_1": 1, "data_2": 2}),
+ ("test_table", "5", {"data_1": 2, "data_2": 1}),
+ ("test_table", "4", {"data_1": 1, "data_2": 2}),
+ ("test_table", "5", {"data_1": 2, "data_2": 1}),
+ ],
+)
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
+ 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 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 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", {"data": 1}),
- ("test", {"data": 2}),
- ("test_table", {"data": 1}),
- ("test_table", {"data": 2}),
- ("test_table", {"data": 1}),
- ("test_table", {"data": 2}),
- ("test", {"data_1": 1, "data_2": 2}),
- ("test", {"data_1": 2, "data_2": 1}),
- ("test", {"data_1": 1, "data_2": 2}),
- ("test", {"data_1": 2, "data_2": 1}),
- ("test_table", {"data_1": 1, "data_2": 2}),
- ("test_table", {"data_1": 2, "data_2": 1}),
- ("test_table", {"data_1": 1, "data_2": 2}),
- ("test_table", {"data_1": 2, "data_2": 1})])
+@pytest.mark.parametrize(
+ "table, data",
+ [
+ ("test", {"data": 1}),
+ ("test", {"data": 2}),
+ ("test", {"data": 1}),
+ ("test", {"data": 2}),
+ ("test_table", {"data": 1}),
+ ("test_table", {"data": 2}),
+ ("test_table", {"data": 1}),
+ ("test_table", {"data": 2}),
+ ("test", {"data_1": 1, "data_2": 2}),
+ ("test", {"data_1": 2, "data_2": 1}),
+ ("test", {"data_1": 1, "data_2": 2}),
+ ("test", {"data_1": 2, "data_2": 1}),
+ ("test_table", {"data_1": 1, "data_2": 2}),
+ ("test_table", {"data_1": 2, "data_2": 1}),
+ ("test_table", {"data_1": 1, "data_2": 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
+ 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", {"data": 1}),
- ("test", {"data": 2}),
- ("test_table", {"data": 1}),
- ("test_table", {"data": 2}),
- ("test_table", {"data": 1}),
- ("test_table", {"data": 2}),
- ("test", {"data_1": 1, "data_2": 2}),
- ("test", {"data_1": 2, "data_2": 1}),
- ("test", {"data_1": 1, "data_2": 2}),
- ("test", {"data_1": 2, "data_2": 1}),
- ("test_table", {"data_1": 1, "data_2": 2}),
- ("test_table", {"data_1": 2, "data_2": 1}),
- ("test_table", {"data_1": 1, "data_2": 2}),
- ("test_table", {"data_1": 2, "data_2": 1})])
+@pytest.mark.parametrize(
+ "table, data",
+ [
+ ("test", {"data": 1}),
+ ("test", {"data": 2}),
+ ("test", {"data": 1}),
+ ("test", {"data": 2}),
+ ("test_table", {"data": 1}),
+ ("test_table", {"data": 2}),
+ ("test_table", {"data": 1}),
+ ("test_table", {"data": 2}),
+ ("test", {"data_1": 1, "data_2": 2}),
+ ("test", {"data_1": 2, "data_2": 1}),
+ ("test", {"data_1": 1, "data_2": 2}),
+ ("test", {"data_1": 2, "data_2": 1}),
+ ("test_table", {"data_1": 1, "data_2": 2}),
+ ("test_table", {"data_1": 2, "data_2": 1}),
+ ("test_table", {"data_1": 1, "data_2": 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 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 len(db_memory_with_data.db[table]) == (4 if table == "test" else 1)
data_inserted = data
- data_inserted['_id'] = returned_id
+ data_inserted["_id"] = returned_id
assert data_inserted in db_memory_with_data.db[table]
@@ -667,29 +817,82 @@
assert excinfo.value.http_code == http.HTTPStatus.NOT_FOUND
-@pytest.mark.parametrize("db_content, update_dict, expected, message", [
- ({"a": {"none": None}}, {"a.b.num": "v"}, {"a": {"none": None, "b": {"num": "v"}}}, "create dict"),
- ({"a": {"none": None}}, {"a.none.num": "v"}, {"a": {"none": {"num": "v"}}}, "create dict over none"),
- ({"a": {"b": {"num": 4}}}, {"a.b.num": "v"}, {"a": {"b": {"num": "v"}}}, "replace_number"),
- ({"a": {"b": {"num": 4}}}, {"a.b.num.c.d": "v"}, None, "create dict over number should fail"),
- ({"a": {"b": {"num": 4}}}, {"a.b": "v"}, {"a": {"b": "v"}}, "replace dict with a string"),
- ({"a": {"b": {"num": 4}}}, {"a.b": None}, {"a": {"b": None}}, "replace dict with None"),
- ({"a": [{"b": {"num": 4}}]}, {"a.b.num": "v"}, None, "create dict over list should fail"),
- ({"a": [{"b": {"num": 4}}]}, {"a.0.b.num": "v"}, {"a": [{"b": {"num": "v"}}]}, "set list"),
- ({"a": [{"b": {"num": 4}}]}, {"a.3.b.num": "v"},
- {"a": [{"b": {"num": 4}}, None, None, {"b": {"num": "v"}}]}, "expand list"),
- ({"a": [[4]]}, {"a.0.0": "v"}, {"a": [["v"]]}, "set nested list"),
- ({"a": [[4]]}, {"a.0.2": "v"}, {"a": [[4, None, "v"]]}, "expand nested list"),
- ({"a": [[4]]}, {"a.2.2": "v"}, {"a": [[4], None, {"2": "v"}]}, "expand list and add number key")])
+@pytest.mark.parametrize(
+ "db_content, update_dict, expected, message",
+ [
+ (
+ {"a": {"none": None}},
+ {"a.b.num": "v"},
+ {"a": {"none": None, "b": {"num": "v"}}},
+ "create dict",
+ ),
+ (
+ {"a": {"none": None}},
+ {"a.none.num": "v"},
+ {"a": {"none": {"num": "v"}}},
+ "create dict over none",
+ ),
+ (
+ {"a": {"b": {"num": 4}}},
+ {"a.b.num": "v"},
+ {"a": {"b": {"num": "v"}}},
+ "replace_number",
+ ),
+ (
+ {"a": {"b": {"num": 4}}},
+ {"a.b.num.c.d": "v"},
+ None,
+ "create dict over number should fail",
+ ),
+ (
+ {"a": {"b": {"num": 4}}},
+ {"a.b": "v"},
+ {"a": {"b": "v"}},
+ "replace dict with a string",
+ ),
+ (
+ {"a": {"b": {"num": 4}}},
+ {"a.b": None},
+ {"a": {"b": None}},
+ "replace dict with None",
+ ),
+ (
+ {"a": [{"b": {"num": 4}}]},
+ {"a.b.num": "v"},
+ None,
+ "create dict over list should fail",
+ ),
+ (
+ {"a": [{"b": {"num": 4}}]},
+ {"a.0.b.num": "v"},
+ {"a": [{"b": {"num": "v"}}]},
+ "set list",
+ ),
+ (
+ {"a": [{"b": {"num": 4}}]},
+ {"a.3.b.num": "v"},
+ {"a": [{"b": {"num": 4}}, None, None, {"b": {"num": "v"}}]},
+ "expand list",
+ ),
+ ({"a": [[4]]}, {"a.0.0": "v"}, {"a": [["v"]]}, "set nested list"),
+ ({"a": [[4]]}, {"a.0.2": "v"}, {"a": [[4, None, "v"]]}, "expand nested list"),
+ (
+ {"a": [[4]]},
+ {"a.2.2": "v"},
+ {"a": [[4], None, {"2": "v"}]},
+ "expand list and add number key",
+ ),
+ ],
+)
def test_set_one(db_memory, db_content, update_dict, expected, message):
- db_memory._find = Mock(return_value=((0, db_content), ))
+ db_memory._find = Mock(return_value=((0, db_content),))
if expected is None:
with pytest.raises(DbException) as excinfo:
db_memory.set_one("table", {}, update_dict)
- assert (excinfo.value.http_code == http.HTTPStatus.NOT_FOUND), message
+ assert excinfo.value.http_code == http.HTTPStatus.NOT_FOUND, message
else:
db_memory.set_one("table", {}, update_dict)
- assert (db_content == expected), message
+ assert db_content == expected, message
class TestDbMemory(unittest.TestCase):
@@ -697,26 +900,79 @@
def test_set_one(self):
test_set = (
# (database content, set-content, expected database content (None=fails), message)
- ({"a": {"none": None}}, {"a.b.num": "v"}, {"a": {"none": None, "b": {"num": "v"}}}, "create dict"),
- ({"a": {"none": None}}, {"a.none.num": "v"}, {"a": {"none": {"num": "v"}}}, "create dict over none"),
- ({"a": {"b": {"num": 4}}}, {"a.b.num": "v"}, {"a": {"b": {"num": "v"}}}, "replace_number"),
- ({"a": {"b": {"num": 4}}}, {"a.b.num.c.d": "v"}, None, "create dict over number should fail"),
- ({"a": {"b": {"num": 4}}}, {"a.b": "v"}, {"a": {"b": "v"}}, "replace dict with a string"),
- ({"a": {"b": {"num": 4}}}, {"a.b": None}, {"a": {"b": None}}, "replace dict with None"),
-
- ({"a": [{"b": {"num": 4}}]}, {"a.b.num": "v"}, None, "create dict over list should fail"),
- ({"a": [{"b": {"num": 4}}]}, {"a.0.b.num": "v"}, {"a": [{"b": {"num": "v"}}]}, "set list"),
- ({"a": [{"b": {"num": 4}}]}, {"a.3.b.num": "v"},
- {"a": [{"b": {"num": 4}}, None, None, {"b": {"num": "v"}}]}, "expand list"),
+ (
+ {"a": {"none": None}},
+ {"a.b.num": "v"},
+ {"a": {"none": None, "b": {"num": "v"}}},
+ "create dict",
+ ),
+ (
+ {"a": {"none": None}},
+ {"a.none.num": "v"},
+ {"a": {"none": {"num": "v"}}},
+ "create dict over none",
+ ),
+ (
+ {"a": {"b": {"num": 4}}},
+ {"a.b.num": "v"},
+ {"a": {"b": {"num": "v"}}},
+ "replace_number",
+ ),
+ (
+ {"a": {"b": {"num": 4}}},
+ {"a.b.num.c.d": "v"},
+ None,
+ "create dict over number should fail",
+ ),
+ (
+ {"a": {"b": {"num": 4}}},
+ {"a.b": "v"},
+ {"a": {"b": "v"}},
+ "replace dict with a string",
+ ),
+ (
+ {"a": {"b": {"num": 4}}},
+ {"a.b": None},
+ {"a": {"b": None}},
+ "replace dict with None",
+ ),
+ (
+ {"a": [{"b": {"num": 4}}]},
+ {"a.b.num": "v"},
+ None,
+ "create dict over list should fail",
+ ),
+ (
+ {"a": [{"b": {"num": 4}}]},
+ {"a.0.b.num": "v"},
+ {"a": [{"b": {"num": "v"}}]},
+ "set list",
+ ),
+ (
+ {"a": [{"b": {"num": 4}}]},
+ {"a.3.b.num": "v"},
+ {"a": [{"b": {"num": 4}}, None, None, {"b": {"num": "v"}}]},
+ "expand list",
+ ),
({"a": [[4]]}, {"a.0.0": "v"}, {"a": [["v"]]}, "set nested list"),
- ({"a": [[4]]}, {"a.0.2": "v"}, {"a": [[4, None, "v"]]}, "expand nested list"),
- ({"a": [[4]]}, {"a.2.2": "v"}, {"a": [[4], None, {"2": "v"}]}, "expand list and add number key"),
+ (
+ {"a": [[4]]},
+ {"a.0.2": "v"},
+ {"a": [[4, None, "v"]]},
+ "expand nested list",
+ ),
+ (
+ {"a": [[4]]},
+ {"a.2.2": "v"},
+ {"a": [[4], None, {"2": "v"}]},
+ "expand list and add number key",
+ ),
({"a": None}, {"b.c": "v"}, {"a": None, "b": {"c": "v"}}, "expand at root"),
)
db_men = DbMemory()
db_men._find = Mock()
for db_content, update_dict, expected, message in test_set:
- db_men._find.return_value = ((0, db_content), )
+ db_men._find.return_value = ((0, db_content),)
if expected is None:
self.assertRaises(DbException, db_men.set_one, "table", {}, update_dict)
else:
@@ -740,9 +996,17 @@
db_men._find = Mock()
for db_content, pull_dict, expected, message in test_set:
db_content = deepcopy(db_content)
- db_men._find.return_value = ((0, db_content), )
+ db_men._find.return_value = ((0, db_content),)
if expected is None:
- self.assertRaises(DbException, db_men.set_one, "table", {}, None, fail_on_empty=False, pull=pull_dict)
+ self.assertRaises(
+ DbException,
+ db_men.set_one,
+ "table",
+ {},
+ None,
+ fail_on_empty=False,
+ pull=pull_dict,
+ )
else:
db_men.set_one("table", {}, None, pull=pull_dict)
self.assertEqual(db_content, expected, message)
@@ -751,12 +1015,36 @@
example = {"a": [1, "1", 1], "d": {}, "n": None}
test_set = (
# (database content, set-content, expected database content (None=fails), message)
- (example, {"d.b.c": 1}, {"a": [1, "1", 1], "d": {"b": {"c": [1]}}, "n": None}, "push non existing arrray2"),
- (example, {"b": 1}, {"a": [1, "1", 1], "d": {}, "b": [1], "n": None}, "push non existing arrray3"),
- (example, {"a.6": 1}, {"a": [1, "1", 1, None, None, None, [1]], "d": {}, "n": None},
- "push non existing arrray"),
- (example, {"a": 2}, {"a": [1, "1", 1, 2], "d": {}, "n": None}, "push one item"),
- (example, {"a": {1: 1}}, {"a": [1, "1", 1, {1: 1}], "d": {}, "n": None}, "push a dict"),
+ (
+ example,
+ {"d.b.c": 1},
+ {"a": [1, "1", 1], "d": {"b": {"c": [1]}}, "n": None},
+ "push non existing arrray2",
+ ),
+ (
+ example,
+ {"b": 1},
+ {"a": [1, "1", 1], "d": {}, "b": [1], "n": None},
+ "push non existing arrray3",
+ ),
+ (
+ example,
+ {"a.6": 1},
+ {"a": [1, "1", 1, None, None, None, [1]], "d": {}, "n": None},
+ "push non existing arrray",
+ ),
+ (
+ example,
+ {"a": 2},
+ {"a": [1, "1", 1, 2], "d": {}, "n": None},
+ "push one item",
+ ),
+ (
+ example,
+ {"a": {1: 1}},
+ {"a": [1, "1", 1, {1: 1}], "d": {}, "n": None},
+ "push a dict",
+ ),
(example, {"d": 1}, None, "push over dict"),
(example, {"n": 1}, None, "push over None"),
)
@@ -764,9 +1052,17 @@
db_men._find = Mock()
for db_content, push_dict, expected, message in test_set:
db_content = deepcopy(db_content)
- db_men._find.return_value = ((0, db_content), )
+ db_men._find.return_value = ((0, db_content),)
if expected is None:
- self.assertRaises(DbException, db_men.set_one, "table", {}, None, fail_on_empty=False, push=push_dict)
+ self.assertRaises(
+ DbException,
+ db_men.set_one,
+ "table",
+ {},
+ None,
+ fail_on_empty=False,
+ push=push_dict,
+ )
else:
db_men.set_one("table", {}, None, push=push_dict)
self.assertEqual(db_content, expected, message)
@@ -775,13 +1071,36 @@
example = {"a": [1, "1", 1], "d": {}, "n": None}
test_set = (
# (database content, set-content, expected database content (None=fails), message)
- (example, {"d.b.c": [1]}, {"a": [1, "1", 1], "d": {"b": {"c": [1]}}, "n": None},
- "push non existing arrray2"),
- (example, {"b": [1]}, {"a": [1, "1", 1], "d": {}, "b": [1], "n": None}, "push non existing arrray3"),
- (example, {"a.6": [1]}, {"a": [1, "1", 1, None, None, None, [1]], "d": {}, "n": None},
- "push non existing arrray"),
- (example, {"a": [2, 3]}, {"a": [1, "1", 1, 2, 3], "d": {}, "n": None}, "push two item"),
- (example, {"a": [{1: 1}]}, {"a": [1, "1", 1, {1: 1}], "d": {}, "n": None}, "push a dict"),
+ (
+ example,
+ {"d.b.c": [1]},
+ {"a": [1, "1", 1], "d": {"b": {"c": [1]}}, "n": None},
+ "push non existing arrray2",
+ ),
+ (
+ example,
+ {"b": [1]},
+ {"a": [1, "1", 1], "d": {}, "b": [1], "n": None},
+ "push non existing arrray3",
+ ),
+ (
+ example,
+ {"a.6": [1]},
+ {"a": [1, "1", 1, None, None, None, [1]], "d": {}, "n": None},
+ "push non existing arrray",
+ ),
+ (
+ example,
+ {"a": [2, 3]},
+ {"a": [1, "1", 1, 2, 3], "d": {}, "n": None},
+ "push two item",
+ ),
+ (
+ example,
+ {"a": [{1: 1}]},
+ {"a": [1, "1", 1, {1: 1}], "d": {}, "n": None},
+ "push a dict",
+ ),
(example, {"d": [1]}, None, "push over dict"),
(example, {"n": [1]}, None, "push over None"),
(example, {"a": 1}, None, "invalid push list non an array"),
@@ -790,10 +1109,17 @@
db_men._find = Mock()
for db_content, push_list, expected, message in test_set:
db_content = deepcopy(db_content)
- db_men._find.return_value = ((0, db_content), )
+ db_men._find.return_value = ((0, db_content),)
if expected is None:
- self.assertRaises(DbException, db_men.set_one, "table", {}, None, fail_on_empty=False,
- push_list=push_list)
+ self.assertRaises(
+ DbException,
+ db_men.set_one,
+ "table",
+ {},
+ None,
+ fail_on_empty=False,
+ push_list=push_list,
+ )
else:
db_men.set_one("table", {}, None, push_list=push_list)
self.assertEqual(db_content, expected, message)
@@ -813,9 +1139,17 @@
db_men._find = Mock()
for db_content, unset_dict, expected, message in test_set:
db_content = deepcopy(db_content)
- db_men._find.return_value = ((0, db_content), )
+ db_men._find.return_value = ((0, db_content),)
if expected is None:
- self.assertRaises(DbException, db_men.set_one, "table", {}, None, fail_on_empty=False, unset=unset_dict)
+ self.assertRaises(
+ DbException,
+ db_men.set_one,
+ "table",
+ {},
+ None,
+ fail_on_empty=False,
+ unset=unset_dict,
+ )
else:
db_men.set_one("table", {}, None, unset=unset_dict)
self.assertEqual(db_content, expected, message)
diff --git a/osm_common/tests/test_fsbase.py b/osm_common/tests/test_fsbase.py
index 4decad0..43349ad 100644
--- a/osm_common/tests/test_fsbase.py
+++ b/osm_common/tests/test_fsbase.py
@@ -55,40 +55,52 @@
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 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 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 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 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 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)
- assert str(excinfo.value).startswith(exception_message("Method 'file_delete' not implemented"))
+ assert str(excinfo.value).startswith(
+ exception_message("Method 'file_delete' not implemented")
+ )
assert excinfo.value.http_code == http.HTTPStatus.INTERNAL_SERVER_ERROR
diff --git a/osm_common/tests/test_fslocal.py b/osm_common/tests/test_fslocal.py
index 7416cfb..6f52984 100644
--- a/osm_common/tests/test_fslocal.py
+++ b/osm_common/tests/test_fslocal.py
@@ -34,41 +34,43 @@
def valid_path():
- return tempfile.gettempdir() + '/'
+ return tempfile.gettempdir() + "/"
def invalid_path():
- return '/#tweeter/'
+ return "/#tweeter/"
@pytest.fixture(scope="function", params=[True, False])
def fs_local(request):
fs = FsLocal(lock=request.param)
- fs.fs_connect({'path': valid_path()})
+ 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)
+ 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)
+ 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)
+ 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)
+ 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)
+ f = storage if isinstance(storage, str) else "/".join(storage)
return "storage exception File {} cannot be opened".format(f)
@@ -78,12 +80,12 @@
def test_constructor_without_logger():
fs = FsLocal()
- assert fs.logger == logging.getLogger('fs')
+ assert fs.logger == logging.getLogger("fs")
assert fs.path is None
def test_constructor_with_logger():
- logger_name = 'fs_local'
+ logger_name = "fs_local"
fs = FsLocal(logger_name=logger_name)
assert fs.logger == logging.getLogger(logger_name)
assert fs.path is None
@@ -98,11 +100,19 @@
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()),
- ({'path': valid_path()}, 'fs', valid_path()),
- ({'path': valid_path()[:-1]}, 'fs', 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(),
+ ),
+ ({"path": valid_path()}, "fs", valid_path()),
+ ({"path": valid_path()[:-1]}, "fs", valid_path()),
+ ],
+)
def test_fs_connect_with_valid_config(config, exp_logger, exp_path):
fs = FsLocal()
fs.fs_connect(config)
@@ -110,11 +120,24 @@
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()}, fs_connect_exception_message(invalid_path())),
- ({'path': invalid_path()[:-1]}, fs_connect_exception_message(invalid_path()[:-1]))])
+@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()}, fs_connect_exception_message(invalid_path())),
+ (
+ {"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:
@@ -144,65 +167,89 @@
assert excinfo.value.http_code == http.HTTPStatus.INTERNAL_SERVER_ERROR
-@pytest.mark.parametrize("storage, mode, expected", [
- (str(uuid.uuid4()), 'file', False),
- ([str(uuid.uuid4())], 'file', False),
- (str(uuid.uuid4()), 'dir', False),
- ([str(uuid.uuid4())], 'dir', False)])
+@pytest.mark.parametrize(
+ "storage, mode, expected",
+ [
+ (str(uuid.uuid4()), "file", False),
+ ([str(uuid.uuid4())], "file", False),
+ (str(uuid.uuid4()), "dir", False),
+ ([str(uuid.uuid4())], "dir", 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),
- ([str(uuid.uuid4())], 'dir', True)])
+@pytest.mark.parametrize(
+ "storage, mode, expected",
+ [
+ (str(uuid.uuid4()), "file", True),
+ ([str(uuid.uuid4())], "file", True),
+ (str(uuid.uuid4()), "dir", 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':
+ path = (
+ valid_path() + storage
+ if isinstance(storage, str)
+ else valid_path() + storage[0]
+ )
+ if mode == "file":
os.mknod(path)
- elif mode == 'dir':
+ elif mode == "dir":
os.mkdir(path)
assert fs_local.file_exists(storage, mode) == expected
- if mode == 'file':
+ if mode == "file":
os.remove(path)
- elif mode == 'dir':
+ elif mode == "dir":
os.rmdir(path)
-@pytest.mark.parametrize("storage, mode", [
- (str(uuid.uuid4()), 'file'),
- ([str(uuid.uuid4())], 'file'),
- (str(uuid.uuid4()), 'dir'),
- ([str(uuid.uuid4())], 'dir')])
+@pytest.mark.parametrize(
+ "storage, mode",
+ [
+ (str(uuid.uuid4()), "file"),
+ ([str(uuid.uuid4())], "file"),
+ (str(uuid.uuid4()), "dir"),
+ ([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':
+ path = (
+ valid_path() + storage
+ if isinstance(storage, str)
+ else valid_path() + storage[0]
+ )
+ if mode == "file":
os.mknod(path)
- elif mode == 'dir':
+ elif mode == "dir":
os.mkdir(path)
size = os.path.getsize(path)
assert fs_local.file_size(storage) == size
- if mode == 'file':
+ if mode == "file":
os.remove(path)
- elif mode == 'dir':
+ elif mode == "dir":
os.rmdir(path)
-@pytest.mark.parametrize("files, path", [
- (['foo', 'bar', 'foobar'], str(uuid.uuid4())),
- (['foo', 'bar', 'foobar'], [str(uuid.uuid4())])])
+@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:
+ 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:
+ 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))
+ 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:
@@ -213,21 +260,29 @@
shutil.rmtree(extracted_path)
-@pytest.mark.parametrize("storage, mode", [
- (str(uuid.uuid4()), 'r'),
- (str(uuid.uuid4()), 'w'),
- (str(uuid.uuid4()), 'a'),
- (str(uuid.uuid4()), 'rb'),
- (str(uuid.uuid4()), 'wb'),
- (str(uuid.uuid4()), 'ab'),
- ([str(uuid.uuid4())], 'r'),
- ([str(uuid.uuid4())], 'w'),
- ([str(uuid.uuid4())], 'a'),
- ([str(uuid.uuid4())], 'rb'),
- ([str(uuid.uuid4())], 'wb'),
- ([str(uuid.uuid4())], 'ab')])
+@pytest.mark.parametrize(
+ "storage, mode",
+ [
+ (str(uuid.uuid4()), "r"),
+ (str(uuid.uuid4()), "w"),
+ (str(uuid.uuid4()), "a"),
+ (str(uuid.uuid4()), "rb"),
+ (str(uuid.uuid4()), "wb"),
+ (str(uuid.uuid4()), "ab"),
+ ([str(uuid.uuid4())], "r"),
+ ([str(uuid.uuid4())], "w"),
+ ([str(uuid.uuid4())], "a"),
+ ([str(uuid.uuid4())], "rb"),
+ ([str(uuid.uuid4())], "wb"),
+ ([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]
+ 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)
@@ -235,11 +290,15 @@
os.remove(path)
-@pytest.mark.parametrize("storage, mode", [
- (str(uuid.uuid4()), 'r'),
- (str(uuid.uuid4()), 'rb'),
- ([str(uuid.uuid4())], 'r'),
- ([str(uuid.uuid4())], 'rb')])
+@pytest.mark.parametrize(
+ "storage, mode",
+ [
+ (str(uuid.uuid4()), "r"),
+ (str(uuid.uuid4()), "rb"),
+ ([str(uuid.uuid4())], "r"),
+ ([str(uuid.uuid4())], "rb"),
+ ],
+)
def test_file_open_file_not_found_exception(fs_local, storage, mode):
with pytest.raises(FsException) as excinfo:
fs_local.file_open(storage, mode)
@@ -247,21 +306,29 @@
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()), 'a'),
- (str(uuid.uuid4()), 'rb'),
- (str(uuid.uuid4()), 'wb'),
- (str(uuid.uuid4()), 'ab'),
- ([str(uuid.uuid4())], 'r'),
- ([str(uuid.uuid4())], 'w'),
- ([str(uuid.uuid4())], 'a'),
- ([str(uuid.uuid4())], 'rb'),
- ([str(uuid.uuid4())], 'wb'),
- ([str(uuid.uuid4())], 'ab')])
+@pytest.mark.parametrize(
+ "storage, mode",
+ [
+ (str(uuid.uuid4()), "r"),
+ (str(uuid.uuid4()), "w"),
+ (str(uuid.uuid4()), "a"),
+ (str(uuid.uuid4()), "rb"),
+ (str(uuid.uuid4()), "wb"),
+ (str(uuid.uuid4()), "ab"),
+ ([str(uuid.uuid4())], "r"),
+ ([str(uuid.uuid4())], "w"),
+ ([str(uuid.uuid4())], "a"),
+ ([str(uuid.uuid4())], "rb"),
+ ([str(uuid.uuid4())], "wb"),
+ ([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]
+ 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:
@@ -271,17 +338,25 @@
os.remove(path)
-@pytest.mark.parametrize("storage, with_files", [
- (str(uuid.uuid4()), True),
- (str(uuid.uuid4()), False),
- ([str(uuid.uuid4())], True),
- ([str(uuid.uuid4())], False)])
+@pytest.mark.parametrize(
+ "storage, with_files",
+ [
+ (str(uuid.uuid4()), True),
+ (str(uuid.uuid4()), False),
+ ([str(uuid.uuid4())], True),
+ ([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]
+ path = (
+ valid_path() + storage
+ if isinstance(storage, str)
+ else valid_path() + storage[0]
+ )
os.mkdir(path)
if with_files is True:
file_name = str(uuid.uuid4())
- file_path = path + '/' + file_name
+ file_path = path + "/" + file_name
os.mknod(file_path)
result = fs_local.dir_ls(storage)
@@ -293,11 +368,13 @@
shutil.rmtree(path)
-@pytest.mark.parametrize("storage", [
- (str(uuid.uuid4())),
- ([str(uuid.uuid4())])])
+@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]
+ 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)
@@ -306,11 +383,13 @@
os.remove(path)
-@pytest.mark.parametrize("storage", [
- (str(uuid.uuid4())),
- ([str(uuid.uuid4())])])
+@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]
+ 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:
@@ -320,28 +399,34 @@
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()), True, False),
- (str(uuid.uuid4()), False, False),
- ([str(uuid.uuid4())], True, True),
- ([str(uuid.uuid4())], False, True),
- ([str(uuid.uuid4())], True, False),
- ([str(uuid.uuid4())], False, False)])
+@pytest.mark.parametrize(
+ "storage, with_files, ignore_non_exist",
+ [
+ (str(uuid.uuid4()), True, True),
+ (str(uuid.uuid4()), False, True),
+ (str(uuid.uuid4()), True, False),
+ (str(uuid.uuid4()), False, False),
+ ([str(uuid.uuid4())], True, True),
+ ([str(uuid.uuid4())], False, True),
+ ([str(uuid.uuid4())], True, False),
+ ([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]
+ path = (
+ valid_path() + storage
+ if isinstance(storage, str)
+ else valid_path() + storage[0]
+ )
os.mkdir(path)
if with_files is True:
- file_path = path + '/' + str(uuid.uuid4())
+ 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
-@pytest.mark.parametrize("storage", [
- (str(uuid.uuid4())),
- ([str(uuid.uuid4())])])
+@pytest.mark.parametrize("storage", [(str(uuid.uuid4())), ([str(uuid.uuid4())])])
def test_file_delete_expect_exception(fs_local, storage):
with pytest.raises(FsException) as excinfo:
fs_local.file_delete(storage)
@@ -349,10 +434,12 @@
assert excinfo.value.http_code == http.HTTPStatus.NOT_FOUND
-@pytest.mark.parametrize("storage", [
- (str(uuid.uuid4())),
- ([str(uuid.uuid4())])])
+@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]
+ 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) is False
diff --git a/osm_common/tests/test_fsmongo.py b/osm_common/tests/test_fsmongo.py
index 3b62569..6a8ec58 100644
--- a/osm_common/tests/test_fsmongo.py
+++ b/osm_common/tests/test_fsmongo.py
@@ -38,11 +38,11 @@
def valid_path():
- return tempfile.gettempdir() + '/'
+ return tempfile.gettempdir() + "/"
def invalid_path():
- return '/#tweeter/'
+ return "/#tweeter/"
@pytest.fixture(scope="function", params=[True, False])
@@ -56,15 +56,13 @@
def mock_gridfs_constructor(a, b):
pass
- monkeypatch.setattr(MongoClient, '__init__', mock_mongoclient_constructor)
- monkeypatch.setattr(MongoClient, '__getitem__', mock_mongoclient_getitem)
- monkeypatch.setattr(GridFSBucket, '__init__', mock_gridfs_constructor)
+ monkeypatch.setattr(MongoClient, "__init__", mock_mongoclient_constructor)
+ monkeypatch.setattr(MongoClient, "__getitem__", mock_mongoclient_getitem)
+ monkeypatch.setattr(GridFSBucket, "__init__", mock_gridfs_constructor)
fs = FsMongo(lock=request.param)
- fs.fs_connect({
- 'path': valid_path(),
- 'host': 'mongo',
- 'port': 27017,
- 'collection': 'files'})
+ fs.fs_connect(
+ {"path": valid_path(), "host": "mongo", "port": 27017, "collection": "files"}
+ )
return fs
@@ -73,26 +71,28 @@
def fs_connect_exception_message(path):
- return "storage exception Invalid configuration param at '[storage]': path '{}' does not exist".format(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)
+ 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)
+ 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)
+ 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)
+ f = storage if isinstance(storage, str) else "/".join(storage)
return "storage exception File {} cannot be opened".format(f)
@@ -102,14 +102,14 @@
def test_constructor_without_logger():
fs = FsMongo()
- assert fs.logger == logging.getLogger('fs')
+ assert fs.logger == logging.getLogger("fs")
assert fs.path is None
assert fs.client is None
assert fs.fs is None
def test_constructor_with_logger():
- logger_name = 'fs_mongo'
+ logger_name = "fs_mongo"
fs = FsMongo(logger_name=logger_name)
assert fs.logger == logging.getLogger(logger_name)
assert fs.path is None
@@ -121,7 +121,7 @@
def mock_gridfs_find(self, search_query, **kwargs):
return []
- monkeypatch.setattr(GridFSBucket, 'find', mock_gridfs_find)
+ monkeypatch.setattr(GridFSBucket, "find", mock_gridfs_find)
params = fs_mongo.get_params()
assert len(params) == 2
assert "fs" in params
@@ -130,79 +130,83 @@
assert params["path"] == valid_path()
-@pytest.mark.parametrize("config, exp_logger, exp_path", [
- (
- {
- 'logger_name': 'fs_mongo',
- 'path': valid_path(),
- 'uri': 'mongo:27017',
- 'collection': 'files'
- },
- 'fs_mongo', valid_path()
- ),
- (
- {
- 'logger_name': 'fs_mongo',
- 'path': valid_path(),
- 'host': 'mongo',
- 'port': 27017,
- 'collection': 'files'
- },
- 'fs_mongo', valid_path()
- ),
- (
- {
- 'logger_name': 'fs_mongo',
- 'path': valid_path()[:-1],
- 'uri': 'mongo:27017',
- 'collection': 'files'
- },
- 'fs_mongo', valid_path()
- ),
- (
- {
- 'logger_name': 'fs_mongo',
- 'path': valid_path()[:-1],
- 'host': 'mongo',
- 'port': 27017,
- 'collection': 'files'
- },
- 'fs_mongo', valid_path()
- ),
- (
- {
- 'path': valid_path(),
- 'uri': 'mongo:27017',
- 'collection': 'files'
- },
- 'fs', valid_path()
- ),
- (
- {
- 'path': valid_path(),
- 'host': 'mongo',
- 'port': 27017,
- 'collection': 'files'
- },
- 'fs', valid_path()
- ),
- (
- {
- 'path': valid_path()[:-1],
- 'uri': 'mongo:27017',
- 'collection': 'files'
- },
- 'fs', valid_path()
- ),
- (
- {
- 'path': valid_path()[:-1],
- 'host': 'mongo',
- 'port': 27017,
- 'collection': 'files'
- },
- 'fs', valid_path()
- )])
+@pytest.mark.parametrize(
+ "config, exp_logger, exp_path",
+ [
+ (
+ {
+ "logger_name": "fs_mongo",
+ "path": valid_path(),
+ "uri": "mongo:27017",
+ "collection": "files",
+ },
+ "fs_mongo",
+ valid_path(),
+ ),
+ (
+ {
+ "logger_name": "fs_mongo",
+ "path": valid_path(),
+ "host": "mongo",
+ "port": 27017,
+ "collection": "files",
+ },
+ "fs_mongo",
+ valid_path(),
+ ),
+ (
+ {
+ "logger_name": "fs_mongo",
+ "path": valid_path()[:-1],
+ "uri": "mongo:27017",
+ "collection": "files",
+ },
+ "fs_mongo",
+ valid_path(),
+ ),
+ (
+ {
+ "logger_name": "fs_mongo",
+ "path": valid_path()[:-1],
+ "host": "mongo",
+ "port": 27017,
+ "collection": "files",
+ },
+ "fs_mongo",
+ valid_path(),
+ ),
+ (
+ {"path": valid_path(), "uri": "mongo:27017", "collection": "files"},
+ "fs",
+ valid_path(),
+ ),
+ (
+ {
+ "path": valid_path(),
+ "host": "mongo",
+ "port": 27017,
+ "collection": "files",
+ },
+ "fs",
+ valid_path(),
+ ),
+ (
+ {"path": valid_path()[:-1], "uri": "mongo:27017", "collection": "files"},
+ "fs",
+ valid_path(),
+ ),
+ (
+ {
+ "path": valid_path()[:-1],
+ "host": "mongo",
+ "port": 27017,
+ "collection": "files",
+ },
+ "fs",
+ valid_path(),
+ ),
+ ],
+)
def test_fs_connect_with_valid_config(config, exp_logger, exp_path):
fs = FsMongo()
fs.fs_connect(config)
@@ -212,90 +216,81 @@
assert type(fs.fs) == GridFSBucket
-@pytest.mark.parametrize("config, exp_exception_message", [
- (
- {
- 'logger_name': 'fs_mongo',
- 'path': invalid_path(),
- 'uri': 'mongo:27017',
- 'collection': 'files'
- },
- fs_connect_exception_message(invalid_path())
- ),
- (
- {
- 'logger_name': 'fs_mongo',
- 'path': invalid_path(),
- 'host': 'mongo',
- 'port': 27017,
- 'collection': 'files'
- },
- fs_connect_exception_message(invalid_path())
- ),
- (
- {
- 'logger_name': 'fs_mongo',
- 'path': invalid_path()[:-1],
- 'uri': 'mongo:27017',
- 'collection': 'files'
- },
- fs_connect_exception_message(invalid_path()[:-1])
- ),
- (
- {
- 'logger_name': 'fs_mongo',
- 'path': invalid_path()[:-1],
- 'host': 'mongo',
- 'port': 27017,
- 'collection': 'files'
- },
- fs_connect_exception_message(invalid_path()[:-1])
- ),
- (
- {
- 'path': invalid_path(),
- 'uri': 'mongo:27017',
- 'collection': 'files'
- },
- fs_connect_exception_message(invalid_path())
- ),
- (
- {
- 'path': invalid_path(),
- 'host': 'mongo',
- 'port': 27017,
- 'collection': 'files'
- },
- fs_connect_exception_message(invalid_path())
- ),
- (
- {
- 'path': invalid_path()[:-1],
- 'uri': 'mongo:27017',
- 'collection': 'files'
- },
- fs_connect_exception_message(invalid_path()[:-1])
- ),
- (
- {
- 'path': invalid_path()[:-1],
- 'host': 'mongo',
- 'port': 27017,
- 'collection': 'files'
- },
- fs_connect_exception_message(invalid_path()[:-1])
- ),
- (
- {
- 'path': '/',
- 'host': 'mongo',
- 'port': 27017,
- 'collection': 'files'
- },
- generic_fs_exception_message(
- "Invalid configuration param at '[storage]': path '/' is not writable"
- )
- )])
+@pytest.mark.parametrize(
+ "config, exp_exception_message",
+ [
+ (
+ {
+ "logger_name": "fs_mongo",
+ "path": invalid_path(),
+ "uri": "mongo:27017",
+ "collection": "files",
+ },
+ fs_connect_exception_message(invalid_path()),
+ ),
+ (
+ {
+ "logger_name": "fs_mongo",
+ "path": invalid_path(),
+ "host": "mongo",
+ "port": 27017,
+ "collection": "files",
+ },
+ fs_connect_exception_message(invalid_path()),
+ ),
+ (
+ {
+ "logger_name": "fs_mongo",
+ "path": invalid_path()[:-1],
+ "uri": "mongo:27017",
+ "collection": "files",
+ },
+ fs_connect_exception_message(invalid_path()[:-1]),
+ ),
+ (
+ {
+ "logger_name": "fs_mongo",
+ "path": invalid_path()[:-1],
+ "host": "mongo",
+ "port": 27017,
+ "collection": "files",
+ },
+ fs_connect_exception_message(invalid_path()[:-1]),
+ ),
+ (
+ {"path": invalid_path(), "uri": "mongo:27017", "collection": "files"},
+ fs_connect_exception_message(invalid_path()),
+ ),
+ (
+ {
+ "path": invalid_path(),
+ "host": "mongo",
+ "port": 27017,
+ "collection": "files",
+ },
+ fs_connect_exception_message(invalid_path()),
+ ),
+ (
+ {"path": invalid_path()[:-1], "uri": "mongo:27017", "collection": "files"},
+ fs_connect_exception_message(invalid_path()[:-1]),
+ ),
+ (
+ {
+ "path": invalid_path()[:-1],
+ "host": "mongo",
+ "port": 27017,
+ "collection": "files",
+ },
+ fs_connect_exception_message(invalid_path()[:-1]),
+ ),
+ (
+ {"path": "/", "host": "mongo", "port": 27017, "collection": "files"},
+ generic_fs_exception_message(
+ "Invalid configuration param at '[storage]': path '/' is not writable"
+ ),
+ ),
+ ],
+)
def test_fs_connect_with_invalid_path(config, exp_exception_message):
fs = FsMongo()
with pytest.raises(FsException) as excinfo:
@@ -303,67 +298,59 @@
assert str(excinfo.value) == exp_exception_message
-@pytest.mark.parametrize("config, exp_exception_message", [
- (
- {
- 'logger_name': 'fs_mongo',
- 'uri': 'mongo:27017',
- 'collection': 'files'
- },
- "Missing parameter \"path\""
- ),
- (
- {
- 'logger_name': 'fs_mongo',
- 'host': 'mongo',
- 'port': 27017,
- 'collection': 'files'
- },
- "Missing parameter \"path\""
- ),
- (
- {
- 'logger_name': 'fs_mongo',
- 'path': valid_path(),
- 'collection': 'files'
- },
- "Missing parameters: \"uri\" or \"host\" + \"port\""
- ),
- (
- {
- 'logger_name': 'fs_mongo',
- 'path': valid_path(),
- 'port': 27017,
- 'collection': 'files'
- },
- "Missing parameters: \"uri\" or \"host\" + \"port\""
- ),
- (
- {
- 'logger_name': 'fs_mongo',
- 'path': valid_path(),
- 'host': 'mongo',
- 'collection': 'files'
- },
- "Missing parameters: \"uri\" or \"host\" + \"port\""
- ),
- (
- {
- 'logger_name': 'fs_mongo',
- 'path': valid_path(),
- 'uri': 'mongo:27017'
- },
- "Missing parameter \"collection\""
- ),
- (
- {
- 'logger_name': 'fs_mongo',
- 'path': valid_path(),
- 'host': 'mongo',
- 'port': 27017,
- },
- "Missing parameter \"collection\""
- )])
+@pytest.mark.parametrize(
+ "config, exp_exception_message",
+ [
+ (
+ {"logger_name": "fs_mongo", "uri": "mongo:27017", "collection": "files"},
+ 'Missing parameter "path"',
+ ),
+ (
+ {
+ "logger_name": "fs_mongo",
+ "host": "mongo",
+ "port": 27017,
+ "collection": "files",
+ },
+ 'Missing parameter "path"',
+ ),
+ (
+ {"logger_name": "fs_mongo", "path": valid_path(), "collection": "files"},
+ 'Missing parameters: "uri" or "host" + "port"',
+ ),
+ (
+ {
+ "logger_name": "fs_mongo",
+ "path": valid_path(),
+ "port": 27017,
+ "collection": "files",
+ },
+ 'Missing parameters: "uri" or "host" + "port"',
+ ),
+ (
+ {
+ "logger_name": "fs_mongo",
+ "path": valid_path(),
+ "host": "mongo",
+ "collection": "files",
+ },
+ 'Missing parameters: "uri" or "host" + "port"',
+ ),
+ (
+ {"logger_name": "fs_mongo", "path": valid_path(), "uri": "mongo:27017"},
+ 'Missing parameter "collection"',
+ ),
+ (
+ {
+ "logger_name": "fs_mongo",
+ "path": valid_path(),
+ "host": "mongo",
+ "port": 27017,
+ },
+ 'Missing parameter "collection"',
+ ),
+ ],
+)
def test_fs_connect_with_missing_parameters(config, exp_exception_message):
fs = FsMongo()
with pytest.raises(FsException) as excinfo:
@@ -371,31 +358,37 @@
assert str(excinfo.value) == generic_fs_exception_message(exp_exception_message)
-@pytest.mark.parametrize("config, exp_exception_message", [
- (
- {
- 'logger_name': 'fs_mongo',
- 'path': valid_path(),
- 'uri': 'mongo:27017',
- 'collection': 'files'
- },
- "MongoClient crashed"
- ),
- (
- {
- 'logger_name': 'fs_mongo',
- 'path': valid_path(),
- 'host': 'mongo',
- 'port': 27017,
- 'collection': 'files'
- },
- "MongoClient crashed"
- )])
-def test_fs_connect_with_invalid_mongoclient(config, exp_exception_message, monkeypatch):
+@pytest.mark.parametrize(
+ "config, exp_exception_message",
+ [
+ (
+ {
+ "logger_name": "fs_mongo",
+ "path": valid_path(),
+ "uri": "mongo:27017",
+ "collection": "files",
+ },
+ "MongoClient crashed",
+ ),
+ (
+ {
+ "logger_name": "fs_mongo",
+ "path": valid_path(),
+ "host": "mongo",
+ "port": 27017,
+ "collection": "files",
+ },
+ "MongoClient crashed",
+ ),
+ ],
+)
+def test_fs_connect_with_invalid_mongoclient(
+ config, exp_exception_message, monkeypatch
+):
def generate_exception(a, b, c=None):
raise Exception(exp_exception_message)
- monkeypatch.setattr(MongoClient, '__init__', generate_exception)
+ monkeypatch.setattr(MongoClient, "__init__", generate_exception)
fs = FsMongo()
with pytest.raises(FsException) as excinfo:
@@ -403,35 +396,41 @@
assert str(excinfo.value) == generic_fs_exception_message(exp_exception_message)
-@pytest.mark.parametrize("config, exp_exception_message", [
- (
- {
- 'logger_name': 'fs_mongo',
- 'path': valid_path(),
- 'uri': 'mongo:27017',
- 'collection': 'files'
- },
- "Collection unavailable"
- ),
- (
- {
- 'logger_name': 'fs_mongo',
- 'path': valid_path(),
- 'host': 'mongo',
- 'port': 27017,
- 'collection': 'files'
- },
- "Collection unavailable"
- )])
-def test_fs_connect_with_invalid_mongo_collection(config, exp_exception_message, monkeypatch):
+@pytest.mark.parametrize(
+ "config, exp_exception_message",
+ [
+ (
+ {
+ "logger_name": "fs_mongo",
+ "path": valid_path(),
+ "uri": "mongo:27017",
+ "collection": "files",
+ },
+ "Collection unavailable",
+ ),
+ (
+ {
+ "logger_name": "fs_mongo",
+ "path": valid_path(),
+ "host": "mongo",
+ "port": 27017,
+ "collection": "files",
+ },
+ "Collection unavailable",
+ ),
+ ],
+)
+def test_fs_connect_with_invalid_mongo_collection(
+ config, exp_exception_message, monkeypatch
+):
def mock_mongoclient_constructor(a, b, c=None):
pass
def generate_exception(a, b):
raise Exception(exp_exception_message)
- monkeypatch.setattr(MongoClient, '__init__', mock_mongoclient_constructor)
- monkeypatch.setattr(MongoClient, '__getitem__', generate_exception)
+ monkeypatch.setattr(MongoClient, "__init__", mock_mongoclient_constructor)
+ monkeypatch.setattr(MongoClient, "__getitem__", generate_exception)
fs = FsMongo()
with pytest.raises(FsException) as excinfo:
@@ -439,27 +438,33 @@
assert str(excinfo.value) == generic_fs_exception_message(exp_exception_message)
-@pytest.mark.parametrize("config, exp_exception_message", [
- (
- {
- 'logger_name': 'fs_mongo',
- 'path': valid_path(),
- 'uri': 'mongo:27017',
- 'collection': 'files'
- },
- "GridFsBucket crashed"
- ),
- (
- {
- 'logger_name': 'fs_mongo',
- 'path': valid_path(),
- 'host': 'mongo',
- 'port': 27017,
- 'collection': 'files'
- },
- "GridFsBucket crashed"
- )])
-def test_fs_connect_with_invalid_gridfsbucket(config, exp_exception_message, monkeypatch):
+@pytest.mark.parametrize(
+ "config, exp_exception_message",
+ [
+ (
+ {
+ "logger_name": "fs_mongo",
+ "path": valid_path(),
+ "uri": "mongo:27017",
+ "collection": "files",
+ },
+ "GridFsBucket crashed",
+ ),
+ (
+ {
+ "logger_name": "fs_mongo",
+ "path": valid_path(),
+ "host": "mongo",
+ "port": 27017,
+ "collection": "files",
+ },
+ "GridFsBucket crashed",
+ ),
+ ],
+)
+def test_fs_connect_with_invalid_gridfsbucket(
+ config, exp_exception_message, monkeypatch
+):
def mock_mongoclient_constructor(a, b, c=None):
pass
@@ -469,9 +474,9 @@
def generate_exception(a, b):
raise Exception(exp_exception_message)
- monkeypatch.setattr(MongoClient, '__init__', mock_mongoclient_constructor)
- monkeypatch.setattr(MongoClient, '__getitem__', mock_mongoclient_getitem)
- monkeypatch.setattr(GridFSBucket, '__init__', generate_exception)
+ monkeypatch.setattr(MongoClient, "__init__", mock_mongoclient_constructor)
+ monkeypatch.setattr(MongoClient, "__getitem__", mock_mongoclient_getitem)
+ monkeypatch.setattr(GridFSBucket, "__init__", generate_exception)
fs = FsMongo()
with pytest.raises(FsException) as excinfo:
@@ -498,53 +503,55 @@
class FakeFS:
- directory_metadata = {'type': 'dir', 'permissions': 509}
- file_metadata = {'type': 'file', 'permissions': 436}
- symlink_metadata = {'type': 'sym', 'permissions': 511}
+ directory_metadata = {"type": "dir", "permissions": 509}
+ file_metadata = {"type": "file", "permissions": 436}
+ symlink_metadata = {"type": "sym", "permissions": 511}
tar_info = {
1: {
- "cursor": FakeCursor(1, 'example_tar', directory_metadata),
+ "cursor": FakeCursor(1, "example_tar", directory_metadata),
"metadata": directory_metadata,
- "stream_content": b'',
+ "stream_content": b"",
"stream_content_bad": b"Something",
- "path": './tmp/example_tar',
+ "path": "./tmp/example_tar",
},
2: {
- "cursor": FakeCursor(2, 'example_tar/directory', directory_metadata),
+ "cursor": FakeCursor(2, "example_tar/directory", directory_metadata),
"metadata": directory_metadata,
- "stream_content": b'',
+ "stream_content": b"",
"stream_content_bad": b"Something",
- "path": './tmp/example_tar/directory',
+ "path": "./tmp/example_tar/directory",
},
3: {
- "cursor": FakeCursor(3, 'example_tar/symlinks', directory_metadata),
+ "cursor": FakeCursor(3, "example_tar/symlinks", directory_metadata),
"metadata": directory_metadata,
- "stream_content": b'',
+ "stream_content": b"",
"stream_content_bad": b"Something",
- "path": './tmp/example_tar/symlinks',
+ "path": "./tmp/example_tar/symlinks",
},
4: {
- "cursor": FakeCursor(4, 'example_tar/directory/file', file_metadata),
+ "cursor": FakeCursor(4, "example_tar/directory/file", file_metadata),
"metadata": file_metadata,
"stream_content": b"Example test",
"stream_content_bad": b"Example test2",
- "path": './tmp/example_tar/directory/file',
+ "path": "./tmp/example_tar/directory/file",
},
5: {
- "cursor": FakeCursor(5, 'example_tar/symlinks/file_link', symlink_metadata),
+ "cursor": FakeCursor(5, "example_tar/symlinks/file_link", symlink_metadata),
"metadata": symlink_metadata,
"stream_content": b"../directory/file",
"stream_content_bad": b"",
- "path": './tmp/example_tar/symlinks/file_link',
+ "path": "./tmp/example_tar/symlinks/file_link",
},
6: {
- "cursor": FakeCursor(6, 'example_tar/symlinks/directory_link', symlink_metadata),
+ "cursor": FakeCursor(
+ 6, "example_tar/symlinks/directory_link", symlink_metadata
+ ),
"metadata": symlink_metadata,
"stream_content": b"../directory/",
"stream_content_bad": b"",
- "path": './tmp/example_tar/symlinks/directory_link',
- }
+ "path": "./tmp/example_tar/symlinks/directory_link",
+ },
}
def upload_from_stream(self, f, stream, metadata=None):
@@ -635,7 +642,9 @@
os.mkdir(path)
os.mkdir("{}example_local".format(path))
os.mkdir("{}example_local/directory".format(path))
- with open("{}example_local/directory/test_file".format(path), "w+") as test_file:
+ with open(
+ "{}example_local/directory/test_file".format(path), "w+"
+ ) as test_file:
test_file.write(file_content)
fs.reverse_sync("example_local")
diff --git a/osm_common/tests/test_msgbase.py b/osm_common/tests/test_msgbase.py
index 9c6c3c5..665968e 100644
--- a/osm_common/tests/test_msgbase.py
+++ b/osm_common/tests/test_msgbase.py
@@ -49,26 +49,36 @@
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 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 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"))
+ 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))
- assert str(excinfo.value).startswith(exception_message("Method 'aioread' not implemented"))
+ assert str(excinfo.value).startswith(
+ exception_message("Method 'aioread' not implemented")
+ )
assert excinfo.value.http_code == http.HTTPStatus.INTERNAL_SERVER_ERROR
diff --git a/osm_common/tests/test_msglocal.py b/osm_common/tests/test_msglocal.py
index 41f6eb8..9548885 100644
--- a/osm_common/tests/test_msglocal.py
+++ b/osm_common/tests/test_msglocal.py
@@ -36,11 +36,11 @@
def valid_path():
- return tempfile.gettempdir() + '/'
+ return tempfile.gettempdir() + "/"
def invalid_path():
- return '/#tweeter/'
+ return "/#tweeter/"
@pytest.fixture(scope="function", params=[True, False])
@@ -90,7 +90,7 @@
def test_constructor():
msg = MsgLocal()
- assert msg.logger == logging.getLogger('msg')
+ assert msg.logger == logging.getLogger("msg")
assert msg.path is None
assert len(msg.files_read) == 0
assert len(msg.files_write) == 0
@@ -98,7 +98,7 @@
def test_constructor_with_logger():
- logger_name = 'msg_local'
+ logger_name = "msg_local"
msg = MsgLocal(logger_name=logger_name)
assert msg.logger == logging.getLogger(logger_name)
assert msg.path is None
@@ -107,15 +107,31 @@
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()),
- ({"logger_name": "msg_local", "path": valid_path() + "test_it/"}, "msg_local", valid_path() + "test_it/"),
- ({"logger_name": "msg_local", "path": valid_path() + "test_it"}, "msg_local", valid_path() + "test_it/"),
- ({"path": valid_path()}, "msg", valid_path()),
- ({"path": valid_path()[:-1]}, "msg", valid_path()),
- ({"path": valid_path() + "test_it/"}, "msg", valid_path() + "test_it/"),
- ({"path": valid_path() + "test_it"}, "msg", valid_path() + "test_it/")])
+@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(),
+ ),
+ (
+ {"logger_name": "msg_local", "path": valid_path() + "test_it/"},
+ "msg_local",
+ valid_path() + "test_it/",
+ ),
+ (
+ {"logger_name": "msg_local", "path": valid_path() + "test_it"},
+ "msg_local",
+ valid_path() + "test_it/",
+ ),
+ ({"path": valid_path()}, "msg", valid_path()),
+ ({"path": valid_path()[:-1]}, "msg", valid_path()),
+ ({"path": valid_path() + "test_it/"}, "msg", valid_path() + "test_it/"),
+ ({"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)
@@ -125,9 +141,13 @@
assert len(msg_local.buffer) == 0
-@pytest.mark.parametrize("config", [
- ({"logger_name": "msg_local", "path": invalid_path()}),
- ({"path": invalid_path()})])
+@pytest.mark.parametrize(
+ "config",
+ [
+ ({"logger_name": "msg_local", "path": invalid_path()}),
+ ({"path": invalid_path()}),
+ ],
+)
def test_connect_with_exception(msg_local, config):
with pytest.raises(MsgException) as excinfo:
msg_local.connect(config)
@@ -146,8 +166,8 @@
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.read("topic1", blocks=False)
+ msg_local_config.read("topic2", blocks=False)
files_read = msg_local_config.files_read.copy()
files_write = msg_local_config.files_write.copy()
msg_local_config.disconnect()
@@ -164,14 +184,14 @@
for f in files_read.values():
assert f.closed
-
+
for f in 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.read("topic1", blocks=False)
+ msg_local_with_data.read("topic2", blocks=False)
files_read = msg_local_with_data.files_read.copy()
files_write = msg_local_with_data.files_write.copy()
@@ -182,77 +202,93 @@
assert f.closed
-@pytest.mark.parametrize("topic, key, msg", [
- ("test_topic", "test_key", "test_msg"),
- ("test", "test_key", "test_msg"),
- ("test_topic", "test", "test_msg"),
- ("test_topic", "test_key", "test"),
- ("test_topic", "test_list", ["a", "b", "c"]),
- ("test_topic", "test_tuple", ("c", "b", "a")),
- ("test_topic", "test_dict", {"a": 1, "b": 2, "c": 3}),
- ("test_topic", "test_number", 123),
- ("test_topic", "test_float", 1.23),
- ("test_topic", "test_boolean", True),
- ("test_topic", "test_none", None)])
+@pytest.mark.parametrize(
+ "topic, key, msg",
+ [
+ ("test_topic", "test_key", "test_msg"),
+ ("test", "test_key", "test_msg"),
+ ("test_topic", "test", "test_msg"),
+ ("test_topic", "test_key", "test"),
+ ("test_topic", "test_list", ["a", "b", "c"]),
+ ("test_topic", "test_tuple", ("c", "b", "a")),
+ ("test_topic", "test_dict", {"a": 1, "b": 2, "c": 3}),
+ ("test_topic", "test_number", 123),
+ ("test_topic", "test_float", 1.23),
+ ("test_topic", "test_boolean", True),
+ ("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.safe_load(stream) == {key: msg if not isinstance(msg, tuple) else list(msg)}
+ with open(file_path, "r") as stream:
+ assert yaml.safe_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", "test_msg", 4),
- ("test_topic", "test_key", "test", 2),
- ("test_topic", "test_list", ["a", "b", "c"], 3),
- ("test_topic", "test_tuple", ("c", "b", "a"), 4),
- ("test_topic", "test_dict", {"a": 1, "b": 2, "c": 3}, 2),
- ("test_topic", "test_number", 123, 3),
- ("test_topic", "test_float", 1.23, 4),
- ("test_topic", "test_boolean", True, 2),
- ("test_topic", "test_none", None, 3)])
+@pytest.mark.parametrize(
+ "topic, key, msg, times",
+ [
+ ("test_topic", "test_key", "test_msg", 2),
+ ("test", "test_key", "test_msg", 3),
+ ("test_topic", "test", "test_msg", 4),
+ ("test_topic", "test_key", "test", 2),
+ ("test_topic", "test_list", ["a", "b", "c"], 3),
+ ("test_topic", "test_tuple", ("c", "b", "a"), 4),
+ ("test_topic", "test_dict", {"a": 1, "b": 2, "c": 3}, 2),
+ ("test_topic", "test_number", 123, 3),
+ ("test_topic", "test_float", 1.23, 4),
+ ("test_topic", "test_boolean", True, 2),
+ ("test_topic", "test_none", None, 3),
+ ],
+)
def test_write_with_multiple_calls(msg_local_config, topic, key, msg, times):
file_path = msg_local_config.path + topic
-
+
for _ in range(times):
msg_local_config.write(topic, key, msg)
assert os.path.exists(file_path)
- with open(file_path, 'r') as stream:
+ with open(file_path, "r") as stream:
for _ in range(times):
data = stream.readline()
- assert yaml.safe_load(data) == {key: msg if not isinstance(msg, tuple) else list(msg)}
+ assert yaml.safe_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()
-
+
with pytest.raises(MsgException) as excinfo:
msg_local_config.write("test", "test", "test")
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"}]),
- (["topic2"], [{"key": "value"}]),
- (["topic", "topic1"], [{"key": "value"}]),
- (["topic", "topic2"], [{"key": "value"}]),
- (["topic1", "topic2"], [{"key": "value"}]),
- (["topic", "topic1", "topic2"], [{"key": "value"}]),
- (["topic"], [{"key": "value"}, {"key1": "value1"}]),
- (["topic1"], [{"key": "value"}, {"key1": "value1"}]),
- (["topic2"], [{"key": "value"}, {"key1": "value1"}]),
- (["topic", "topic1"], [{"key": "value"}, {"key1": "value1"}]),
- (["topic", "topic2"], [{"key": "value"}, {"key1": "value1"}]),
- (["topic1", "topic2"], [{"key": "value"}, {"key1": "value1"}]),
- (["topic", "topic1", "topic2"], [{"key": "value"}, {"key1": "value1"}])])
+@pytest.mark.parametrize(
+ "topics, datas",
+ [
+ (["topic"], [{"key": "value"}]),
+ (["topic1"], [{"key": "value"}]),
+ (["topic2"], [{"key": "value"}]),
+ (["topic", "topic1"], [{"key": "value"}]),
+ (["topic", "topic2"], [{"key": "value"}]),
+ (["topic1", "topic2"], [{"key": "value"}]),
+ (["topic", "topic1", "topic2"], [{"key": "value"}]),
+ (["topic"], [{"key": "value"}, {"key1": "value1"}]),
+ (["topic1"], [{"key": "value"}, {"key1": "value1"}]),
+ (["topic2"], [{"key": "value"}, {"key1": "value1"}]),
+ (["topic", "topic1"], [{"key": "value"}, {"key1": "value1"}]),
+ (["topic", "topic2"], [{"key": "value"}, {"key1": "value1"}]),
+ (["topic1", "topic2"], [{"key": "value"}, {"key1": "value1"}]),
+ (["topic", "topic1", "topic2"], [{"key": "value"}, {"key1": "value1"}]),
+ ],
+)
def test_read(msg_local_with_data, topics, datas):
def write_to_topic(topics, datas):
# Allow msglocal to block while waiting
@@ -282,21 +318,25 @@
t.join()
-@pytest.mark.parametrize("topics, datas", [
- (["topic"], [{"key": "value"}]),
- (["topic1"], [{"key": "value"}]),
- (["topic2"], [{"key": "value"}]),
- (["topic", "topic1"], [{"key": "value"}]),
- (["topic", "topic2"], [{"key": "value"}]),
- (["topic1", "topic2"], [{"key": "value"}]),
- (["topic", "topic1", "topic2"], [{"key": "value"}]),
- (["topic"], [{"key": "value"}, {"key1": "value1"}]),
- (["topic1"], [{"key": "value"}, {"key1": "value1"}]),
- (["topic2"], [{"key": "value"}, {"key1": "value1"}]),
- (["topic", "topic1"], [{"key": "value"}, {"key1": "value1"}]),
- (["topic", "topic2"], [{"key": "value"}, {"key1": "value1"}]),
- (["topic1", "topic2"], [{"key": "value"}, {"key1": "value1"}]),
- (["topic", "topic1", "topic2"], [{"key": "value"}, {"key1": "value1"}])])
+@pytest.mark.parametrize(
+ "topics, datas",
+ [
+ (["topic"], [{"key": "value"}]),
+ (["topic1"], [{"key": "value"}]),
+ (["topic2"], [{"key": "value"}]),
+ (["topic", "topic1"], [{"key": "value"}]),
+ (["topic", "topic2"], [{"key": "value"}]),
+ (["topic1", "topic2"], [{"key": "value"}]),
+ (["topic", "topic1", "topic2"], [{"key": "value"}]),
+ (["topic"], [{"key": "value"}, {"key1": "value1"}]),
+ (["topic1"], [{"key": "value"}, {"key1": "value1"}]),
+ (["topic2"], [{"key": "value"}, {"key1": "value1"}]),
+ (["topic", "topic1"], [{"key": "value"}, {"key1": "value1"}]),
+ (["topic", "topic2"], [{"key": "value"}, {"key1": "value1"}]),
+ (["topic1", "topic2"], [{"key": "value"}, {"key1": "value1"}]),
+ (["topic", "topic1", "topic2"], [{"key": "value"}, {"key1": "value1"}]),
+ ],
+)
def test_read_non_block(msg_local_with_data, topics, datas):
def write_to_topic(topics, datas):
for topic in topics:
@@ -316,7 +356,9 @@
for topic in topics:
for data in datas:
- recv_topic, recv_key, recv_msg = msg_local_with_data.read(topic, blocks=False)
+ 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
@@ -324,21 +366,25 @@
assert recv_msg == val
-@pytest.mark.parametrize("topics, datas", [
- (["topic"], [{"key": "value"}]),
- (["topic1"], [{"key": "value"}]),
- (["topic2"], [{"key": "value"}]),
- (["topic", "topic1"], [{"key": "value"}]),
- (["topic", "topic2"], [{"key": "value"}]),
- (["topic1", "topic2"], [{"key": "value"}]),
- (["topic", "topic1", "topic2"], [{"key": "value"}]),
- (["topic"], [{"key": "value"}, {"key1": "value1"}]),
- (["topic1"], [{"key": "value"}, {"key1": "value1"}]),
- (["topic2"], [{"key": "value"}, {"key1": "value1"}]),
- (["topic", "topic1"], [{"key": "value"}, {"key1": "value1"}]),
- (["topic", "topic2"], [{"key": "value"}, {"key1": "value1"}]),
- (["topic1", "topic2"], [{"key": "value"}, {"key1": "value1"}]),
- (["topic", "topic1", "topic2"], [{"key": "value"}, {"key1": "value1"}])])
+@pytest.mark.parametrize(
+ "topics, datas",
+ [
+ (["topic"], [{"key": "value"}]),
+ (["topic1"], [{"key": "value"}]),
+ (["topic2"], [{"key": "value"}]),
+ (["topic", "topic1"], [{"key": "value"}]),
+ (["topic", "topic2"], [{"key": "value"}]),
+ (["topic1", "topic2"], [{"key": "value"}]),
+ (["topic", "topic1", "topic2"], [{"key": "value"}]),
+ (["topic"], [{"key": "value"}, {"key1": "value1"}]),
+ (["topic1"], [{"key": "value"}, {"key1": "value1"}]),
+ (["topic2"], [{"key": "value"}, {"key1": "value1"}]),
+ (["topic", "topic1"], [{"key": "value"}, {"key1": "value1"}]),
+ (["topic", "topic2"], [{"key": "value"}, {"key1": "value1"}]),
+ (["topic1", "topic2"], [{"key": "value"}, {"key1": "value1"}]),
+ (["topic", "topic1", "topic2"], [{"key": "value"}, {"key1": "value1"}]),
+ ],
+)
def test_read_non_block_none(msg_local_with_data, topics, datas):
def write_to_topic(topics, datas):
time.sleep(2)
@@ -347,6 +393,7 @@
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:
@@ -360,9 +407,7 @@
t.join()
-@pytest.mark.parametrize("blocks", [
- (True),
- (False)])
+@pytest.mark.parametrize("blocks", [(True), (False)])
def test_read_exception(msg_local_with_data, blocks):
msg_local_with_data.files_read = MagicMock()
msg_local_with_data.files_read.__contains__.side_effect = Exception()
@@ -373,21 +418,25 @@
assert excinfo.value.http_code == http.HTTPStatus.INTERNAL_SERVER_ERROR
-@pytest.mark.parametrize("topics, datas", [
- (["topic"], [{"key": "value"}]),
- (["topic1"], [{"key": "value"}]),
- (["topic2"], [{"key": "value"}]),
- (["topic", "topic1"], [{"key": "value"}]),
- (["topic", "topic2"], [{"key": "value"}]),
- (["topic1", "topic2"], [{"key": "value"}]),
- (["topic", "topic1", "topic2"], [{"key": "value"}]),
- (["topic"], [{"key": "value"}, {"key1": "value1"}]),
- (["topic1"], [{"key": "value"}, {"key1": "value1"}]),
- (["topic2"], [{"key": "value"}, {"key1": "value1"}]),
- (["topic", "topic1"], [{"key": "value"}, {"key1": "value1"}]),
- (["topic", "topic2"], [{"key": "value"}, {"key1": "value1"}]),
- (["topic1", "topic2"], [{"key": "value"}, {"key1": "value1"}]),
- (["topic", "topic1", "topic2"], [{"key": "value"}, {"key1": "value1"}])])
+@pytest.mark.parametrize(
+ "topics, datas",
+ [
+ (["topic"], [{"key": "value"}]),
+ (["topic1"], [{"key": "value"}]),
+ (["topic2"], [{"key": "value"}]),
+ (["topic", "topic1"], [{"key": "value"}]),
+ (["topic", "topic2"], [{"key": "value"}]),
+ (["topic1", "topic2"], [{"key": "value"}]),
+ (["topic", "topic1", "topic2"], [{"key": "value"}]),
+ (["topic"], [{"key": "value"}, {"key1": "value1"}]),
+ (["topic1"], [{"key": "value"}, {"key1": "value1"}]),
+ (["topic2"], [{"key": "value"}, {"key1": "value1"}]),
+ (["topic", "topic1"], [{"key": "value"}, {"key1": "value1"}]),
+ (["topic", "topic2"], [{"key": "value"}, {"key1": "value1"}]),
+ (["topic1", "topic2"], [{"key": "value"}, {"key1": "value1"}]),
+ (["topic", "topic1", "topic2"], [{"key": "value"}, {"key1": "value1"}]),
+ ],
+)
def test_aioread(msg_local_with_data, event_loop, topics, datas):
def write_to_topic(topics, datas):
time.sleep(2)
@@ -396,6 +445,7 @@
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:
@@ -405,7 +455,9 @@
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 = 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]
@@ -435,55 +487,69 @@
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", "test_msg"),
- ("test_topic", "test_key", "test"),
- ("test_topic", "test_list", ["a", "b", "c"]),
- ("test_topic", "test_tuple", ("c", "b", "a")),
- ("test_topic", "test_dict", {"a": 1, "b": 2, "c": 3}),
- ("test_topic", "test_number", 123),
- ("test_topic", "test_float", 1.23),
- ("test_topic", "test_boolean", True),
- ("test_topic", "test_none", None)])
+@pytest.mark.parametrize(
+ "topic, key, msg",
+ [
+ ("test_topic", "test_key", "test_msg"),
+ ("test", "test_key", "test_msg"),
+ ("test_topic", "test", "test_msg"),
+ ("test_topic", "test_key", "test"),
+ ("test_topic", "test_list", ["a", "b", "c"]),
+ ("test_topic", "test_tuple", ("c", "b", "a")),
+ ("test_topic", "test_dict", {"a": 1, "b": 2, "c": 3}),
+ ("test_topic", "test_number", 123),
+ ("test_topic", "test_float", 1.23),
+ ("test_topic", "test_boolean", True),
+ ("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.safe_load(stream) == {key: msg if not isinstance(msg, tuple) else list(msg)}
+ with open(file_path, "r") as stream:
+ assert yaml.safe_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", "test_msg", 4),
- ("test_topic", "test_key", "test", 2),
- ("test_topic", "test_list", ["a", "b", "c"], 3),
- ("test_topic", "test_tuple", ("c", "b", "a"), 4),
- ("test_topic", "test_dict", {"a": 1, "b": 2, "c": 3}, 2),
- ("test_topic", "test_number", 123, 3),
- ("test_topic", "test_float", 1.23, 4),
- ("test_topic", "test_boolean", True, 2),
- ("test_topic", "test_none", None, 3)])
-def test_aiowrite_with_multiple_calls(msg_local_config, event_loop, topic, key, msg, times):
+@pytest.mark.parametrize(
+ "topic, key, msg, times",
+ [
+ ("test_topic", "test_key", "test_msg", 2),
+ ("test", "test_key", "test_msg", 3),
+ ("test_topic", "test", "test_msg", 4),
+ ("test_topic", "test_key", "test", 2),
+ ("test_topic", "test_list", ["a", "b", "c"], 3),
+ ("test_topic", "test_tuple", ("c", "b", "a"), 4),
+ ("test_topic", "test_dict", {"a": 1, "b": 2, "c": 3}, 2),
+ ("test_topic", "test_number", 123, 3),
+ ("test_topic", "test_float", 1.23, 4),
+ ("test_topic", "test_boolean", True, 2),
+ ("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:
+ with open(file_path, "r") as stream:
for _ in range(times):
data = stream.readline()
- assert yaml.safe_load(data) == {key: msg if not isinstance(msg, tuple) else list(msg)}
+ assert yaml.safe_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()
-
+
with pytest.raises(MsgException) as excinfo:
event_loop.run_until_complete(msg_local_config.aiowrite("test", "test", "test"))
assert str(excinfo.value).startswith(empty_exception_message())
diff --git a/osm_common/tests/test_sol004_package.py b/osm_common/tests/test_sol004_package.py
index b9f13af..dc71d91 100644
--- a/osm_common/tests/test_sol004_package.py
+++ b/osm_common/tests/test_sol004_package.py
@@ -28,82 +28,124 @@
class SOL004ValidatorTest(unittest.TestCase):
def test_get_package_file_hash_algorithm_from_manifest_with_metadata_dir(self):
- package = SOL004Package('osm_common/tests/packages/native_charm_with_metadata_dir_vnf')
- algorithm = package.get_package_file_hash_algorithm_from_manifest('Scripts/charms/simple/src/charm.py')
- self.assertEqual(algorithm, 'SHA-256')
+ package = SOL004Package(
+ "osm_common/tests/packages/native_charm_with_metadata_dir_vnf"
+ )
+ algorithm = package.get_package_file_hash_algorithm_from_manifest(
+ "Scripts/charms/simple/src/charm.py"
+ )
+ self.assertEqual(algorithm, "SHA-256")
def test_get_package_file_hash_algorithm_from_manifest_without_metadata_dir(self):
- package = SOL004Package('osm_common/tests/packages/native_charm_without_metadata_dir_vnf')
- algorithm = package.get_package_file_hash_algorithm_from_manifest('Scripts/charms/simple/src/charm.py')
- self.assertEqual(algorithm, 'SHA-256')
+ package = SOL004Package(
+ "osm_common/tests/packages/native_charm_without_metadata_dir_vnf"
+ )
+ algorithm = package.get_package_file_hash_algorithm_from_manifest(
+ "Scripts/charms/simple/src/charm.py"
+ )
+ self.assertEqual(algorithm, "SHA-256")
def test_get_package_file_hash_algorithm_from_manifest_on_non_existent_file(self):
- package = SOL004Package('osm_common/tests/packages/native_charm_with_metadata_dir_vnf')
+ package = SOL004Package(
+ "osm_common/tests/packages/native_charm_with_metadata_dir_vnf"
+ )
with self.assertRaises(SOL004PackageException):
- package.get_package_file_hash_algorithm_from_manifest('Non/Existing/file')
+ package.get_package_file_hash_algorithm_from_manifest("Non/Existing/file")
def test_get_package_file_hash_digest_from_manifest_with_metadata_dir(self):
- package = SOL004Package('osm_common/tests/packages/native_charm_with_metadata_dir_vnf')
- digest = package.get_package_file_hash_digest_from_manifest('Scripts/charms/simple/src/charm.py')
- self.assertEqual(digest, '7895f7b9e1b7ed5b5bcd64398950ca95b456d7fc973334351474eed466c2f480')
+ package = SOL004Package(
+ "osm_common/tests/packages/native_charm_with_metadata_dir_vnf"
+ )
+ digest = package.get_package_file_hash_digest_from_manifest(
+ "Scripts/charms/simple/src/charm.py"
+ )
+ self.assertEqual(
+ digest, "ea72f897a966e6174ed9164fabc3c500df5a2f712eb6b22ab2408afb07d04d14"
+ )
def test_get_package_file_hash_digest_from_manifest_without_metadata_dir(self):
- package = SOL004Package('osm_common/tests/packages/native_charm_without_metadata_dir_vnf')
- digest = package.get_package_file_hash_digest_from_manifest('Scripts/charms/simple/src/charm.py')
- self.assertEqual(digest, '7895f7b9e1b7ed5b5bcd64398950ca95b456d7fc973334351474eed466c2f480')
+ package = SOL004Package(
+ "osm_common/tests/packages/native_charm_without_metadata_dir_vnf"
+ )
+ digest = package.get_package_file_hash_digest_from_manifest(
+ "Scripts/charms/simple/src/charm.py"
+ )
+ self.assertEqual(
+ digest, "ea72f897a966e6174ed9164fabc3c500df5a2f712eb6b22ab2408afb07d04d14"
+ )
def test_get_package_file_hash_digest_from_manifest_on_non_existent_file(self):
- package = SOL004Package('osm_common/tests/packages/native_charm_with_metadata_dir_vnf')
+ package = SOL004Package(
+ "osm_common/tests/packages/native_charm_with_metadata_dir_vnf"
+ )
with self.assertRaises(SOL004PackageException):
- package.get_package_file_hash_digest_from_manifest('Non/Existing/file')
+ package.get_package_file_hash_digest_from_manifest("Non/Existing/file")
- def test_get_package_file_hash_digest_from_manifest_on_non_existing_hash_entry(self):
- package = SOL004Package('osm_common/tests/packages/invalid_package_vnf')
+ def test_get_package_file_hash_digest_from_manifest_on_non_existing_hash_entry(
+ self,
+ ):
+ package = SOL004Package("osm_common/tests/packages/invalid_package_vnf")
with self.assertRaises(SOL004PackageException):
- package.get_package_file_hash_digest_from_manifest('Scripts/charms/simple/hooks/upgrade-charm')
+ package.get_package_file_hash_digest_from_manifest(
+ "Scripts/charms/simple/hooks/upgrade-charm"
+ )
def test_validate_package_file_hash_with_metadata_dir(self):
- package = SOL004Package('osm_common/tests/packages/native_charm_with_metadata_dir_vnf')
- package.validate_package_file_hash('Scripts/charms/simple/src/charm.py')
+ package = SOL004Package(
+ "osm_common/tests/packages/native_charm_with_metadata_dir_vnf"
+ )
+ package.validate_package_file_hash("Scripts/charms/simple/src/charm.py")
def test_validate_package_file_hash_without_metadata_dir(self):
- package = SOL004Package('osm_common/tests/packages/native_charm_without_metadata_dir_vnf')
- package.validate_package_file_hash('Scripts/charms/simple/src/charm.py')
+ package = SOL004Package(
+ "osm_common/tests/packages/native_charm_without_metadata_dir_vnf"
+ )
+ package.validate_package_file_hash("Scripts/charms/simple/src/charm.py")
def test_validate_package_file_hash_on_non_existing_file(self):
- package = SOL004Package('osm_common/tests/packages/native_charm_with_metadata_dir_vnf')
+ package = SOL004Package(
+ "osm_common/tests/packages/native_charm_with_metadata_dir_vnf"
+ )
with self.assertRaises(SOL004PackageException):
- package.validate_package_file_hash('Non/Existing/file')
+ package.validate_package_file_hash("Non/Existing/file")
def test_validate_package_file_hash_on_wrong_manifest_hash(self):
- package = SOL004Package('osm_common/tests/packages/invalid_package_vnf')
+ package = SOL004Package("osm_common/tests/packages/invalid_package_vnf")
with self.assertRaises(SOL004PackageException):
- package.validate_package_file_hash('Scripts/charms/simple/hooks/start')
+ package.validate_package_file_hash("Scripts/charms/simple/hooks/start")
def test_validate_package_file_hash_on_unsupported_hash_algorithm(self):
- package = SOL004Package('osm_common/tests/packages/invalid_package_vnf')
+ package = SOL004Package("osm_common/tests/packages/invalid_package_vnf")
with self.assertRaises(SOL004PackageException):
- package.validate_package_file_hash('Scripts/charms/simple/src/charm.py')
+ package.validate_package_file_hash("Scripts/charms/simple/src/charm.py")
def test_validate_package_hashes_with_metadata_dir(self):
- package = SOL004Package('osm_common/tests/packages/native_charm_with_metadata_dir_vnf')
+ package = SOL004Package(
+ "osm_common/tests/packages/native_charm_with_metadata_dir_vnf"
+ )
package.validate_package_hashes()
def test_validate_package_hashes_without_metadata_dir(self):
- package = SOL004Package('osm_common/tests/packages/native_charm_without_metadata_dir_vnf')
+ package = SOL004Package(
+ "osm_common/tests/packages/native_charm_without_metadata_dir_vnf"
+ )
package.validate_package_hashes()
def test_validate_package_hashes_on_invalid_package(self):
- package = SOL004Package('osm_common/tests/packages/invalid_package_vnf')
+ package = SOL004Package("osm_common/tests/packages/invalid_package_vnf")
with self.assertRaises(SOL004PackageException):
package.validate_package_hashes()
def test_get_descriptor_location_with_metadata_dir(self):
- package = SOL004Package('osm_common/tests/packages/native_charm_with_metadata_dir_vnf')
+ package = SOL004Package(
+ "osm_common/tests/packages/native_charm_with_metadata_dir_vnf"
+ )
descriptor_path = package.get_descriptor_location()
- self.assertEqual(descriptor_path, 'Definitions/native_charm_vnfd.yaml')
+ self.assertEqual(descriptor_path, "Definitions/native_charm_vnfd.yaml")
def test_get_descriptor_location_without_metadata_dir(self):
- package = SOL004Package('osm_common/tests/packages/native_charm_without_metadata_dir_vnf')
+ package = SOL004Package(
+ "osm_common/tests/packages/native_charm_without_metadata_dir_vnf"
+ )
descriptor_path = package.get_descriptor_location()
- self.assertEqual(descriptor_path, 'native_charm_vnfd.yaml')
+ self.assertEqual(descriptor_path, "native_charm_vnfd.yaml")