談談項目的重構與測試

時間 2019-12-06

標籤談談項目測試简体版

原文原文鏈接

這篇文章摘自個人博客, 歡迎你們沒事去逛逛~python

背景

這幾個月我開發了公司裏的一個restful webservice，起初技術選型的時候是採用了flask框架。雖然flask是一個同步的框架，可是能夠配合gevent或者其它方式運行在異步的容器中(測試連接)，效果看上去也還能夠，所以就採用了這種方式。git

後面閱讀了tornado的源碼，也去了解了各類協程框架以及運行的原理。總感受flask的這種同步方式編程不夠好，同時對於這種運行在容器裏的模式目前還缺少了解。但至少如今對於tornado的運行原理有了必定的瞭解，若是用tornado寫的話，是很可控的，並且能夠保證運行是高效的。所以就決定把原來基於flask的項目用tornado重構了。github

重構的過程

項目重構的過程當中遇到了一些問題，也學習了一些東西，這裏作一個簡單的總結。web

接入層

全部框架都要處理的一個接入層的事情就是：mongodb

url-mapping數據庫
項目初始化編程
參數解析json

對於restful風格的接口以及項目的初始化，每一個框架都有本身的方式，在它們的文檔中都演示得特別清楚，因此關於這些我就不展開了。flask

關於參數解析，這裏並非指簡單地調用相似於get_argument這樣的方法去獲取數據。而是 如何從不可靠的client端傳來的數據中過濾掉服務器不關注的數據，同時對服務器關注的數據做一些更強的校驗，這就是協議層的事情了。api

使用谷歌的ProtocolBuffer是一個不錯的方案，它有很不錯的數據壓縮率，也支持目前大多數主流的開發語言。但在一些小型項目中，我仍是更偏向於使用json的方式，它顯得更加靈活。可是對於json的話，如何做數據校驗就是另一個問題了。

在重構前，我是經過python中的裝飾器來實現的這個功能:

class SomeHandlerInFlask(Resource):
    @util.deco({
        'key_x': (str, 'form'),
        'key_y': (int, 'form'),
        'key_z': (str, 'url')
    })
    def post(self):
        # logic code
        pass

在裝飾器中分別從不一樣的地方，form或者url中獲取相應的參數。若是取不到，則直接報錯，邏輯也不會進入到post函數中。

這是我基於flask這個框架本身總結出來的一套尚且還能看能用的參數解析方式，若是在每一個函數中經過框架提供的get_argument來逐一獲取參數，則顯得太醜，並且每一個接口所須要的數據是什麼也不夠直觀。不過這種方式我本身還不是特別滿意，總感受仍是有點不太舒服，也說不清不舒服在哪裏。那就乾脆放棄它，使用別的方式吧。

後來我瞭解到了jsonschema這個東西，看了一下感受與ProtocolBuffer很類似，只不過它是採用json的格式定義，正合我意(對於它我也有點吐槽，在數據庫層有提到)，每次數據進來就對數據和schema做一次validate操做，再進入業務邏輯層。

業務邏輯層

業務邏輯層的重構其實改動的代碼並很少，把一些同步的操做改爲異步的操做。就拿如何重構某個接口來講吧，重構前的代碼多是這樣的:

def function_before_refactor(some_params):
    result_1 = sync_call_1(some_params)
    result_2 = sync_call_2(some_params)
    # some other processes
    return result

使用gen.coroutine重構後:

from tornado import gen

@gen.coroutine
def function_after_refactor(some_params):
    # if you don't want to refactor
    # just call it as it always be
    result_1 = sync_call_1(some_params)
    result_2 = yield async_call_2(some_params)
    # some other processes
    raise gen.Return(result)
    # python3及以上的版本不須要採用拋出異常的方式，直接return就能夠了
    # return result

考慮到函數名根本不用改，重構的過程很是容易：

函數用gen.coroutine包裝成協程
已經重構成異步方式的函數調用時添加yield關鍵字便可
函數返回採用raise gen.Return(result)的方式（僅限於Python 2.7）

由於我目前採用的是python 2.7，因此在處理返回的時候要用拋出異常的方式，在這種方式下有一個點須要注意到，那就是與日常異常的處理的混用，否則會致使邏輯流執行混亂：

from tornado import gen

