添加测试方法

This commit is contained in:
haotian 2025-07-28 09:48:26 +08:00
parent 8cca070349
commit 7470dd2bd8
6 changed files with 102 additions and 0 deletions

View File

@ -25,6 +25,7 @@ loginController = APIRouter()
async def login(
request: Request, form_data: CustomOAuth2PasswordRequestForm = Depends(), query_db: AsyncSession = Depends(get_db)
):
# 是否启用验证码
captcha_enabled = (
True
if await request.app.state.redis.get(f'{RedisInitKeyConfig.SYS_CONFIG.key}:sys.account.captchaEnabled')

View File

@ -0,0 +1,52 @@
from fastapi import APIRouter, Body, Depends, Form, Request
from module_admin.service.login_service import LoginService
from module_admin.entity.vo.test_vo import TestUserName, TestPostForm, TestUserNamePage
from typing import List, Optional
from module_admin.service.test_service import TestService
from config.get_db import get_db
from sqlalchemy.ext.asyncio import AsyncSession
from module_admin.annotation.log_annotation import Log
from config.enums import BusinessType, RedisInitKeyConfig
from utils.response_util import ResponseUtil
testController = APIRouter(prefix='/test')
@testController.get('/hello', response_model= List[TestUserName]
# ,dependencies=[Depends(LoginService.get_current_user)]
)
async def hello(request: Request,
name: Optional[str] = None,
query_db: AsyncSession = Depends(get_db), ):
print("名称", name)
user_name = await TestService.get_user_name(query_db,)
return user_name
@testController.post('/hello_post_json', response_model=str)
async def hello_post_json(
request: Request,
login_info: TestUserNamePage,
query_db: AsyncSession = Depends(get_db)
):
print(login_info.user_name)
print(login_info.page_num)
print(login_info.page_size)
# print(login_info.password)
user_name = await TestService.get_user_name(query_db, login_info)
return ResponseUtil.success(model_content=user_name)
@Log(title='测试post_form_data', business_type=BusinessType.OTHER, log_type='post')
@testController.post('/hello_post_form_data', response_model=str)
async def hello_post_form_data(
request: Request,
user_name: Optional[str] = Form(),
password: Optional[str] = Form()
):
print("*"*100)
print(user_name)
print(password)
return "success"

View File

@ -0,0 +1,16 @@
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
from module_admin.entity.do.user_do import SysUser
from utils.page_util import PageUtil
from module_admin.entity.vo.test_vo import TestUserNamePage
class TestDao:
@classmethod
async def get_user_name(cls, db: AsyncSession, query_paramter: TestUserNamePage):
query = (
select(SysUser.user_name)
)
# user_name = await db.execute(select(SysUser.user_name))
user_name = await PageUtil.paginate(db, query, query_paramter.page_num, query_paramter.page_size, True)
return user_name

View File

@ -0,0 +1,21 @@
from typing import Optional
from pydantic import BaseModel, Field, ConfigDict
from module_admin.annotation.pydantic_annotation import as_query
from pydantic.alias_generators import to_camel
class TestUserName(BaseModel):
model_config = ConfigDict(alias_generator=to_camel, from_attributes=True)
user_name: Optional[str] = Field(default=None, description='用户名')
class TestPostForm(BaseModel):
user_name: Optional[str] = None
password: Optional[str] = None
@as_query
class TestUserNamePage(TestUserName):
page_num: int = Field(default=1, description='当前页码')
page_size: int = Field(default=10, description='每页记录数')

View File

@ -0,0 +1,10 @@
from module_admin.dao.test_dao import TestDao
from sqlalchemy.ext.asyncio import AsyncSession
from module_admin.entity.vo.test_vo import TestUserNamePage
class TestService:
@staticmethod
async def get_user_name(db: AsyncSession, query_paramter:TestUserNamePage):
user_name = await TestDao.get_user_name(db, query_paramter)
return user_name

View File

@ -22,6 +22,7 @@ from module_admin.controller.post_controler import postController
from module_admin.controller.role_controller import roleController
from module_admin.controller.server_controller import serverController
from module_admin.controller.user_controller import userController
from module_admin.controller.test_controller import testController
from module_generator.controller.gen_controller import genController
from sub_applications.handle import handle_sub_applications
from utils.common_util import worship
@ -79,6 +80,7 @@ controller_list = [
{'router': cacheController, 'tags': ['系统监控-缓存监控']},
{'router': commonController, 'tags': ['通用模块']},
{'router': genController, 'tags': ['代码生成']},
{'router': testController, 'tags': ['测试模块']},
]
for controller in controller_list: