我正在尝试使用 Pytest 对处理 csv 文件的函数进行单元测试。虽然我的函数有效，但我觉得在项目目录中创建“示例”csv 文件来测试该函数时有很多代码重复。保存真实数据的实际 csv 文件有数百万条记录。
这些不是我必须在模块中测试的唯一csv 文件，因此了解测试使用不同文件结构的函数的最佳方法将非常有帮助。
现在，我正在创建一个非常短的 csv 文件，该文件通过单行数据以及通过函数处理文件后的预期数据帧输出来模拟实际的文件架构。
也许嘲笑是正确的选择？但我觉得你不应该需要嘲笑这种测试
@pytest.mark.parametrize('test_file, expected', [
(r'Path\To\Project\Output\Folder\mock_sales1.csv',
pd.DataFrame([['A0A0A0', 1, 4000]], columns=['Postal_Code', 'Store_Num', 'Sales'])),
(r'Path\To\Project\Output\Folder\mock_sales2.csv',
pd.DataFrame([['A0A0A0', 1, 4000]], columns=['Postal_Code', 'Store_Num', 'Sales']))
])
def test_sales_dataframe(test_file, expected):
# This part is repetitive, different tests each need a seperate file written within the test function.
# Writing sample file to test that files with 7 columns are read correctly.
mock_mks_sales1 = [['Data0', 'A0A0A0', 1, 'Data3', 'Data4', 'Data5', 4000]]
with open(r'Path\To\Project\Output\Folder\mock_sales1.csv', …我想设置一个模拟数据库（如果可能的话，而不是创建一个测试数据库）来检查数据是否被正确查询而不是被转换为 Pandas 数据帧。我有一些模拟和单元测试的经验，并且成功地设置了以前的测试。但是，我在应用如何模拟现实生活中的对象（如数据库）进行测试时遇到了困难。
目前，我在运行测试时无法生成结果。我相信我没有正确地模拟数据库对象，我错过了一个涉及的步骤或者我的思考过程不正确。我把我的测试和我的代码放在同一个脚本中进行测试以简化事情。
import pandas as pd
import pyodbc
import unittest
import pandas.util.testing as tm
from unittest import mock
# Function that I want to test
def p2ctt_data_frame():
conn = pyodbc.connect(
r'Driver={Microsoft Access Driver (*.mdb, *.accdb)};'
r'DBQ=My\Path\To\Actual\Database\Access Database.accdb;'
)
query = 'select * from P2CTT_2016_Plus0HHs'
# I want to make sure this dataframe object is created as intended
df = pd.read_sql(query, conn)
return df
class TestMockDatabase(unittest.TestCase):
@mock.patch('directory1.script1.pyodbc.connect') # …我正在尝试测试我的函数，该函数接受 3 个参数并且还包含另一个我试图模拟的函数。问题是模拟不会通过副作用循环；我无法将要测试的第一组值（第一个参数化参数）与副作用中的第一个值“匹配”；第二组参数化参数与副作用中的第二个值，依此类推。
我尝试使用灯具并尝试使用参数化，但无济于事。
# I go through each set of values I want to test with a different expected output
@pytest.mark.parametrize('banner, year, month, expected', [
('ctr', 2019, 1, r'A\path\file.csv'),
('mks', 2019, 1, r'B\path\file.csv'),
('sc', 2019, 1, r'C\path\file.csv'),
('atm', 2019, 1, r'D\path\file.csv')
])
@mock.patch('import.path.to.function.path_formatting')
def test_files(mock_directory_path_banner, banner, year, month, expected):
# the mocked function always returns 'ctr'
mock_directory_path_banner.side_effect = ['ctr', 'mww', 'fgl', 'fgl']
result = get_loyalty_sales_files(banner, year, month)
assert expected == result
总共有四个测试。第一次通过，因为参数化测试使用来自模拟 (ctr) 的正确值。然而，其余的测试都失败了，因为没有为每个参数化测试调用一个新的模拟。我希望每次循环通过一组新的参数化值时都会调用一个新的模拟，但事实并非如此。
一旦测试数据通过测试函数运行，我正在尝试对我的预期数据帧输出与结果数据帧的相等性进行非常简单的断言。但是，我没有比较它们是否相同，而是没有。
我的任何其他使用数据帧的测试都没有这个问题。我也可以打印出预期和结果的数据框，所以我知道它们都存在。最后，我什至可以通过该函数运行完整的生产数据，并按预期输出。
我想我错过了一些东西，但我不确定是什么。除了玩它之外，我还没有真正尝试过任何其他东西。我认为这与数据帧的结构有关。
def get_valid_postal_code(postal_code):
valid_pc = len(postal_code) == 6 and postal_code[0:5:2].isalpha() and postal_code[1:6:2].isnumeric()
return valid_pc
def clean_postal_codes(sales):
sales = sales.dropna(subset=['Postal_Code']).copy()
sales['Postal_Code'] = sales['Postal_Code'].str.replace(" ", "")
sales = sales[sales['Postal_Code'].apply(get_valid_postal_code)]
sales['Postal_Code'] = [x.upper() for x in sales['Postal_Code']]
return sales
def test_clean_postal_codes():
test_data = pd.DataFrame(
[[344, 'a0a 0a0', 3000], [344, 'a0a 0a0', -2000], [484, 'A0A 0a0', 1000], [494, 'A0a 0A0', 1000],
[700, 'A0A0A0', 1000], [900, 'A0A0A0', 1000], [50, 'Not null', 2352], [600, 'asdffa', 23523],
[634, '3HN3H3', 8839], [32, '3523', 238], …我正在尝试在一次加载时使用 Pandas 处理一个不适合内存的相当大的数据集，因此我使用的是 Dask。但是，在使用 read_csv 方法时，一旦读取，我就很难将唯一 ID 列添加到数据集。我不断收到错误消息（请参阅代码）。我正在尝试创建一个索引列，以便我可以将该新列设置为数据的索引，但错误似乎告诉我在创建列之前先设置索引。
df = dd.read_csv(r'path\to\file\file.csv') # File does not have a unique ID column, so I have to create one.
df['index_col'] = dd.from_array(np.arange(len(pc_df))) # Trying to add an index column and fill it
# ValueError: Not all divisions are known, can't align partitions. Please use `set_index` to set the index.
使用range(1, len(df) + 1将错误更改为：TypeError: Column assignment does not support type range