@gen.coroutine
def function_after_refactor(some_params):
    try:
        # some logic code
        pass
    except Exception as e:
        if isinstance(e, gen.Return):
            # return the value raised by logic
            raise gen.Return(e.value)
        # more exception process

數據庫層

數據庫採用的是mongodb，在flask框架中採用了mongoengine做爲數據庫層的orm，對於這個python-mongodb的orm產品，我我的並非很喜歡(多是由於我習慣了mongoose的工做方式)，這裏面嵌套json的定義竟然不能體如今schema中，須要分開定義兩個schema，而後再做引入的操做。好比（代碼只是用做演示，與項目無關）：

class Comment(EmbeddedDocument):
    content = StringField()
    # more comment details

class Page(Document):
    comments = ListField(EmbeddedDocumentField(Comment))
    # more page details

而在mongoose中就直觀多了:

var PageSchema = new Schema({
    title       :   {type : String, required : true},
    time        :   {type : Date, default : Date.now(), required : true},
    comments    :   [{
        content   :   {type : String}
        // more comment details
    }]
    // more page details
});

扯遠了，在tornado的框架中，再使用mongoengine就不合適了，畢竟有着異步和同步的區別。那有什麼比較好的python-mongodb的異步orm框架呢？搜了下，有一個叫作motorengine的東西，orm的使用方式和mongoengine基本同樣，但看它的star數實在不敢用呀。並且它處理異步的方式是使用回調，如今都是使用協程的年代了，想一想仍是算了吧。

最後找了個motor，感受還不錯，它有對目前大部分主流協程框架的支持，操做mongodb的方式與直接使用pymongo的方式差很少（畢竟都是基於pymongo的封裝嘛），可是就是沒有orm的驗證層，那就本身再去另外搞一個簡化的orm層吧。(mongokit的orm方式看上去還不錯，但貌似對協程框架的支持通常)。這裏暫時先懶惰一下，仍是採用了jsonschema。每次保存前都validate一下對象是否符合schema的定義。若是沒有類mongoose的python-mongodb異步框架，有時間就本身寫一個吧~

這裏順帶吐槽一下jsonschema，簡直太瑣碎了，一個很短的文檔結構定義，它會描述成好幾十行，我就不貼代碼了，有興趣的朋友能夠戳這裏http://jsonschema.net/玩玩。並且python中的jsonschema庫還不支持對於default關鍵字的操做，參見這個issue。

測試

本身摸索的一種接口測試方案

python中的測試框架有不少，只要選擇一個合適的可以很方便與項目集成就好。我我的仍是很喜歡unittest這個框架，小而精。個人這套測試方案也是基於unittest框架的。

# TestUserPostAccessComponents.py
class TestUserPostAccessComponents(unittest.TestCase):
    @classmethod
    def setUpClass(cls):
        # 定義在其它地方，具體細節就不展現了
        # 在setup中使用測試帳號獲取登錄態
        # 並把各類中間用獲得的信息放在TestUserPostAccess類上
        setup(cls)

    @classmethod
    def tearDownClass(cls):
        pass

    def setUp(self):
        pass

    def tearDown(self):
        pass

    def test_1_user_1_user_2_add_friend(self):
        pass

    def test_2_user_1_user_2_del_friend(self):
        pass

    def test_3_user_1_add_public_user_post(self):
        pass

    # more other components

最頂層的測試文件：

# run_test.py
# 各類import

def user_basic_post_access_test():
    tests = ['test_3_user_1_add_public_user_post',
             'test_5_user_2_as_a_stranger_can_access_public_user_post',
             'test_4_user_1_del_public_user_post',
             'test_6_user_1_add_private_user_post',
             'test_8_user_2_as_a_stranger_can_not_access_private_user_post',
             'test_9_user_1_self_can_access_private_user_post',
             'test_7_user_1_del_private_user_post']
    return unittest.TestSuite(map(TestUserPostAccessComponents, tests))

def other_process_test():
    tests = [
        # compose a process by components by yourself
    ]
    return unittest.TestSuite(map(OtherTestCaseComponents, tests))

runner = unittest.TextTestRunner(verbosity=2)
runner.run(user_basic_post_access_test())
runner.run(other_process_test())

