Flask应用程序不像unit testing那样从同一个数据库检索数据

我使用docker构build了一个应用程序，其中有三个容器。 一个托pipe后台数据库（ db ），一个用于web api（ app ），另一个用于运行自动化testing（ test ）。 这是描述三个容器之间链接docker-compose.yml的副本。

 version: "3" services: app: build: context: ../ dockerfile: docker/app.Dockerfile environment: DB_NAME: ${PSQL_DATABASE} DB_USER: ${PSQL_USER} DB_PASSWORD: ${PSQL_PASSWORD} DB_PORT: ${PSQL_PORT} API_HOST: ${API_HOST} API_PORT: ${API_PORT} ports: - "5000:5000" depends_on: - db networks: - todo_backend tests: build: context: ../ dockerfile: docker/tests.Dockerfile environment: DB_NAME: ${PSQL_DATABASE} DB_USER: ${PSQL_USER} DB_PASSWORD: ${PSQL_PASSWORD} DB_PORT: ${PSQL_PORT} API_HOST: ${API_HOST} API_PORT: ${API_PORT} ports: - "80:8000" depends_on: - db - app networks: - todo_backend db: image: postgres:9.5 restart: on-failure environment: POSTGRES_PASSWORD: changeme POSTGRES_USER: docker POSTGRES_DB: db ports: - "5432" networks: - todo_backend networks: todo_backend: driver: bridge 

我写了一个unit testing，在这里我添加一些数据到后端数据库，并试图通过http请求通过web api检索它。

 def test_retrieve_task(self, session, task): # given session.add(task) session.commit() assert len(Task.query.filter_by(id=task.id).all()) == 1 assert Task.query.filter_by(id=task.id).all()[0].id == 1 # when response = requests.get( 'http://docker_app_1:5000/todo/api/v1.0/tasks/1') # then assert response.status_code == 200 assert response.json() == { 'json_list': { 'id': task.id, 'description': task.description } } 

这不工作，因为我期望，事实上，即使前两个assert声明通过，似乎烧瓶应用程序不是从同一个数据库中读取。

 tests_1 | def test_retrieve_task(self, session, task): tests_1 | # given tests_1 | session.add(task) tests_1 | session.commit() tests_1 | tests_1 | assert len(Task.query.filter_by(id=task.id).all()) == 1 tests_1 | assert Task.query.filter_by(id=task.id).all()[0].id == 1 tests_1 | tests_1 | # when tests_1 | response = requests.get( tests_1 | 'http://docker_app_1:5000/todo/api/v1.0/tasks/1') tests_1 | tests_1 | # then tests_1 | assert response.status_code == 200 tests_1 | > assert response.json() == { tests_1 | 'json_list': { tests_1 | 'id': task.id, tests_1 | 'description': task.description tests_1 | } tests_1 | } tests_1 | E AssertionError: assert {'json_list': []} == {'json_list': {'desc...road bike', 'id': 1}} tests_1 | E Differing items: tests_1 | E {'json_list': []} != {'json_list': {'description': 'service road bike', 'id': 1}} tests_1 | E Use -v to get the full diff tests_1 | tests_1 | tests/test_api.py:42: AssertionError tests_1 | ===================== 1 failed, 14 passed in 1.92 seconds ====================== 

最后一个assert失败，因为Web应用程序从一个看起来是空的数据库中读取数据，即使在我刚刚向数据库提交的一些数据的同一个testing中。

这是views.py文件的样子：

 from flask import Blueprint from flask import abort, current_app, g, make_response, jsonify from todo.models import Task todo = Blueprint('todo', __name__) @todo.route('/todo/api/v1.0/tasks/1', methods=['GET']) def get_task(): tasks = Task.query.all() return jsonify(json_list=[todo.as_dict() for todo in tasks]) 

这里是conftest.py ，它解释了我如何为我的testing构build灯具：

 import datetime import os import pytest from sqlalchemy import create_engine from sqlalchemy.orm import Session from app import config from app.factory import create_app from todo.models import db as _db from todo.models import Board, State, Task, User @pytest.yield_fixture(scope='session') def app(request): """Session-wide test Flask application.""" app = create_app(database_uri=config.TEST_DATABASE_URI, debug=True) context = app.app_context() context.push() yield app context.pop() @pytest.yield_fixture(scope='session') def db(app, request): """Session-wide test database.""" _db.app = app _db.create_all() yield _db _db.drop_all() # a bug in flask-sqlalchemy prevents empty dictionaries from being correctly # interpreted when passed as binds class _dict(dict): def __nonzero__(self): return True @pytest.yield_fixture(scope='function') def session(db, request): """Create a new database session for a test.""" connection = db.engine.connect() transaction = connection.begin() options = dict(bind=connection, binds=_dict()) session = db.create_scoped_session(options=options) db.session = session yield session transaction.rollback() connection.close() session.remove() @pytest.fixture(scope='function') def task(backlog, tomorrow, user, hit): return Task(description='service road bike', assignee=user, state=hit, deadline=tomorrow) 

最后这是models.py ：

 from flask_sqlalchemy import SQLAlchemy from sqlalchemy import create_engine from sqlalchemy import Column, Date, Integer, String, Text, ForeignKey from sqlalchemy.ext.declarative import declarative_base from sqlalchemy.orm import relationship, sessionmaker from sqlalchemy.schema import UniqueConstraint db = SQLAlchemy() ... class Task(db.Model): __tablename__ = 'task' id = Column(Integer, primary_key=True) description = Column(Text) state_id = Column(Integer, ForeignKey('state.id')) state = relationship('State', back_populates='tasks') user_id = Column(Integer, ForeignKey('user.id')) assignee = relationship('User', back_populates='tasks') deadline = Column(Date) def as_dict(self): return { 'id': self.id, 'description': self.description } def __repr__(self): return '<Task(description={}, state={}, deadline={})>'.format( self.description, self.state, self.deadline)