沒想到這個系列居然還有2出現(xiàn)，實在是因為近期AI這邊的接口測試量太大了，一些微小的細節(jié)上沒處理好，也會很影響效率。所以這篇內(nèi)容也在解決問題的過程中順勢誕生了。

問題：如何解決用例亂序的問題。

先說一下背景，做過GPT應用的同學應該知道，上下文對于GPT來說是非常重要的，同樣的問題搭配不同的上下文，能得到的結(jié)果是完全不同的。

這也要求測試這邊需要在測試過程中保證用例執(zhí)行的順序，才不會偏離預期的結(jié)果。

在上一篇文章時，我的處理方案是：

1. 給同一批用例使用同一個UUID作為區(qū)分，并且給某一個用例設置一個 is_first=True 的標志。
2. 在執(zhí)行過程中，輪到某個UUID時，會先去判斷是否為 is_first ，如果是就直接執(zhí)行，如果不是，就會去一個文件中檢查這個UUID的首位是否已經(jīng)執(zhí)行，如果未執(zhí)行則等到首位執(zhí)行完后繼續(xù)。

這個方案大體也能解決順序的問題，但是存在一個比較嚴重的問題就是等待時間過長。甚至在某種程度上抵消了 xdist 的并發(fā)性。

為什么？

這里涉及到 xdist 的默認用例分配方案，在 xdist 中這些方案一般稱之為 Schedule（別問，問就是源碼都是這么寫的）。默認的方案會將用例平均分配到每個 Worker 中，這樣就會導致某個首位用例，會排在 Worker 用例列表的靠后的地方，那另一個Worker 就被迫要等執(zhí)行到這個用例之后才能繼續(xù)。

雖然較之前30分鐘到一個小時的執(zhí)行時間來說，已經(jīng)快了一些了，但是依舊不盡如人意。那就改唄。

問題很明顯，就是因為同一個 UUID 的用例分散在了不同的 Worker 中執(zhí)行，那就讓同 UUID 的用例都在同一個 Worker 中執(zhí)行就行。

所以第一步要搞清楚， xdist 是如何分配用例的？

通過一番搜索后以及根據(jù)搜索內(nèi)容再對源碼進行閱讀后發(fā)現(xiàn)了以下代碼：

# xdist/scheduler/loadscope.py
    def schedule(self):
        
        ......
        
        for nodeid in self.collection:
            scope = self._split_scope(nodeid)
            work_unit = self.workqueue.setdefault(scope, default=OrderedDict())
            work_unit[nodeid] = False

解釋一下，這段代碼的作用就是把所有已經(jīng)搜集到的用例進行一次循環(huán)，然后將相同 scope 值的用例分發(fā)到同一個 Worker 中。

scope 的獲取就在 self._split_scope(nodeid) 之中，而我們要做的就是重寫一下這個方法。

但是在重寫之前我們還要搞清楚如何讓我們自己寫的 Schedule 生效？

又是一番搜索，最后從 GPT 中獲得了如下答案。

class HashScheduler(LoadFileScheduling):
    def _get_key(self, nodeid):
        return hash(nodeid)

    def compare(self, left, right):
        return left[1] < right[1]


def pytest_xdist_make_scheduler(config, log):
    return HashScheduler(config)

簡單調(diào)試了一下之后發(fā)現(xiàn) pytest_xdist_make_scheduler 是正常生效的，同時日志也輸出了如下內(nèi)容

4 workers [200 items]   
scheduling tests via HashScheduler

說明，GPT給的答案是有效的，于是在將 _split_scope 方法重寫之后，發(fā)現(xiàn)日志順利的打印上了。

_split_scope runner/test_single_intent.py::TestCorpus::test_intent[single_intent_corpus0]
_split_scope runner/test_single_intent.py::TestCorpus::test_intent[single_intent_corpus1]
_split_scope runner/test_single_intent.py::TestCorpus::test_intent[single_intent_corpus2]

后續(xù)再根據(jù)這個名稱去獲取到我需要的 UUID 后，將UUID 返回即可。

最終代碼如下

class UuidScheduler(LoadFileScheduling):
    def mark_test_pending(self, item):
        pass

    def _split_scope(self, nodeid: str) -> str:
        """
        返回用例分類依據(jù)
        相同的依據(jù)會使用同一個worker執(zhí)行
        """
        qcg = QuestionCloselyCorpusGenerator()
        with open(qcg.save_path, 'r') as f:
            case_list = json.loads(f.read().encode('utf-8'))
        this_id = int(nodeid.split('corpus')[1][:-1])
        current_case = case_list[this_id]
        return current_case['uuid']


def pytest_xdist_make_scheduler(config, log):
    return UuidScheduler(config)

至此，就解決了將相同批次的語料放到同一個 Worker 中執(zhí)行的問題。