1

我已经使用 python unittest 为烧瓶编写了一些单元测试/集成测试。我广泛使用此处编写的示例代码：https ://blog.miguelgrinberg.com/post/the-flask-mega-tutorial-part-vii-unit-testing-legacy

这些测试过去可以完美运行，但现在我刚刚尝试再次运行它们并遇到错误：

sqlalchemy.exc.OperationalError: (pymysql.err.OperationalError) (2003, "Can't connect to MySQL server on 'cloud-sql' ([Errno -2] Name or service not known)")

（此错误的背景：http ://sqlalche.me/e/e3q8 ）

显然，这告诉我它正在尝试连接到 cloud-sql 并且不能（这是部署后的正确 DNS 名称）。但是，我为测试配置了不同的 URI。

这就是我的 test_web_api.py 的样子：

import os
import unittest
import json
from flask import url_for
from requests.auth import _basic_auth_str
from app import app, db
from app.models import User, UserToken, UserRoles, RolesList, BandwidthResults

TEST_DB = 'test.db'


class API_Tests(unittest.TestCase):

    ############################
    #### setup and teardown ####
    ############################
    test_admin_password = 'testpass'
    test_admin_user = 'admin'
    api_token = None
    # executed prior to each test

    def setUp(self):
        app.config['TESTING'] = True
        app.config['WTF_CSRF_ENABLED'] = False
        app.config['DEBUG'] = False
        app.config['SQLALCHEMY_ENGINE_OPTIONS'] = {}
        app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///' + \
            os.path.join(app.config['BASEDIR'], TEST_DB)
        app.config['SERVER_NAME'] = '127.0.0.1 localhost.dev'
        self.client = app.test_client(use_cookies=True)
        self.app_context = app.app_context()
        self.app_context.push()
        
        db.drop_all()
        db.create_all()

        # # Create admin user
        user = User(username=self.test_admin_user, email='test@example.com')
        user.set_password(self.test_admin_password)
        db.session.add(user)
        # Give admin user admin roles
        admin_role = RolesList(name='admin')
        db.session.add(admin_role)
        db.session.commit()
        user_role = UserRoles(role=RolesList.query.first().id,
                              user_id=User.query.filter_by(username=self.test_admin_user).first().id)
        db.session.add(user_role)
        # Generate an API token
        db.session.commit()
        response = self.client.get(url_for('api.token_token_api'), headers={
                                   "Authorization": _basic_auth_str(self.test_admin_user, self.test_admin_password)},
                                   query_string={'expiry': 300})
        self.api_token = json.loads(response.data)['token']

    # executed after each test
    def tearDown(self):
        db.session.remove()
        db.drop_all()
        self.app_context.pop()
        pass

###########################
#### tests-integration ####
###########################

    def test_01_api_token_get_auth(self):
        response = self.client.get(url_for('api.token_token_api'), headers={
                                   "Authorization": _basic_auth_str(self.test_admin_user, self.test_admin_password)})
        self.assertEqual(response.status_code, 200)
        self.assertIsNotNone(json.loads(response.data)['token'])

if __name__ == "__main__":
    unittest.main()

这就是我__init__.py的“应用程序”的样子

from flask import Flask

from config import Config
from flask_sqlalchemy import SQLAlchemy
from flask_migrate import Migrate
from flask_login import LoginManager
from flask_socketio import SocketIO
from flask_marshmallow import Marshmallow
from flask_principal import Principal
import sentry_sdk
from sentry_sdk.integrations.flask import FlaskIntegration

app = Flask(__name__)

from app.tasks.init import make_celery  # noqa: E402

app.config.from_object(Config)

sentry_sdk.init(
    dsn="",
    integrations=[FlaskIntegration()],
    environment=app.config['SENTRY_ENV'],
    release=app.config['SENTRY_RELEASE'],

)

db = SQLAlchemy(app)
migrate = Migrate(app, db)
login = LoginManager(app)
login.login_view = 'login'
socketio = SocketIO(app, message_queue=app.config['CELERY_BROKER_URL'], async_mode='gevent', cors_allowed_origins="")
ma = Marshmallow(app)
celery = make_celery(app)
root_path = app.root_path
principal = Principal(app)

from app import routes, models  # noqa: E402,F401
from app.apis import blueprint as api  # noqa: E402

app.register_blueprint(api, url_prefix='/api/1')

我不知道为什么。代码库没有任何变化，一切都在 docker 镜像中运行，带有固定版本，所有包在 requirements.txt 中都有固定版本。感谢任何提示。

4

1 回答 1

0

我通过在更新应用程序配置后删除会话解决了这个问题（这是在我的 setUp() 函数中）

注意 db.session.remove()
这似乎清除了仍在尝试访问旧值数据库的当前数据库会话。

def setUp(self):
        app.config['TESTING'] = True
        app.config['WTF_CSRF_ENABLED'] = False
        app.config['DEBUG'] = False
        app.config['SQLALCHEMY_ENGINE_OPTIONS'] = {}
        app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///' + \
            os.path.join(app.config['BASEDIR'], TEST_DB)
        app.config['SERVER_NAME'] = '127.0.0.1 localhost.dev'
        self.client = app.test_client(use_cookies=True)
        self.app_context = app.app_context()
        self.app_context.push()

        db.session.remove()

        db.drop_all()
        db.create_all()
        # Create admin user
        user = User(username=self.test_admin_user, email='test@example.com')
        user.set_password(self.test_admin_password)
        db.session.add(user)
        # Give admin user admin roles
        admin_role = RolesList(name='admin')
        db.session.add(admin_role)
        db.session.commit()
        user_role = UserRoles(role=RolesList.query.first().id,
                              user_id=User.query.filter_by(username=self.test_admin_user).first().id)
        db.session.add(user_role)
        # Generate an API token
        db.session.commit()
        response = self.client.get(url_for('api.token_token_api'), headers={
                                   "Authorization": _basic_auth_str(self.test_admin_user, self.test_admin_password)},
                                   query_string={'expiry': 300})
        self.api_token = json.loads(response.data)['token']
于 2020-09-18T11:33:44.290 回答