pytest-xdist一些不为人知的细节2——如何加载用例？

没想到这个系列居然还有2出现，实在是因为近期AI这边的接口测试量太大了，一些微小的细节上没处理好，也会很影响效率。所以这篇内容也在解决问题的过程中顺势诞生了。

问题：如何解决用例乱序的问题。

先说一下背景，做过GPT应用的同学应该知道，上下文对于GPT来说是非常重要的，同样的问题搭配不同的上下文，能得到的结果是完全不同的。

这也要求测试这边需要在测试过程中保证用例执行的顺序，才不会偏离预期的结果。

在上一篇文章时，我的处理方案是：

1. 给同一批用例使用同一个UUID作为区分，并且给某一个用例设置一个 is_first=True 的标志。
2. 在执行过程中，轮到某个UUID时，会先去判断是否为 is_first ，如果是就直接执行，如果不是，就会去一个文件中检查这个UUID的首位是否已经执行，如果未执行则等到首位执行完后继续。

这个方案大体也能解决顺序的问题，但是存在一个比较严重的问题就是等待时间过长。甚至在某种程度上抵消了 xdist 的并发性。

为什么？

这里涉及到 xdist 的默认用例分配方案，在 xdist 中这些方案一般称之为 Schedule（别问，问就是源码都是这么写的）。默认的方案会将用例平均分配到每个 Worker 中，这样就会导致某个首位用例，会排在 Worker 用例列表的靠后的地方，那另一个Worker 就被迫要等执行到这个用例之后才能继续。

虽然较之前30分钟到一个小时的执行时间来说，已经快了一些了，但是依旧不尽如人意。那就改呗。

问题很明显，就是因为同一个 UUID 的用例分散在了不同的 Worker 中执行，那就让同 UUID 的用例都在同一个 Worker 中执行就行。

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

通过一番搜索后以及根据搜索内容再对源码进行阅读后发现了以下代码：

# 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

解释一下，这段代码的作用就是把所有已经搜集到的用例进行一次循环，然后将相同 scope 值的用例分发到同一个 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)

简单调试了一下之后发现 pytest_xdist_make_scheduler 是正常生效的，同时日志也输出了如下内容

4 workers [200 items]   
scheduling tests via HashScheduler

说明，GPT给的答案是有效的，于是在将 _split_scope 方法重写之后，发现日志顺利的打印上了。

_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]

后续再根据这个名称去获取到我需要的 UUID 后，将UUID 返回即可。

最终代码如下

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

    def _split_scope(self, nodeid: str) -> str:
        """
        返回用例分类依据
        相同的依据会使用同一个worker执行
        """
        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 中执行的问题。