Flask+ Dependency-injecter+pytest 写测试类-Toy模板网

这篇具有很好参考价值的文章主要介绍了Flask+ Dependency-injecter+pytest 写测试类。希望对大家有所帮助。如果存在错误或未考虑完全的地方，请大家不吝赐教，您也可以点击"举报违法"按钮提交疑问。

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

1.测试类位置

Flask+ Dependency-injecter+pytest 写测试类,Python,pytest

首先测试类约定会放在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, providers

from main.config.database_config import DatabaseConfig



class 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 Callable

from sqlalchemy import create_engine, orm,event
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import Session
from datetime import datetime

from 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)


    @contextmanager
    def session(self) -> Callable[..., AbstractContextManager[Session]]:
        session: Session = self._session_factory()
        try:
            yield session
        except Exception:
            session.rollback()
            raise
        else:
            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 session
        except Exception:
            session.rollback()
            raise
        else:
            session.rollback()
            #rollback之后就不能再close了，否则就会报错
            #session.close()

    @event.listens_for(CommonEntity, 'before_insert', propagate=True)
    def before_insert_listener(self, mapper, target):
        # 在创建时自动更新 created_dt，version
        target.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，version
        target.updated_dt = datetime.now()
        target.updated_by = 'MAAS'
        target.version += 1