为哪些分析代码编写单元测试？
What parts of analytical code to write unit tests for?
最近我一直在编写分析 Python 代码，当用户通过基于队列的批处理与前端工具交互时，这些代码会按需获得 运行。
通常，用户在前端工具中设置一些值，这些值作为参数传递给分析代码，他们要么提供数据集，要么从他们公司提供的整体数据源中选择数据子集。
通常每个分析模型都位于其他分析模型中的一个更大的 repo 中，因此每个模型通常都位于它自己的模块中，并且该模块将导出一个函数，该函数是该模型的入口点。这些模型的范围从需要几分钟的简单模型到非常复杂的基于统计或机器学习的模型，并且可能使用 numpy/Pandas/Numba 或 Dask 数据帧的组合，需要几个小时。
现在回答我的问题，我一直在反复讨论应该将我的测试工作集中在此类代码上的哪些方面。在我职业生涯的早期，我天真地认为每个函数都应该有一个单元测试，这样我的代码就会有一组全面的测试。
我很快意识到这会适得其反，因为即使是很小的性能重构也可能导致分崩离析，甚至可能丢掉很多单元测试。很明显，感觉我应该只为每个模型的主要 public 函数编写测试，然而，这通常意味着相反的情况，对于一些更复杂的模型，控制非常深入的边缘情况流量现在很难测试。
那么我的问题是我应该如何正确测试这些分析模型？有些人可能会说 "Only test public facing functions, if you can't test edge cases through the public facing functions then they should technically not be reachable so don't need to be there"。但是，我发现，实际上这并不奏效。
举个简单的例子，假设特定模型是计算出租车数据集中 dropoff/pickoff 个点的频率矩阵。
import pandas as pd
def _cat(col1, col2):
cat_col = col1.astype(str).str.cat(col2.astype(str), ', ')
return cat_col
def _make_points_df(taxi_df):
pickup_points = _cat(taxi_df["pickup_longitude"], taxi_df["pickup_latitude"])
dropoff_points = _cat(taxi_df["dropoff_longitude"], taxi_df["dropoff_latitude"])
points_df = pd.DataFrame({"pickup": pickup_points, "dropoff": dropoff_points})
return points_df
def _points_df_to_freq_mat(points_df):
mat_df = points_df.groupby(['pickup', 'dropoff']).size().unstack(fill_value=0)
return mat_df
def _validate_taxi_df(taxi_df):
if type(taxi_df) is not pd.DataFrame:
raise TypeError(f"taxi_df param must be a pandas dataframe, got: {type(taxi_df)}")
expected_cols = {
"pickup_longitude",
"pickup_latitude",
"dropoff_longitude",
"dropoff_latitude",
}
if set(taxi_df) != expected_cols:
raise RuntimeError(
f"Expected the following columns for taxi_df param: {expected_cols}."
f"Got: {set(taxi_df)}"
)
def calculate_frequency_matrix(taxi_df, long_round=1, lat_round=1):
"""Calculate a dropoff/pickup frequency matrix which tells you the number of times
passengers have been picked up and dropped from a given discrete point. The
resolution of these points is controlled by using the long_round and lat_round params
Paramaters
----------
taxi_df : pandas.DataFrame
Dataframe specifying dropoff and pickup long/lat coordinates
long_round : int
Number of decimal places to round the dropoff and pickup longitude values to
lat_round : int
Number of decimal places to round the dropoff and pickup latitude values to
Returns
-------
pandas.DataFrame
Dataframe in matrix format of frequency of dropoff/pickup points
Raises
------
TypeError : If taxi_df is not a pandas DataFrame
RuntimeError : If taxi_df does not contain correct columns
"""
_validate_taxi_df(taxi_df)
taxi_df = taxi_df.copy()
taxi_df["pickup_longitude"] = taxi_df["pickup_longitude"].round(long_round)
taxi_df["dropoff_longitude"] = taxi_df["dropoff_longitude"].round(long_round)
taxi_df["pickup_latitude"] = taxi_df["pickup_latitude"].round(lat_round)
taxi_df["dropoff_latitude"] = taxi_df["dropoff_latitude"].round(lat_round)
points_df = _make_points_df(taxi_df)
mat_df = _points_df_to_freq_mat(points_df)
return mat_df
接受像
这样的数据框
pickup_longitude pickup_latitude dropoff_longitude dropoff_latitude
0 -73.988129 40.732029 -73.990173 40.756680
1 -73.964203 40.679993 -73.959808 40.655403
2 -73.997437 40.737583 -73.986160 40.729523
3 -73.956070 40.771900 -73.986427 40.730469
4 -73.970215 40.761475 -73.961510 40.755890
5 -73.991302 40.749798 -73.980515 40.786549
6 -73.978310 40.741550 -73.952072 40.717003
7 -74.012711 40.701527 -73.986481 40.719509
就文件夹结构而言，这段代码将位于
analytics/models/taxi_freq/taxi_freq.py
和
analytics/models/taxi_freq/__init__.py
文件看起来像
from taxi_freq import calculate_frequency_matrix
显然，上述代码中的私有函数可以拆分为 analytics/models/taxi_freq/ 中的多个实用程序文件。
共识是只测试 calculate_frequency_matrix 函数，还是应该测试 taxi_freq 模块中的 "private" 辅助方法和其他实用程序 files/functions？
与一般的软件开发一样，对于测试，您始终必须搜索代表竞争目标之间（理想情况下最佳）权衡的解决方案。一般测试和单元测试的主要目标之一是发现错误（请参阅 Myers、Badgett、Sandler：软件测试的艺术，或 Beizer：软件测试技术，还有许多其他内容）。
在你的项目中，你可能对此持更宽松的立场，但有许多软件项目如果实现级别的错误逃逸到后期开发阶段甚至现场，将产生严重后果。有人说，您的目标应该是增加对代码的信心——这也是事实，但信心只能是正确进行测试的结果。如果你不测试找bug，那你测试完了我对你的代码根本就没有信心。
当发现错误是单元测试的主要目标时，尝试使单元测试套件完全独立于实现细节可能会导致测试套件效率低下——即不适合查找错误的测试套件所有可以找到的错误。不同的实现有不同的潜在错误。如果您不使用单元测试来查找这些错误，那么任何其他测试级别（集成、子系统、系统）肯定不太适合系统地查找它们。
例如，考虑实现斐波那契函数的不同方法：作为迭代或递归函数，作为封闭形式表达式 (Moivre/Binet)，或作为查找 table：界面总是一样的，可能的错误有很大的不同，单元测试策略也是如此。将有一组有用的独立于实现的测试用例，但仅凭这些不足以找到特定实现可能存在的所有错误。
因此，拥有有效测试套件的目标与另一个目标竞争，即拥有易于维护的测试套件。然而，这个目标以不同的形式出现，结果也不同：您可以要求单元测试套件在实现细节发生变化时不受影响。这非常困难，IMO 将维护友好测试代码的次要目标置于查找错误的主要目标之上。
Meszaros 有一个更平衡的公式，即 "The effort for changes to the code base shall be commensurate with the effort to maintain the test suite."（参见 Meszaros：Principles of Test Automation: Ensure Commensurate Effort）。也就是说，对 SUT 的少量更改只需要对测试套件进行少量更改，对于对 SUT 的较大更改，它接受table 测试套件需要同样大的更改。 （不过，对我个人而言，公式 "the effort for test code maintenance shall be low" 就足够了。）
结论：
对我来说，因为我将发现错误视为主要目标，将测试套件的可维护性视为次要目标，这导致了以下结果：我承认我还必须测试实现细节以发现更多错误。但是，尽管如此，我仍然试图保持较低的维护工作量：我主要通过应用以下机制来做到这一点，这些机制旨在在 SUT 发生变化时更简单地调整测试套件：
- 首先，如果与实现无关的测试用例和与实现相关的测试用例可以达到特定测试用例的目标，则更喜欢与实现无关的测试用例。换句话说，不要使单个测试用例不必要地依赖于实现。
- 其次，隐藏辅助函数背后的实现细节。可以有用于特定设置、拆卸、断言等的辅助函数。这是一种非常强大的机制，可以限制测试套件中实现细节的影响。
最近我一直在编写分析 Python 代码，当用户通过基于队列的批处理与前端工具交互时，这些代码会按需获得 运行。
通常，用户在前端工具中设置一些值，这些值作为参数传递给分析代码，他们要么提供数据集，要么从他们公司提供的整体数据源中选择数据子集。
通常每个分析模型都位于其他分析模型中的一个更大的 repo 中，因此每个模型通常都位于它自己的模块中，并且该模块将导出一个函数，该函数是该模型的入口点。这些模型的范围从需要几分钟的简单模型到非常复杂的基于统计或机器学习的模型，并且可能使用 numpy/Pandas/Numba 或 Dask 数据帧的组合，需要几个小时。
现在回答我的问题，我一直在反复讨论应该将我的测试工作集中在此类代码上的哪些方面。在我职业生涯的早期，我天真地认为每个函数都应该有一个单元测试，这样我的代码就会有一组全面的测试。 我很快意识到这会适得其反，因为即使是很小的性能重构也可能导致分崩离析，甚至可能丢掉很多单元测试。很明显，感觉我应该只为每个模型的主要 public 函数编写测试，然而，这通常意味着相反的情况，对于一些更复杂的模型，控制非常深入的边缘情况流量现在很难测试。
那么我的问题是我应该如何正确测试这些分析模型？有些人可能会说 "Only test public facing functions, if you can't test edge cases through the public facing functions then they should technically not be reachable so don't need to be there"。但是，我发现，实际上这并不奏效。
举个简单的例子，假设特定模型是计算出租车数据集中 dropoff/pickoff 个点的频率矩阵。
import pandas as pd
def _cat(col1, col2):
cat_col = col1.astype(str).str.cat(col2.astype(str), ', ')
return cat_col
def _make_points_df(taxi_df):
pickup_points = _cat(taxi_df["pickup_longitude"], taxi_df["pickup_latitude"])
dropoff_points = _cat(taxi_df["dropoff_longitude"], taxi_df["dropoff_latitude"])
points_df = pd.DataFrame({"pickup": pickup_points, "dropoff": dropoff_points})
return points_df
def _points_df_to_freq_mat(points_df):
mat_df = points_df.groupby(['pickup', 'dropoff']).size().unstack(fill_value=0)
return mat_df
def _validate_taxi_df(taxi_df):
if type(taxi_df) is not pd.DataFrame:
raise TypeError(f"taxi_df param must be a pandas dataframe, got: {type(taxi_df)}")
expected_cols = {
"pickup_longitude",
"pickup_latitude",
"dropoff_longitude",
"dropoff_latitude",
}
if set(taxi_df) != expected_cols:
raise RuntimeError(
f"Expected the following columns for taxi_df param: {expected_cols}."
f"Got: {set(taxi_df)}"
)
def calculate_frequency_matrix(taxi_df, long_round=1, lat_round=1):
"""Calculate a dropoff/pickup frequency matrix which tells you the number of times
passengers have been picked up and dropped from a given discrete point. The
resolution of these points is controlled by using the long_round and lat_round params
Paramaters
----------
taxi_df : pandas.DataFrame
Dataframe specifying dropoff and pickup long/lat coordinates
long_round : int
Number of decimal places to round the dropoff and pickup longitude values to
lat_round : int
Number of decimal places to round the dropoff and pickup latitude values to
Returns
-------
pandas.DataFrame
Dataframe in matrix format of frequency of dropoff/pickup points
Raises
------
TypeError : If taxi_df is not a pandas DataFrame
RuntimeError : If taxi_df does not contain correct columns
"""
_validate_taxi_df(taxi_df)
taxi_df = taxi_df.copy()
taxi_df["pickup_longitude"] = taxi_df["pickup_longitude"].round(long_round)
taxi_df["dropoff_longitude"] = taxi_df["dropoff_longitude"].round(long_round)
taxi_df["pickup_latitude"] = taxi_df["pickup_latitude"].round(lat_round)
taxi_df["dropoff_latitude"] = taxi_df["dropoff_latitude"].round(lat_round)
points_df = _make_points_df(taxi_df)
mat_df = _points_df_to_freq_mat(points_df)
return mat_df
接受像
这样的数据框 pickup_longitude pickup_latitude dropoff_longitude dropoff_latitude
0 -73.988129 40.732029 -73.990173 40.756680
1 -73.964203 40.679993 -73.959808 40.655403
2 -73.997437 40.737583 -73.986160 40.729523
3 -73.956070 40.771900 -73.986427 40.730469
4 -73.970215 40.761475 -73.961510 40.755890
5 -73.991302 40.749798 -73.980515 40.786549
6 -73.978310 40.741550 -73.952072 40.717003
7 -74.012711 40.701527 -73.986481 40.719509
就文件夹结构而言，这段代码将位于
analytics/models/taxi_freq/taxi_freq.py
和
analytics/models/taxi_freq/__init__.py
文件看起来像
from taxi_freq import calculate_frequency_matrix
显然，上述代码中的私有函数可以拆分为 analytics/models/taxi_freq/ 中的多个实用程序文件。
共识是只测试 calculate_frequency_matrix 函数，还是应该测试 taxi_freq 模块中的 "private" 辅助方法和其他实用程序 files/functions？
与一般的软件开发一样，对于测试，您始终必须搜索代表竞争目标之间（理想情况下最佳）权衡的解决方案。一般测试和单元测试的主要目标之一是发现错误（请参阅 Myers、Badgett、Sandler：软件测试的艺术，或 Beizer：软件测试技术，还有许多其他内容）。
在你的项目中，你可能对此持更宽松的立场，但有许多软件项目如果实现级别的错误逃逸到后期开发阶段甚至现场，将产生严重后果。有人说，您的目标应该是增加对代码的信心——这也是事实，但信心只能是正确进行测试的结果。如果你不测试找bug，那你测试完了我对你的代码根本就没有信心。
当发现错误是单元测试的主要目标时，尝试使单元测试套件完全独立于实现细节可能会导致测试套件效率低下——即不适合查找错误的测试套件所有可以找到的错误。不同的实现有不同的潜在错误。如果您不使用单元测试来查找这些错误，那么任何其他测试级别（集成、子系统、系统）肯定不太适合系统地查找它们。
例如，考虑实现斐波那契函数的不同方法：作为迭代或递归函数，作为封闭形式表达式 (Moivre/Binet)，或作为查找 table：界面总是一样的，可能的错误有很大的不同，单元测试策略也是如此。将有一组有用的独立于实现的测试用例，但仅凭这些不足以找到特定实现可能存在的所有错误。
因此，拥有有效测试套件的目标与另一个目标竞争，即拥有易于维护的测试套件。然而，这个目标以不同的形式出现，结果也不同：您可以要求单元测试套件在实现细节发生变化时不受影响。这非常困难，IMO 将维护友好测试代码的次要目标置于查找错误的主要目标之上。
Meszaros 有一个更平衡的公式，即 "The effort for changes to the code base shall be commensurate with the effort to maintain the test suite."（参见 Meszaros：Principles of Test Automation: Ensure Commensurate Effort）。也就是说，对 SUT 的少量更改只需要对测试套件进行少量更改，对于对 SUT 的较大更改，它接受table 测试套件需要同样大的更改。 （不过，对我个人而言，公式 "the effort for test code maintenance shall be low" 就足够了。）
结论：
对我来说，因为我将发现错误视为主要目标，将测试套件的可维护性视为次要目标，这导致了以下结果：我承认我还必须测试实现细节以发现更多错误。但是，尽管如此，我仍然试图保持较低的维护工作量：我主要通过应用以下机制来做到这一点，这些机制旨在在 SUT 发生变化时更简单地调整测试套件：
- 首先，如果与实现无关的测试用例和与实现相关的测试用例可以达到特定测试用例的目标，则更喜欢与实现无关的测试用例。换句话说，不要使单个测试用例不必要地依赖于实现。
- 其次，隐藏辅助函数背后的实现细节。可以有用于特定设置、拆卸、断言等的辅助函数。这是一种非常强大的机制，可以限制测试套件中实现细节的影响。