test: use asynctest to test async methods

Author: NightMarcher

Pipfile

@@ -8,6 +8,7 @@ tortoise-orm = "*"
 
 [dev-packages]
 aiomysql = "*"
+asynctest = "*"
 autopep8 = "*"
 cryptography = "*"
 faker = "*"

README.md

@@ -163,8 +163,8 @@ Generate sql and execute
 ```python
 await AccountMgr.bulk_update_from_dicts(
     [
-        {'id': 7, 'active': False, 'gender': <GenderEnum.male: 1>},
-        {'id': 15, 'active': True, 'gender': <GenderEnum.unknown: 0>}
+        {'id': 7, 'active': False, 'gender': GenderEnum.male},
+        {'id': 15, 'active': True, 'gender': GenderEnum.unknown}
     ],
     join_fields=["id"],
     update_fields=["active", "gender"],

requirements-dev.txt

@@ -1,4 +1,5 @@
 aiomysql
+asynctest
 autopep8
 cryptography
 faker

tests/__init__.py

@@ -2,6 +2,8 @@
 
 from tortoise import Tortoise
 
+TEST_CONN = "test"
+
 
 def init_test_orm():
     """
@@ -11,7 +13,7 @@ def init_test_orm():
     config = {
         "timezone": "Asia/Shanghai",
         "connections": {
-            "test": {
+            TEST_CONN: {
                 "engine": "tortoise.backends.mysql",
                 "credentials": {
                     "host": "localhost",
@@ -27,11 +29,15 @@ def init_test_orm():
         "apps": {
             "demo": {
                 "models": ["examples.service.models.demo"],
-                "default_connection": "test"
+                "default_connection": TEST_CONN
             }
         }
     }
     asyncio.run(Tortoise.init(config=config))
 
 
+def get_test_conn():
+    return Tortoise.get_connection(TEST_CONN)
+
+
 init_test_orm()

tests/test_cursor_handler.py

@@ -0,0 +1,30 @@
+from logging import getLogger
+
+from asynctest import CoroutineMock, TestCase, patch
+
+from . import get_test_conn
+from fastapi_esql import CursorHandler
+
+logger = getLogger(__name__)
+
+
+class TestCursorHandler(TestCase):
+
+    conn = get_test_conn()
+
+    async def test_fetch_dicts(self):
+        with patch(
+            "tortoise.backends.mysql.client_class.execute_query_dict",
+            # new=CoroutineMock()
+        ) as mock_exec:
+            mock_exec.side_effect = Exception("Error")
+            assert await CursorHandler.fetch_dicts("SELECT true", self.conn, logger) is None
+
+    @patch("tortoise.backends.mysql.client_class.execute_query")
+    async def test_sum_row_cnt(self, mock_exec):
+        mock_exec.return_value = 10, object()
+        assert await CursorHandler.sum_row_cnt("SELECT true", self.conn, logger) == 10
+
+    async def test_exec_if_ok(self):
+        # True means test connection is reachable
+        assert await CursorHandler.exec_if_ok("SELECT true", self.conn, logger) == True

tests/test_sqlizer.py

@@ -1,14 +1,13 @@
 from datetime import datetime
 from unittest import TestCase
 
+from examples.service.models.demo import Account
+from examples.service.constants.enums import GenderEnum, LocaleEnum
 from fastapi_esql import (
     Cases, RawSQL, SQLizer,
     Q, QsParsingError, WrongParamsError,
 )
 
-from examples.service.models.demo import Account
-from examples.service.constants.enums import GenderEnum, LocaleEnum
-
 
 class TestRawSQL(TestCase):