ISSUE #367

* Add tests for `AiopgDatabaseClient`.

Author: Sergio García Prado

packages/core/minos-microservice-common/minos/common/database/clients/aiopg.py

@@ -41,7 +41,7 @@
 
 
 class AiopgDatabaseClient(DatabaseClient):
-    """TODO"""
+    """Aiopg Database Client class."""
 
     _connection: Optional[Connection]
     _cursor: Optional[Cursor]
@@ -106,8 +106,11 @@ async def _close_connection(self):
         self._connection = None
         logger.debug(f"Destroyed {self.database!r} database connection identified by {id(self._connection)}!")
 
-    async def is_valid(self):
-        """TODO"""
+    async def is_valid(self) -> bool:
+        """Check if the instance is valid.
+
+        :return: ``True`` if it is valid or ``False`` otherwise.
+        """
         if self._connection is None:
             return False
 
@@ -120,43 +123,32 @@ async def is_valid(self):
         return not self._connection.closed
 
     async def reset(self, **kwargs) -> None:
-        """TODO"""
-        await self._destroy_cursor(**kwargs)
+        """Reset the current instance status.
 
-    @property
-    def notifications(self) -> ClosableQueue:
-        return self._connection.notifies
+        :param kwargs: Additional named parameters.
+        :return: This method does not return anything.
+        """
+        await self._destroy_cursor(**kwargs)
 
     # noinspection PyUnusedLocal
     async def fetch_all(
         self,
         *args,
         timeout: Optional[float] = None,
         lock: Optional[int] = None,
-        streaming_mode: bool = False,
         **kwargs,
     ) -> AsyncIterator[tuple]:
         """Submit a SQL query and return an asynchronous iterator.
 
         :param timeout: An optional timeout.
         :param lock: Optional key to perform the query with locking. If not set, the query is performed without any
             lock.
-        :param streaming_mode: If ``True`` the data fetching is performed in streaming mode, that is iterating over the
-            cursor and yielding once a time (requires an opening connection to do that). Otherwise, all the data is
-            fetched and keep in memory before yielding it.
         :param kwargs: Additional named arguments.
         :return: This method does not return anything.
         """
         await self._create_cursor()
 
-        if streaming_mode:
-            async for row in self._cursor:
-                yield row
-            return
-
-        rows = await self._cursor.fetchall()
-
-        for row in rows:
+        async for row in self._cursor:
             yield row
 
     # noinspection PyUnusedLocal
@@ -183,7 +175,11 @@ async def _create_cursor(self, *args, lock: Optional[Hashable] = None, **kwargs)
         if self._cursor is None:
             self._cursor = await self._connection.cursor(*args, **kwargs)
 
-        if lock is not None and self._lock is None:
+        if lock is not None:
+            if self._lock is not None:
+                if self._lock.key != lock:
+                    raise ValueError(f"Only one lock per instance is supported. Currently locked with {self._lock!r}")
+                return
             from ..locks import (
                 DatabaseLock,
             )
@@ -197,10 +193,18 @@ async def _destroy_cursor(self, **kwargs):
             self._lock = None
 
         if self._cursor is not None:
-            if self._cursor.closed:
+            if not self._cursor.closed:
                 self._cursor.close()
             self._cursor = None
 
+    @property
+    def lock(self) -> Optional[DatabaseLock]:
+        """Get the lock.
+
+        :return: A ``DatabaseLock`` instance.
+        """
+        return self._lock
+
     @property
     def cursor(self) -> Optional[Cursor]:
         """Get the cursor.
@@ -209,6 +213,14 @@ def cursor(self) -> Optional[Cursor]:
         """
         return self._cursor
 
+    @property
+    def notifications(self) -> ClosableQueue:
+        """Get the notifications queue.
+
+        :return: A ``ClosableQueue`` instance.
+        """
+        return self._connection.notifies
+
     @property
     def connection(self) -> Optional[Connection]:
         """Get the connection.

packages/core/minos-microservice-common/tests/test_common/test_database/test_clients/__init__.py

@@ -0,0 +1,176 @@
+import unittest
+from unittest.mock import (
+    PropertyMock,
+    call,
+    patch,
+)
+
+import aiopg
+from aiopg import (
+    Connection,
+    Cursor,
+)
+from psycopg2 import (
+    IntegrityError,
+    OperationalError,
+)
+
+from minos.common import (
+    AiopgDatabaseClient,
+    DatabaseLock,
+    IntegrityException,
+    UnableToConnectException,
+)
+from minos.common.testing import (
+    PostgresAsyncTestCase,
+)
+from tests.utils import (
+    CommonTestCase,
+)
+
+
+# noinspection SqlNoDataSourceInspection
+class TestAiopgDatabaseClient(CommonTestCase, PostgresAsyncTestCase):
+    def setUp(self):
+        super().setUp()
+        self.sql = "SELECT * FROM information_schema.tables"
+
+    def test_constructor(self):
+        client = AiopgDatabaseClient("foo")
+        self.assertEqual("foo", client.database)
+        self.assertEqual("postgres", client.user)
+        self.assertEqual("", client.password)
+        self.assertEqual("localhost", client.host)
+        self.assertEqual(5432, client.port)
+
+    def test_from_config(self):
+        default_database = self.config.get_default_database()
+        client = AiopgDatabaseClient.from_config(self.config)
+        self.assertEqual(default_database["database"], client.database)
+        self.assertEqual(default_database["user"], client.user)
+        self.assertEqual(default_database["password"], client.password)
+        self.assertEqual(default_database["host"], client.host)
+        self.assertEqual(default_database["port"], client.port)
+
+    async def test_is_valid_true(self):
+        async with AiopgDatabaseClient.from_config(self.config) as client:
+            self.assertTrue(client.is_valid())
+
+    async def test_is_valid_false_not_setup(self):
+        client = AiopgDatabaseClient.from_config(self.config)
+        self.assertFalse(await client.is_valid())
+
+    async def test_is_valid_false_operational_error(self):
+        async with AiopgDatabaseClient.from_config(self.config) as client:
+            with patch.object(Connection, "isolation_level", new_callable=PropertyMock, side_effect=OperationalError):
+                self.assertFalse(await client.is_valid())
+
+    async def test_is_valid_false_closed(self):
+        async with AiopgDatabaseClient.from_config(self.config) as client:
+            with patch.object(Connection, "closed", new_callable=PropertyMock, return_valud=False):
+                self.assertFalse(await client.is_valid())
+
+    async def test_connection(self):
+        client = AiopgDatabaseClient.from_config(self.config)
+        self.assertIsNone(client.connection)
+        async with client:
+            self.assertIsInstance(client.connection, Connection)
+        self.assertIsNone(client.connection)
+
+    async def test_connection_raises(self):
+        with patch.object(aiopg, "connect", new_callable=PropertyMock, side_effect=OperationalError):
+            with self.assertRaises(UnableToConnectException):
+                async with AiopgDatabaseClient.from_config(self.config):
+                    pass
+
+    async def test_notifications(self):
+        async with AiopgDatabaseClient.from_config(self.config) as client:
+            self.assertEqual(client.connection.notifies, client.notifications)
+
+    async def test_cursor(self):
+        client = AiopgDatabaseClient.from_config(self.config)
+        self.assertIsNone(client.cursor)
+        async with client:
+            self.assertIsNone(client.cursor)
+            await client.execute("SELECT * FROM information_schema.tables")
+            self.assertIsInstance(client.cursor, Cursor)
+
+        self.assertIsNone(client.cursor)
+
+    async def test_cursor_reset(self):
+        client = AiopgDatabaseClient.from_config(self.config)
+        async with client:
+            await client.execute("SELECT * FROM information_schema.tables")
+            self.assertIsInstance(client.cursor, Cursor)
+            await client.reset()
+            self.assertIsNone(client.cursor)
+
+    async def test_lock(self):
+        client = AiopgDatabaseClient.from_config(self.config)
+        self.assertIsNone(client.lock)
+        async with client:
+            self.assertIsNone(client.lock)
+            await client.execute(self.sql, lock="foo")
+            self.assertIsInstance(client.lock, DatabaseLock)
+
+        self.assertIsNone(client.lock)
+
+    async def test_lock_reset(self):
+        async with AiopgDatabaseClient.from_config(self.config) as client:
+            await client.execute(self.sql, lock="foo")
+            self.assertIsInstance(client.lock, DatabaseLock)
+            await client.reset()
+            self.assertIsNone(client.lock)
+
+    async def test_execute(self):
+        async with AiopgDatabaseClient.from_config(self.config) as client:
+            with patch.object(Cursor, "execute") as execute_mock:
+                await client.execute(self.sql)
+        self.assertEqual(
+            [call(operation=self.sql, parameters=None, timeout=None)],
+            execute_mock.call_args_list,
+        )
+
+    async def test_execute_with_lock(self):
+        with patch.object(DatabaseLock, "__aenter__") as enter_lock_mock:
+            with patch.object(DatabaseLock, "__aexit__") as exit_lock_mock:
+                async with AiopgDatabaseClient.from_config(self.config) as client:
+                    await client.execute(self.sql, lock="foo")
+                    self.assertEqual(1, enter_lock_mock.call_count)
+                    self.assertEqual(0, exit_lock_mock.call_count)
+                    enter_lock_mock.reset_mock()
+                    exit_lock_mock.reset_mock()
+            self.assertEqual(0, enter_lock_mock.call_count)
+            self.assertEqual(1, exit_lock_mock.call_count)
+
+    async def test_execute_with_lock_multiple(self):
+        async with AiopgDatabaseClient.from_config(self.config) as client:
+            await client.execute(self.sql, lock="foo")
+            await client.execute(self.sql, lock="foo")
+            with self.assertRaises(ValueError):
+                await client.execute(self.sql, lock="bar")
+
+    async def test_execute_raises_integrity(self):
+        async with AiopgDatabaseClient.from_config(self.config) as client:
+            with patch.object(Cursor, "execute", side_effect=IntegrityError):
+                with self.assertRaises(IntegrityException):
+                    await client.execute(self.sql)
+
+    async def test_fetch_one(self):
+        async with AiopgDatabaseClient.from_config(self.config) as client:
+            await client.execute(self.sql)
+            observed = await client.fetch_one()
+        self.assertIsInstance(observed, tuple)
+
+    async def test_fetch_all(self):
+        async with AiopgDatabaseClient.from_config(self.config) as client:
+            await client.execute(self.sql)
+            observed = [value async for value in client.fetch_all()]
+
+        self.assertGreater(len(observed), 0)
+        for obs in observed:
+            self.assertIsInstance(obs, tuple)
+
+
+if __name__ == "__main__":
+    unittest.main()