Flask+ Dependency-injecter+pytest 写测试类-编程知识

Flask+ Dependency-injecter+pytest 写测试类

最近在使用这几个在做项目，因为第一次用这个，所以不免有些问题。总结下踩的坑

1.测试类位置

首先测试类约定会放在tests里面，不然有可能发生引入包的问题，会报错某些包找不到。

2. 测试类依赖注入

这里我就用的真实的数据库操作,但是我用了一个专门为测试写的事务管理，所有操作都不会commit而是会rollback，相当于一个内存数据库了，只会在内存里面，不会提交，你也可以直接用内存数据库或者mock,

# tests/test_app.py
import pytest
from main.application import create_app
from main.config.database_config import DatabaseConfig
from main.containers import Container
from dependency_injector.wiring import inject, Provide
from main.services.common_service.db_access.domain.user import User
from main.services.common_service.db_access.service.user_service import UserService
from dependency_injector.providers import Factory
from main.services.common_service.db_access.repository.impl.user_repository_impl import UserRepositoryImpl#获取app
@pytest.fixture
def app():app = create_app()with app.app_context():yield app
#获取user_service
@pytest.fixture
def user_service(app):with app.app_context():user_repo=UserRepositoryImpl()db=app.container.db()yield UserService(user_repository=user_repo,session_factory=db.force_rowback_session)#测试方法，需要传入user_serivice，会从上面加了@pytest.fixture注解获取同名方法进行注入
def test_create_user(user_service):user=user_service.create_user()assert user is not None

我的userservice需要传入两个参数，一个是repo的实现类，一个是sqla的session

"""Containers module."""import os
from dependency_injector import containers, providersfrom main.config.database_config import DatabaseConfigclass Container(containers.DeclarativeContainer):# wiring_config = containers.WiringConfiguration(auto_wire=True)wiring_config = containers.WiringConfiguration(packages=["main.config", "main.web.controller","main"])config_path = os.path.join(os.path.dirname(__file__), "../config.yml")config = providers.Configuration(yaml_files=[config_path])db=providers.Singleton(DatabaseConfig,db_url=config.db.url)

有个很重要的一点就是
这里如果写成这样过的话，启动项目是没有什么问题。但是测试类的时候就会加载不到，就会导致需要config的类加载不到你需要的配置。

config = providers.Configuration(yaml_files=['config.yml'])

所以必须写成这样

config_path = os.path.join(os.path.dirname(__file__), "../config.yml")
config = providers.Configuration(yaml_files=[config_path])

事务控制

"""Database module."""from contextlib import contextmanager, AbstractContextManager
from typing import Callablefrom sqlalchemy import create_engine, orm,event
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import Session
from datetime import datetimefrom main.services.common_service.db_access.domain.common_field_entity import CommonEntity
Base = declarative_base()class DatabaseConfig:def __init__(self, db_url: str) -> None:self._engine = create_engine(db_url, echo=True)self._session_factory = orm.scoped_session(orm.sessionmaker(autocommit=False,autoflush=False,expire_on_commit=False,bind=self._engine,),)def create_database(self) -> None:Base.metadata.create_all(self._engine)@contextmanagerdef session(self) -> Callable[..., AbstractContextManager[Session]]:session: Session = self._session_factory()try:yield sessionexcept Exception:session.rollback()raiseelse:if session._transaction.is_active:session.commit()session.close()@contextmanager#专门负责测试类的回滚操作，不论任何情况，都进行回滚操作def force_rowback_session(self) -> Callable[..., AbstractContextManager[Session]]:session: Session = self._session_factory()try:yield sessionexcept Exception:session.rollback()raiseelse:session.rollback()session.close()@event.listens_for(CommonEntity, 'before_insert', propagate=True)def before_insert_listener(self, mapper, target):# 在创建时自动更新 created_dt，versiontarget.created_dt = datetime.now()target.created_by = 'MAAS'target.version = 1@event.listens_for(CommonEntity, 'before_update', propagate=True)def before_update_listener(self, mapper, target):# 在更新时自动更新 updated_dt，versiontarget.updated_dt = datetime.now()target.updated_by = 'MAAS'target.version += 1