X-Git-Url: https://osm.etsi.org/gitweb/?a=blobdiff_plain;f=osm_common%2Ftests%2Ftest_dbbase.py;fp=osm_common%2Ftests%2Ftest_dbbase.py;h=e582c7e8f08699c93ba5a1ffe36e2657171329ff;hb=a06b854f2b278aaee015fc1f76015895f8cf50c1;hp=050abdbb2377816cf4ed20c01711ae7666206a36;hpb=b2d732a70efa33e4bc478d351d64bc4adb4ea332;p=osm%2Fcommon.git diff --git a/osm_common/tests/test_dbbase.py b/osm_common/tests/test_dbbase.py index 050abdb..e582c7e 100644 --- a/osm_common/tests/test_dbbase.py +++ b/osm_common/tests/test_dbbase.py @@ -997,7 +997,6 @@ class TestAsyncEncryption(unittest.TestCase): def setUp(self, mock_logger): mock_logger = logging.getLogger() mock_logger.disabled = True - self.loop = asyncio.get_event_loop() self.encryption = Encryption(uri="uri", config={}) self.encryption.encoding_type = encoding_type self.encryption.encrypt_mode = encyrpt_mode @@ -1022,7 +1021,8 @@ class TestAsyncEncryption(unittest.TestCase): "ip": "192.168.12.23", } fields = ["secret", "cacert"] - self.loop.run_until_complete( + + asyncio.run( self.encryption.decrypt_fields(input_item, fields, schema_version, salt) ) self.assertEqual(input_item, expected_item) @@ -1035,7 +1035,7 @@ class TestAsyncEncryption(unittest.TestCase): """item is empty and fields exists.""" input_item = {} fields = ["secret", "cacert"] - self.loop.run_until_complete( + asyncio.run( self.encryption.decrypt_fields(input_item, fields, schema_version, salt) ) self.assertEqual(input_item, {}) @@ -1046,7 +1046,7 @@ class TestAsyncEncryption(unittest.TestCase): """item exists and fields is empty.""" input_item = copy.deepcopy(item) fields = [] - self.loop.run_until_complete( + asyncio.run( self.encryption.decrypt_fields(input_item, fields, schema_version, salt) ) self.assertEqual(input_item, item) @@ -1064,7 +1064,7 @@ class TestAsyncEncryption(unittest.TestCase): "path": "/var", "ip": "192.168.12.23", } - self.loop.run_until_complete( + asyncio.run( self.encryption.decrypt_fields(input_item, fields, schema_version, salt) ) self.assertEqual(input_item, expected_item) @@ -1081,7 +1081,7 @@ class TestAsyncEncryption(unittest.TestCase): mock_decrypt.return_value = "mysecret" input_item = copy.deepcopy(item) fields = ["secret"] - self.loop.run_until_complete( + asyncio.run( self.encryption.decrypt_fields(input_item, fields, schema_version, salt) ) self.assertEqual(input_item, item) @@ -1097,7 +1097,7 @@ class TestAsyncEncryption(unittest.TestCase): fields = ["secret"] input_item = copy.deepcopy(item) with self.assertRaises(Exception) as error: - self.loop.run_until_complete( + asyncio.run( self.encryption.decrypt_fields(input_item, fields, schema_version, salt) ) self.assertEqual( @@ -1113,9 +1113,7 @@ class TestAsyncEncryption(unittest.TestCase): def test_encrypt(self, mock_encrypt_value, mock_get_secret_key): """Method decrypt raises error.""" mock_encrypt_value.return_value = encyrpted_value - result = self.loop.run_until_complete( - self.encryption.encrypt(value, schema_version, salt) - ) + result = asyncio.run(self.encryption.encrypt(value, schema_version, salt)) self.assertEqual(result, encyrpted_value) mock_get_secret_key.assert_called_once() mock_encrypt_value.assert_called_once_with(value, schema_version, salt) @@ -1128,9 +1126,7 @@ class TestAsyncEncryption(unittest.TestCase): """Method get_secret_key raises error.""" mock_get_secret_key.side_effect = DbException("Unexpected type.") with self.assertRaises(Exception) as error: - self.loop.run_until_complete( - self.encryption.encrypt(value, schema_version, salt) - ) + asyncio.run(self.encryption.encrypt(value, schema_version, salt)) self.assertEqual(str(error.exception), "database exception Unexpected type.") mock_get_secret_key.assert_called_once() mock_encrypt_value.assert_not_called() @@ -1143,9 +1139,7 @@ class TestAsyncEncryption(unittest.TestCase): "A bytes-like object is required, not 'str'" ) with self.assertRaises(Exception) as error: - self.loop.run_until_complete( - self.encryption.encrypt(value, schema_version, salt) - ) + asyncio.run(self.encryption.encrypt(value, schema_version, salt)) self.assertEqual( str(error.exception), "A bytes-like object is required, not 'str'" ) @@ -1157,7 +1151,7 @@ class TestAsyncEncryption(unittest.TestCase): def test_decrypt(self, mock_decrypt_value, mock_get_secret_key): """Decrypted successfully.""" mock_decrypt_value.return_value = value - result = self.loop.run_until_complete( + result = asyncio.run( self.encryption.decrypt(encyrpted_value, schema_version, salt) ) self.assertEqual(result, value) @@ -1174,9 +1168,7 @@ class TestAsyncEncryption(unittest.TestCase): """Method get_secret_key raises error.""" mock_get_secret_key.side_effect = DbException("Unexpected type.") with self.assertRaises(Exception) as error: - self.loop.run_until_complete( - self.encryption.decrypt(encyrpted_value, schema_version, salt) - ) + asyncio.run(self.encryption.decrypt(encyrpted_value, schema_version, salt)) self.assertEqual(str(error.exception), "database exception Unexpected type.") mock_get_secret_key.assert_called_once() mock_decrypt_value.assert_not_called() @@ -1191,9 +1183,7 @@ class TestAsyncEncryption(unittest.TestCase): "A bytes-like object is required, not 'str'" ) with self.assertRaises(Exception) as error: - self.loop.run_until_complete( - self.encryption.decrypt(encyrpted_value, schema_version, salt) - ) + asyncio.run(self.encryption.decrypt(encyrpted_value, schema_version, salt)) self.assertEqual( str(error.exception), "A bytes-like object is required, not 'str'" ) @@ -1281,7 +1271,7 @@ class TestAsyncEncryption(unittest.TestCase): def test_get_secret_key_exists(self, mock_join_keys): """secret_key exists.""" self.encryption._secret_key = secret_key - self.loop.run_until_complete(self.encryption.get_secret_key()) + asyncio.run(self.encryption.get_secret_key()) self.assertEqual(self.encryption.secret_key, secret_key) mock_join_keys.assert_not_called() @@ -1295,7 +1285,7 @@ class TestAsyncEncryption(unittest.TestCase): self.encryption._admin_collection.find_one.return_value = None self.encryption._config = {"database_commonkey": "osm_new_key"} mock_join_keys.return_value = joined_key - self.loop.run_until_complete(self.encryption.get_secret_key()) + asyncio.run(self.encryption.get_secret_key()) self.assertEqual(self.encryption.secret_key, joined_key) self.assertEqual(mock_join_keys.call_count, 1) mock_b64decode.assert_not_called() @@ -1310,7 +1300,7 @@ class TestAsyncEncryption(unittest.TestCase): self.encryption._admin_collection.find_one.return_value = {"version": "1.0"} self.encryption._config = {"database_commonkey": "osm_new_key"} mock_join_keys.return_value = joined_key - self.loop.run_until_complete(self.encryption.get_secret_key()) + asyncio.run(self.encryption.get_secret_key()) self.assertEqual(self.encryption.secret_key, joined_key) self.assertEqual(mock_join_keys.call_count, 1) mock_b64decode.assert_not_called() @@ -1335,7 +1325,7 @@ class TestAsyncEncryption(unittest.TestCase): self.encryption._config = {"database_commonkey": "osm_new_key"} mock_join_keys.side_effect = [secret_key, joined_key] mock_b64decode.return_value = base64_decoded_serial - self.loop.run_until_complete(self.encryption.get_secret_key()) + asyncio.run(self.encryption.get_secret_key()) self.assertEqual(self.encryption.secret_key, joined_key) self.assertEqual(mock_join_keys.call_count, 2) mock_b64decode.assert_called_once_with(serial_bytes) @@ -1360,7 +1350,7 @@ class TestAsyncEncryption(unittest.TestCase): self.encryption._config = {"database_commonkey": "osm_new_key"} mock_join_keys.side_effect = DbException("Invalid data type.") with self.assertRaises(Exception) as error: - self.loop.run_until_complete(self.encryption.get_secret_key()) + asyncio.run(self.encryption.get_secret_key()) self.assertEqual(str(error.exception), "database exception Invalid data type.") self.assertEqual(mock_join_keys.call_count, 1) check_if_assert_not_called( @@ -1384,7 +1374,7 @@ class TestAsyncEncryption(unittest.TestCase): "A bytes-like object is required, not 'str'" ) with self.assertRaises(Exception) as error: - self.loop.run_until_complete(self.encryption.get_secret_key()) + asyncio.run(self.encryption.get_secret_key()) self.assertEqual( str(error.exception), "A bytes-like object is required, not 'str'" ) @@ -1410,7 +1400,7 @@ class TestAsyncEncryption(unittest.TestCase): self.encryption._config = {"database_commonkey": "osm_new_key"} mock_join_keys.return_value = secret_key with self.assertRaises(Exception) as error: - self.loop.run_until_complete(self.encryption.get_secret_key()) + asyncio.run(self.encryption.get_secret_key()) self.assertEqual(str(error.exception), "database exception Connection failed.") self.assertEqual(self.encryption.secret_key, None) self.assertEqual(mock_join_keys.call_count, 1) @@ -1423,10 +1413,10 @@ class TestAsyncEncryption(unittest.TestCase): def test_encrypt_decrypt_with_schema_version_1_1_with_salt(self): """Encrypt and decrypt with schema version 1.1, salt exists.""" - encrypted_msg = self.loop.run_until_complete( + encrypted_msg = asyncio.run( self.encryption.encrypt(value, schema_version, salt) ) - decrypted_msg = self.loop.run_until_complete( + decrypted_msg = asyncio.run( self.encryption.decrypt(encrypted_msg, schema_version, salt) ) self.assertEqual(value, decrypted_msg) @@ -1434,10 +1424,10 @@ class TestAsyncEncryption(unittest.TestCase): def test_encrypt_decrypt_with_schema_version_1_0_with_salt(self): """Encrypt and decrypt with schema version 1.0, salt exists.""" schema_version = "1.0" - encrypted_msg = self.loop.run_until_complete( + encrypted_msg = asyncio.run( self.encryption.encrypt(value, schema_version, salt) ) - decrypted_msg = self.loop.run_until_complete( + decrypted_msg = asyncio.run( self.encryption.decrypt(encrypted_msg, schema_version, salt) ) self.assertEqual(value, decrypted_msg) @@ -1446,9 +1436,7 @@ class TestAsyncEncryption(unittest.TestCase): """Encrypt and decrypt with schema version 1.1, without salt.""" salt = None with self.assertRaises(Exception) as error: - self.loop.run_until_complete( - self.encryption.encrypt(value, schema_version, salt) - ) + asyncio.run(self.encryption.encrypt(value, schema_version, salt)) self.assertEqual(str(error.exception), "'NoneType' object is not iterable")