我的项目结构如下-
基本文件夹/
|-- 公共模块 (python)
|--其他常用文件
|--API 1 文件夹(python azure 函数)
|--API 2 文件夹(python azure 函数)
公共模块是一个子模块(位于其自己的 git 存储库中),并且拥有自己的一组多个 python 类和代码以及单元测试。
在我的 commonmodules 文件夹中,我有多个 python 类和一些数据库代码,包括一个基本存储库类,用于通过 SqlAlchemy 连接到 Azure DB,如下所示 -
**baserepository.py - **
from datetime import datetime
import os
import logging
import urllib
import utilities.utilities as utilities
from sqlalchemy import create_engine, text
from sqlalchemy.engine import URL
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
Base = declarative_base()
SQLConnectionString = os.environ.get("AzureSQLConnectionString")
SQLConnectionString = urllib.parse.quote_plus(SQLConnectionString)
SQLConnectionString = "mssql+pyodbc:///?odbc_connect=%s" % SQLConnectionString
engine = create_engine(SQLConnectionString)
# Create a session
Session = sessionmaker(bind=engine)
session = Session()
.... some common DB functions
存储库类 - 导入系统 导入操作系统 sys.path.append(os.path.dirname(os.path.dirname(file))) 从 sqlalchemy.ext.declarative 导入 declarative_base 从 data_models.get_site_overview 导入 GetSiteOverview 来自 data_repositories.baserepository 导入 get_database_results、combine_database_results、会话
# Get database session from baserepository
database_session = session
def get_site_overview(siteId: str):
.......
常见模块中的一些其他类,例如这些 -
dashboardutilities 文件夹 - dashboard_utilities.py
import json
import sys
import os
sys.path.append(os.path.dirname(os.path.dirname(__file__)))
import logging
from urllib.parse import quote_plus
import azure.functions as func
import data_repositories.base_repository as dbrepo
import unit_conversion.unit_converter as unitconvert
from typing import Tuple
from sqlalchemy import create_engine, text
def combine_database_results(headers: list, data: list) -> dict:
return {
....
} if data else {}
API1 - 使用 Python V2 的 Azure 函数
function_app.py-
import logging
import os
import sys
import pref
import azure.functions as func
sys.path.append(os.path.dirname(os.path.dirname(__file__)))
import commonmodule.utilities as utilities
app = func.FunctionApp()
@app.function_name(name="GetSettings")
@app.route(route="site/{siteId}/Settings", methods=['GET'], auth_level=func.AuthLevel.FUNCTION)
def GetSiteSettings(req: func.HttpRequest) -> func.HttpResponse:
site_id = req.route_params.get('siteId')
logging.info("Validating input parameters")
try:
response = pref.get_site_preferences(site_id)
except Exception as error:
err = "Fetching site settings failed for %s, error = %s", site_id, error
return func.HttpResponse(err, mimetype = "application/json", status_code = 400)
logging.info("Fetching settings completed for %s", site_id)
return func.HttpResponse(response["body"], mimetype = "application/json", status_code = response["statusCode"])
...像这样的多个端点
API2 的类似端点
公共模块中的测试用例是类似的,它们是那些没有发现一些测试的测试用例,只有那些存在 urllib.parse.quote_plus(SQLConnectionString) 或通过导入引用的测试用例。
import json
import allure
import sys
import pytest
import repositories.site_repository as siterepo
from math import floor
from datetime import datetime
from unittest import mock
from unittest.mock import patch, Mock, ANY
from fixtures.dashboard_utilities import test_queryparam_data, mock_request
from functools import partial
sys.path.append('src/')
import dashboard_utilities.dashboard_utilities as app;
@allure.feature('TestDashboardUtilities')
class TestDashboardUtilities():
mock_partial = partial(mock.patch.object, app)
@allure.epic('v0.1')
@pytest.mark.parametrize('from_metric, to_metric', [
('Numeric', 'mm'),
(None, 'mm'),
('mm', 'mm'),
])
def test_convert_and_round_no_convert(self, from_metric, to_metric):
value = 1.25
with patch('unit_convert.unit_converter.convert_unit') as mock_convert_unit:
results = app.convert_and_round(
from_metric,
to_metric,
value,
dp=2,
)
assert results == value
mock_convert_unit.assert_not_called()
我使用 Pytest 框架为 commonmodule 编写了单元测试,但是当 SQLConnectionString = urllib.parse.quote_plus(SQLConnectionString) 行被注释时,测试没有被发现。仅当我取消注释/删除此行时,才会发现“测试”选项卡中的“测试”。
commonmodule 的入口点仅通过 API1 或 API2,因此本地设置存在于这些文件夹中,并且环境变量仅通过它们传递。
任何人都可以提供替代方案或建议为什么它会破坏代码并且没有发现测试。
我不太确定它的原因,但是这个代码更改修复了我的测试用例并且代码运行得很好。
在 BaseRepository.py 中,以下行破坏了测试发现 -
SQLConnectionString = urllib.parse.quote_plus(SQLConnectionString)
所以 urllib 出了问题,我不知道为什么会这样。所以我将代码更改为这个 -
from sqlalchemy import create_engine, text, URL
from sqlalchemy.engine import URL
SQLConnectionString = os.environ.get("AzureSQLConnectionString")
connection_string = SQLConnectionString
connection_url = URL.create("mssql+pyodbc", query={"odbc_connect": str(connection_string)})
engine = create_engine(connection_url)