過往,我們在寫程式中,常常會去呼叫到參考的函式庫,或是請求外部的服務
這導致在單元測試中,我們會因為不具備其環境或是資源,而陷入難以測試的場景
若以整個系統來看,模組之間也可以說是相對的是一個”單元”吧,最常相依的就是資料!
這種情況,通常我們有的時間的話,可以大費周章的建立數據庫、測試檔案與環境,測試前後準備資料,刪除資料流程。
不過這往往是整測時才會遇到的場景,這邊先不討論。
我希望的單元測試是集中著重在自已這端的程式運作是否符合預期!!
管他外部環境怎樣,網路有沒有通、世界是否毀滅都與我暫時無關
mock的技巧就是來拯救我們上述說到的難題與困境,mock是用來模擬相依的外部服務或物件的物件。
也就是我們透過mock來將所有程式外部的相依性解除吧!!這麼說很玄
在python的unittest的函式庫中已有包含
這邊直接演示(記錄)程式使用的方式,這個例子有點簡單,但是幾乎完全相依!!
from pywebhdfs.webhdfs import PyWebHdfsClient
class WebHDFSHelper:
__hdfs_client = None
def __init__(self, host='', port='', user=''):
self.__hdfs_client = PyWebHdfsClient(host=host, port=port,
user_name=user)
def create_dir(self, full_dir_path):
try:
#before - do something
result = self.__hdfs_client.make_dir(full_dir_path)
#after - do something
except BaseException:
#exception - do something
raise
return result
def test_create_dir_temp(self):
helper = WebHDFSHelper(host="192.168.65.130", port=50070, user="root")
result = helper.create_dir( "exchange_test" )
self.assertEqual(result, True)
pass
當我們要測試建立目錄這個行為時,我們往往會遇到困難就是我必須去真實連到一個hdfs的環境!!
通常可能開發者本機都會有,或者確保團隊有一些共同的測試環境
但這種相依性會導致,若程式移轉到其他人手上或是環境一變動
測試就會無法運作!
例如:ip變了!
這次我們來透過mock的技巧來演示如何將如何模擬這些hdfs服務
@mock.patch('pywebhdfs.webhdfs.PyWebHdfsClient.__init__')
@mock.patch('pywebhdfs.webhdfs.PyWebHdfsClient.make_dir')
def test_create_dir(self, mock_hdfs_client ,mock_init_service):
mock_init_service.return_value = None
helper = WebHDFSHelper(host="localhost", port=50070, user="root")
mock_init_service.assert_called_with(host="localhost", port=50070, user_name="root")
helper.create_dir( DATA_TEMP_PATH )
mock_hdfs_client.assert_called_with(DATA_TEMP_PATH)
pass
首先,我們要先模擬的就是
第一塊WebHDFSHelper下面的初始函式會去依照參數連線到外部的hdfs服務
self.__hdfs_client = PyWebHdfsClient(host=host, port=port,user_name=user)
第二塊是呼叫create_dir的時候的這一行會去呼叫參考物件的make_dir的方法,這個會相依上面那塊初始化後的物件。
result = self.__hdfs_client.make_dir(full_dir_path)
所以我們知道這兩塊是我們能獨立單元測試的排除目標
因此我們對應的宣告
@mock.patch(‘pywebhdfs.webhdfs.PyWebHdfsClient.__init__’)
@mock.patch(‘pywebhdfs.webhdfs.PyWebHdfsClient.make_dir’)
並對我們的test_測試函式,注入2組參數
def test_create_dir(self, mock_hdfs_client ,mock_init_service):
前者是我們到時要呼叫make_dir的方法,後者是我們初始化時要mock的行為。
註:這邊mock的patch宣告所對test函式注入的順序為何是倒序,我則不知道
接著依我們的場景,mock物件可以做幾件事
1.指定回傳物件,因為是模擬,就可以一併模擬回傳的結果
mock_init_service.return_value = None #若要模擬初始函式,就是回傳值一定是none
2.透過mock物件來確認是否有被呼叫到,連同參數一併帶入,注意參數名稱必須符合原始模擬物件的參數名稱)
mock_init_service.assert_called_with(host=”localhost”, port=50070, user_name=”root”)
因此測試如下:
class HDFSHelperTest(unittest.TestCase):
@mock.patch('pywebhdfs.webhdfs.PyWebHdfsClient.__init__')
@mock.patch('pywebhdfs.webhdfs.PyWebHdfsClient.make_dir')
def test_create_dir(self, mock_hdfs_client ,mock_init_service):
mock_init_service.return_value = None
helper = WebHDFSHelper(host="localhost", port=50070, user="root")
mock_init_service.assert_called_with(host="localhost", port=50070, user_name="root")
helper.create_dir( DATA_TEMP_PATH )
mock_hdfs_client.assert_called_with(DATA_TEMP_PATH)
我們設個DEBUG中斷點看看,在初始化後,我們的內部變數已經被轉換成一個含有magicmock行為make_dir的物件。
因為初始化這邊沒有其他需要驗證的流程,因此我只去assert是否有照著我給的參數呼叫,這時管他這個環境是不是活著,我只需要知道”若”我給活的環境 就會正常。 若需要驗證例外行為,以這個例子,可以加入更多的例外處理。 這時一樣可以透過mock的return_value raise exception,這是做的到的,然而驗證的東西還是要自行定義,在一般函式環境的話,你可以因為例外給定預設行為,這樣也是有助於提高程式測試的覆蓋率。 例如:
mock_init_service.return_value = None
mock_init_service.side_effect = Exception("環境異常")
helper = WebHDFSHelper( host="localhost", port=50070, user="root" )
mock_init_service.assert_called_with( host="localhost", port=50070, user_name="root" )
如願拋出異常 最後,我們實際測試create_dir吧,第1段,我將他設定為mock_hdfs_client.return_value = True,第2段設定為False
確實會如願的走向1次成功、1次失敗,當然我這邊程式只為了驗證是否如配置去跑,所以只有這樣寫,一般成功、失敗處理上的邏輯差別還是會依原本物件設計的流程來運作 因此要「驗證」的東西與結果,仍然可以依自己的需求去調整與配置。 以上就是python的mock初體驗,以自己的例子還是比較清楚mock技巧的使用場景與方式…有錯的部分請不吝指教,感謝
測試程式完整版
from pywebhdfs.webhdfs import PyWebHdfsClient
class WebHDFSHelper:
def __init__(self, data_path='', host='', port='', user=''):
try:
self.__hdfs_client = PyWebHdfsClient(host=host, port=port,
user_name=user)
except Exception as err:
print(str(err))
def create_dir(self, full_dir_path):
try:
result = self.__hdfs_client.make_dir(full_dir_path)
if result:
print("成功")
else:
print("失敗")
except:
raise
return result
class HDFSHelperTest(unittest.TestCase):
@mock.patch('pywebhdfs.webhdfs.PyWebHdfsClient.__init__')
@mock.patch('pywebhdfs.webhdfs.PyWebHdfsClient.make_dir')
def test_create_dir(self, mock_hdfs_client, mock_init_service):
mock_init_service.return_value = None
mock_init_service.side_effect = Exception("環境異常")
helper = WebHDFSHelper(host="localhost", port=50070, user="root")
mock_init_service.assert_called_with(
host="localhost", port=50070, user_name="root")
mock_init_service.return_value = None
mock_init_service.side_effect = None
helper = WebHDFSHelper(host="localhost", port=50070, user="root")
mock_init_service.assert_called_with(
host="localhost", port=50070, user_name="root")
mock_hdfs_client.return_value = True
helper.create_dir(DATA_TEMP_PATH)
mock_hdfs_client.assert_called_with(DATA_TEMP_PATH)
mock_hdfs_client.return_value = False
helper.create_dir(DATA_TEMP_PATH)
mock_hdfs_client.assert_called_with(DATA_TEMP_PATH)
pass