add test code (#37)

* add test code

* add a pre test prompt

* exec pre-commit

Author: wu-clan

.gitignore

@@ -7,3 +7,4 @@ backend/app/log/
 backend/app/alembic/versions/
 backend/app/static/media/
 .ruff_cache/
+.pytest_cache/

README.md

@@ -70,3 +70,21 @@ git clone https://github.com/wu-clan/fastapi_best_architecture.git
 ## Init the test data
 
 Execute the `backend/app/init_test_data.py` file
+
+## Test
+
+Perform tests via pytest
+
+**Tip**: Before the test starts, please execute init the test data first, also, the fastapi service needs to be started
+
+1. First, go to the app directory
+
+   ```shell
+   cd backend/app/
+   ```
+
+2. Execute the test command
+
+   ```shell
+   pytest -vs --disable-warnings
+   ```

backend/app/init_test_data.py

@@ -17,6 +17,22 @@ class InitData:
     def __init__(self):
         self.fake = Faker('zh_CN')
 
+    @staticmethod
+    async def create_test_user():
+        """创建测试用户"""
+        username = 'test'
+        password = 'test'
+        email = '[email protected]'
+        user_obj = User(
+            username=username,
+            password=get_hash_password(password),
+            email=email,
+            is_superuser=True,
+        )
+        async with async_db_session.begin() as db:
+            db.add(user_obj)
+        log.info(f'测试用户创建成功，账号：{username}，密码：{password}')
+
     @staticmethod
     async def create_superuser_by_yourself():
         """手动创建管理员账户"""
@@ -108,6 +124,7 @@ async def fake_no_active_superuser(self):
     async def init_data(self):
         """自动创建数据"""
         log.info('⏳ 开始初始化数据')
+        await self.create_test_user()
         await self.create_superuser_by_yourself()
         await self.fake_user()
         await self.fake_no_active_user()

backend/app/test/conftest.py

@@ -0,0 +1,31 @@
+#!/usr/bin/env python3
+# -*- coding: utf-8 -*-
+import sys
+
+import pytest
+from httpx import AsyncClient
+
+sys.path.append('../../')
+
+from backend.app.common.redis import redis_client  # noqa: E402
+from backend.app.core.conf import settings  # noqa: E402
+
+
+@pytest.fixture(scope='session')
+def anyio_backend():
+    return 'asyncio'
+
+
+@pytest.fixture(scope='package', autouse=True)
+async def function_fixture(anyio_backend):
+    auth_data = {
+        'url': f'http://{settings.UVICORN_HOST}:{settings.UVICORN_PORT}{settings.TOKEN_URL}',
+        'headers': {'accept': 'application/json', 'Content-Type': 'application/x-www-form-urlencoded'},
+        'data': {'username': 'test', 'password': 'test'},
+    }
+    async with AsyncClient() as client:
+        response = await client.post(**auth_data)
+        token = response.json()['access_token']
+        test_token = await redis_client.get('test_token')
+        if not test_token:
+            await redis_client.set('test_token', token, ex=86400)

backend/app/test/test_auth.py

@@ -0,0 +1,72 @@
+#!/usr/bin/env python3
+# -*- coding: utf-8 -*-
+import sys
+
+import pytest
+from faker import Faker
+from httpx import AsyncClient
+
+sys.path.append('../../')
+
+from backend.app.core.conf import settings  # noqa: E402
+from backend.app.main import app  # noqa: E402
+from backend.app.common.redis import redis_client  # noqa: E402
+
+
+class TestAuth:
+    pytestmark = pytest.mark.anyio
+    faker = Faker(locale='zh_CN')
+    users_api_base_url = f'http://{settings.UVICORN_HOST}:{settings.UVICORN_PORT}/v1/auth/users'
+
+    @property
+    async def get_token(self):
+        token = await redis_client.get('test_token')
+        return token
+
+    async def test_login(self):
+        async with AsyncClient(
+            app=app, headers={'accept': 'application/json', 'Content-Type': 'application/x-www-form-urlencoded'}
+        ) as client:
+            response = await client.post(
+                url=f'{self.users_api_base_url}/login', data={'username': '1', 'password': '1'}
+            )
+        assert response.status_code == 200
+        assert response.json()['token_type'] == 'Bearer'
+
+    async def test_register(self):
+        async with AsyncClient(
+            app=app, headers={'accept': 'application/json', 'Content-Type': 'application/json'}
+        ) as client:
+            response = await client.post(
+                url=f'{self.users_api_base_url}/register',
+                json={
+                    'username': f'{self.faker.user_name()}',
+                    'password': f'{self.faker.password()}',
+                    'email': f'{self.faker.email()}',
+                },
+            )
+        assert response.status_code == 200
+        r_json = response.json()
+        assert r_json['code'] == 200
+        assert r_json['msg'] == 'Success'
+
+    async def test_get_userinfo(self):
+        async with AsyncClient(
+            app=app, headers={'accept': 'application/json', 'Authorization': f'Bearer {await self.get_token}'}
+        ) as client:
+            response = await client.get(url=f'{self.users_api_base_url}/1')
+        assert response.status_code == 200
+        r_json = response.json()
+        assert r_json['code'] == 200
+        assert r_json['msg'] == 'Success'
+
+    async def test_get_all_users(self):
+        async with AsyncClient(
+            app=app, headers={'accept': 'application/json', 'Authorization': f'Bearer {await self.get_token}'}
+        ) as client:
+            response = await client.get(url=f'{self.users_api_base_url}?page=1&size=20')
+        assert response.status_code == 200
+        r_json = response.json()
+        assert isinstance(r_json['data'], list)
+        assert isinstance(r_json['links'], dict)
+        assert isinstance(r_json['links']['self'], str)

requirements.txt

@@ -12,11 +12,14 @@ fast-captcha==0.1.3
 fastapi==0.95.0
 fastapi-pagination==0.12.1
 gunicorn==20.1.0
+httpx==0.23.0
 loguru==0.6.0
 passlib==1.7.4
 path==15.1.2
 pre-commit==3.2.2
 pydantic==1.10.5
+pytest==7.2.2
+pytest-pretty==1.2.0
 python-jose==3.3.0
 python-multipart==0.0.5
 redis==4.5.4