python之如何模拟和测试 python open 和 pandas to_pickle

我正在尝试测试这个函数，该函数采用一个 Pandas 数据帧行，它用于进行保存到 csv 的 ftp 调用，打开该 csv 文件，对其进行格式化，并将其保存为泡菜。

我想测试以下内容:

builtins.open 被调用一次 (path_to_raw, 'wb')

to_pickle 被调用一次 (LOCAL_PKL.format(row.name))

修补 builtins.open 似乎不起作用，因为它被 to_pickle 间接调用，因此测试失败，因为 builtins.open 被调用两次。

测试功能:

def download_file(row): 
    path_from = row['source'] 
    path_to_raw = LOCAL_RAW.format(row.name) 
 
    self.connection = FTP(self.url) 
    self.connection.login(self.username, self.password) 
    with open(path_to_raw, 'wb') as f: 
        self.connection.retrbinary('RETR ' + path_from, f.write) 
    self.connection.quit() 
 
    data = pd.read_csv(path_to_raw) 
    data.columns = ['a','b','c'] 
    data.to_pickle(LOCAL_PKL.format(row.name)) 

单元测试:

import pandas as pd 
import unittest.mock as mock 
from unittest.mock import patch, mock_open, MagicMock, call 
import maintain 
 
@patch('builtins.open', create=True) 
@patch('maintain.pd.read_csv') 
def test_download_path(self, mock_open, mock_pd_read_csv): 
 
    mock_pd_read_csv.return_value = pd.DataFrame()       
 
    @mock.create_autospec 
    def mock_pd_to_pickle(self, path): 
        pass 
 
    with patch.object(pd.DataFrame, 'to_pickle', mock_pd_to_pickle): 
 
        real = maintain.DataFTP() 
        real.connection = MagicMock(name='connection') 
 
        row = pd.Series(data=['a','b'], index=['c','d']) 
        row.name = 'anything' 
 
        print(mock_open.assert_called_once_with(maintain.LOCAL_RAW.format(row.name), 'wb')) 
        print(mock_pd_to_pickle.assert_called_once_with(maintain.LOCAL_PKL.format(row.name))) 

所以......这显然是错误的，但我不确定为什么。
此测试产生此错误:

AssertionError: Expected 'read_csv' to be called once. Called 0 times. 

有没有人有任何建议或知道如何解决这个问题。
谢谢!

请您参考如下方法:

我终于得到了它的工作:

@patch('builtins.open', new_callable=mock_open) 
@patch('maintain.pd.read_csv', return_value=pd.DataFrame()) 
@patch.object(pd.DataFrame, 'to_pickle') 
def test_download_path(self, mock_to_pickle, mock_read_csv, mock_open): 
 
    real = maintain.EODDataFTP() 
    real.connection = mock.Mock(name='connection') 
 
    row = pd.Series(data=['','nyse'], index=['source','exchange']) 
    row.name = 'anything' 
 
    real.download_file(row) 
 
    mock_open.assert_called_once_with(maintain.LOCAL_RAW.format(row.name), 'wb') 
    mock_read_csv.assert_called_once() 
    mock_to_pickle.assert_called_once_with(maintain.LOCAL_PKL.format(row.name))