如何在 Python 单元测试中断言在基于 class 的线程中抛出异常

Question

我在 python 中编写了类似于以下的代码 - 基于线程 class

from threading import Thread

class ThreadClass(Thread):

    def __init__(self, li):
        super(ThreadClass, self).__init__()
        self.li = li
        self.ps = list()

    def __validate(self, val):
        if val > 10:
            raise ValueError("Given value is greater than 10")

        return val % 10
    
    def __fun(self):
        ps = list()
        for i in self.li:
            p = self.__validate(i)
            ps.append(p)
        self.ps = ps
    
    def get_ps(self):
        return self.ps

    def run(self):
        self.__fun()

和下面的单元测试来测试一个失败场景

from unittest import TestCase
from thread_class import ThreadClass

class TestThreadClassNegative(TestCase):

    def test_val_greater_than_valid_fail(self):
        get_data = ThreadClass(li = [2,3,5,11,6])
        with self.assertRaises(ValueError):
            get_data.start()
            get_data.join()

上面的测试失败说 AssertionError: ValueError not raised 但我可以清楚地看到引发了异常。

================================== FAILURES ===================================
__________ TestThreadClassNegative.test_val_greater_than_valid_fail ___________
[gw0] win32 -- Python 3.7.10 C:\Users\<user>\AppData\Local\Continuum\anaconda3\envs\venv\python.exe

self = <tests.unit.test_thread_class.TestThreadClassNegative testMethod=test_val_greater_than_valid_fail>

    def test_val_greater_than_valid_fail(self):
        get_data = ThreadClass(li=[2, 3, 5, 11, 6])
        with self.assertRaises(ValueError):
            get_data.start()
>           get_data.join()
E           AssertionError: ValueError not raised

test_thread_class.py:9: AssertionError
---------------------------- Captured stderr call -----------------------------
Exception in thread Thread-1:
Traceback (most recent call last):
  File "C:\Users\<user>\AppData\Local\Continuum\anaconda3\envs\venv\lib\threading.py", line 926, in _bootstrap_inner
    self.run()
  File "C:\Users\<user>\projects\pytest_project\thread_class.py", line 27, in run
    self.__fun()
  File "C:\Users\<user>\projects\pytest_project\thread_class.py", line 19, in __fun
    p = self.__validate(i)
  File "C:\Users\<user>\projects\pytest_project\thread_class.py", line 12, in __validate
    raise ValueError("Given value is greater than 10")
ValueError: Given value is greater than 10

为什么那个测试用例失败了？我做错了什么，我该如何解决？

Answer 1

在子线程中引发异常，终止它，到此结束。

那是因为在子线程中引发的异常不会传播到调用者线程。

一个可能的解决方案是创建一个 class 来捕获运行() 中的异常并在 join() 中引发它：

编辑：我搞砸了 subclassing，这次测试了它，应该可以工作

class Propogate(Thread):
    def __init__(self, cls):
        super().__init__(target=cls.run)
        self.ex = None

    def run(self):
        try:
            self._target()
        except BaseException as e:
            self.ex = e

    def join(self):
        super().join()
        if self.ex is not None:
            raise self.ex

更改此行的用法：

get_data = Propogate(ThreadClass(li = [2,3,5,11,6]))

并且不继承另一个线程class:

class ThreadClass():

我的回答是改编自这个answer

如何在 Python 单元测试中断言在基于 class 的线程中抛出异常

How to assert that exception is thrown in a Thread based class in Python unit tests

python

pytest

python-multithreading

python-unittest