這套測試是基於 BDD (行爲驅動)的測試方式，針對每個邏輯模塊，定義一個components類，把全部子操做都定義成單獨的測試單元。這裏面的測試單元能夠是徹底無序的，把邏輯有序化組織成測試用例的過程會在最外面經過TestSuit的方式組織起來。這裏可能會有一些異議，由於有些人在使用這個測試類的時候是把它做爲一個測試用例來組織的，固然這些都是不一樣的使用方式。

這套測試方案中的每一個component都是api級別的測試，並非函數級別的測試（集成測試與單元測試），每一個TestSuit都是完整的一個業務流程。這樣的好處在於 測試和項目徹底解耦。測試代碼不用關心項目的代碼是同步仍是異步的。就算項目重構了，測試徹底無感知，只要api沒變，就能夠繼續工做。

固然以上都是理想的狀態，由於在剛開始寫這些測試的時候我尚未總結到這些點，致使了一些耦合性的存在。好比說測試代碼中import了項目中的某個函數去獲取一些數據，用於檢查某個component的更新操做是否成功。在重構的過程當中，該函數被重構成了協程。這樣一來，在測試代碼中就不能採用原來同樣的方式去調用了，也就是說測試代碼受到了框架同步與異步的影響，下一節咱們就來談談同步與異步的測試，以及對於這種問題的解決方案。

異步測試&同步測試

在tornado中，也提供了一套測試的功能，具體在tornado.testing這個模塊，看它源碼其實能夠發現它也是基於unittest的一層封裝。
我內心一直有一個問題：unittest的執行流程是同步的，既然這樣，它是怎麼去測一個由gen.coroutine包裝的協程的呢，畢竟後者是異步的。
直到看了源碼，恍然大悟，原來是io_loop.run_sync這個函數的功勞，具體實如今gen_test這個裝飾器中，摘一部分源碼（對於tornado源碼不熟的同窗能夠先去看看tornado中的ioloop模塊的實現，看完會對這個部分有更深入的理解）：

def gen_test(func=None, timeout=None):
    if timeout is None:
        timeout = get_async_test_timeout()

    def wrap(f):
        # Stack up several decorators to allow us to access the generator
        # object itself.  In the innermost wrapper, we capture the generator
        # and save it in an attribute of self.  Next, we run the wrapped
        # function through @gen.coroutine.  Finally, the coroutine is
        # wrapped again to make it synchronous with run_sync.
        #
        # This is a good case study arguing for either some sort of
        # extensibility in the gen decorators or cancellation support.
        @functools.wraps(f)
        def pre_coroutine(self, *args, **kwargs):
            result = f(self, *args, **kwargs)
            if isinstance(result, types.GeneratorType):
                self._test_generator = result
            else:
                self._test_generator = None
            return result

        coro = gen.coroutine(pre_coroutine)

        @functools.wraps(coro)
        def post_coroutine(self, *args, **kwargs):
            try:
                return self.io_loop.run_sync(
                    functools.partial(coro, self, *args, **kwargs),
                    timeout=timeout)
            except TimeoutError as e:
                # run_sync raises an error with an unhelpful traceback.
                # If we throw it back into the generator the stack trace
                # will be replaced by the point where the test is stopped.
                self._test_generator.throw(e)
                # In case the test contains an overly broad except clause,
                # we may get back here.  In this case re-raise the original
                # exception, which is better than nothing.
                raise
        return post_coroutine

    if func is not None:
        # Used like:
        #     @gen_test
        #     def f(self):
        #         pass
        return wrap(func)
    else:
        # Used like @gen_test(timeout=10)
        return wrap

在源碼中，先把某個測試單元封裝成一個協程，而後獲取當前線程的ioloop對象，把協程拋給他去執行，直到執行完畢。這樣就完美地實現了異步到同步的過渡，知足unittest測試框架的同步需求。
在具體的使用中只須要繼承tornado提供的AsyncTestCase類就好了，注意這裏不是unittest.TestCase。看了源碼也能夠發現，前者就是繼承自後者的。

# This test uses coroutine style.
class MyTestCase(AsyncTestCase):
    @tornado.testing.gen_test
    def test_http_fetch(self):
        client = AsyncHTTPClient(self.io_loop)
        response = yield client.fetch("http://www.tornadoweb.org")
        # Test contents of response
        self.assertIn("FriendFeed", response.body)

回到上一節的問題，有了這種方式，就能夠很容易地解決同步異步的問題了。若是測試用例中某一個函數已經被項目重構成了協程，只須要作如下三步：