Unittest Python – My First Mock Lesson！！How to Mock 3rd party Component!!

By paul | 2017-06-27 | Comments 0 Comment

過往，我們在寫程式中，常常會去呼叫到參考的函式庫，或是請求外部的服務

這導致在單元測試中，我們會因為不具備其環境或是資源，而陷入難以測試的場景

若以整個系統來看，模組之間也可以說是相對的是一個”單元”吧，最常相依的就是資料！

這種情況，通常我們有的時間的話，可以大費周章的建立數據庫、測試檔案與環境，測試前後準備資料，刪除資料流程。

不過這往往是整測時才會遇到的場景，這邊先不討論。

我希望的單元測試是集中著重在自已這端的程式運作是否符合預期！！

管他外部環境怎樣，網路有沒有通、世界是否毀滅都與我暫時無關

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

測試程式完整版

發佈留言 取消回覆

發佈留言取消回覆