Python 面试题之多方法实现程序超时检测报错

题目描述：检测当前程序运行如果超过了我们指定的时间则为失败报错！

下面是需要检测的代码信息

class SimpleTest():
    def __init__(self):
        self.tests = []
       

    def add_test(self, test_func):
        self.tests.append(test_func)
    
    def run_tests(self):
        for test_func in self.tests:
            print(f"Running {test_func.__name__}...")
            try:
                test_func()
                print(f"{test_func.__name__} passed\n")
            except Exception as e:
                print(f"{test_func.__name__} failed: {e}\n")


# 示例测试用例
def test_example():
    assert 1 + 1 == 2


def test_failure():
    assert 1 + 1 == 3


# 创建测试框架实例
test_runner = SimpleTest()
# 添加测试用例
test_runner.add_test(test_example)  # 该测试用例应通过
test_runner.add_test(test_failure)  # 该测试用例应失败
# 运行测试用例
test_runner.run_tests()

使用装饰器实现

这样的目的是我们不需要入侵源代码，就可以实现该功能！

import datetime
import functools
import time

def run_time(expire_time):
    def decorator(func):
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            start_time = datetime.datetime.now()
            try:
                result = func(*args, **kwargs)
            finally:
                end_time = datetime.datetime.now()
                elapsed_time = end_time - start_time
                if elapsed_time > expire_time:
                    raise TimeoutError(f"Function {func.__name__} timed out after {elapsed_time} (longer than {expire_time})")
            return result
        return wrapper
    return decorator

class SimpleTest:
    def __init__(self):
        self.tests = []
       
    def add_test(self, test_func):
        self.tests.append(test_func)
    
    @run_time(expire_time=datetime.timedelta(seconds=2))
    def run_individual_test(self, test_func):
        print(f"Running {test_func.__name__}...")
        try:
            test_func()
            print(f"{test_func.__name__} passed\n")
        except Exception as e:
            print(f"{test_func.__name__} failed: {e}\n")
    
    def run_tests(self):
        for test_func in self.tests:
            try:
                self.run_individual_test(test_func)
            except TimeoutError as te:
                print(te)

# 示例测试用例
def test_example():
    time.sleep(1)  # Simulate a test that takes 1 second
    assert 1 + 1 == 2

def test_failure():
    time.sleep(3)  # Simulate a test that takes 3 seconds and should timeout
    assert 1 + 1 == 3

# 创建测试框架实例
test_runner = SimpleTest()
# 添加测试用例
test_runner.add_test(test_example)  # 该测试用例应通过
test_runner.add_test(test_failure)  # 该测试用例应失败（超时）
# 运行测试用例
test_runner.run_tests()

代码解释：通过装饰器传入的过期时间，然后获取运行前的时间和运行后的时间，两者进行对比，如果差值小于指定的时间则通过，否则失败！

运行结果如下：

Running test_example...
test_example passed

Running test_failure...
test_failure failed: 

Function run_individual_test timed out after 0:00:03.000197 (longer than 0:00:02)

使用线程池实现

如果我们想要通过线程来实现则可以:

import time
from concurrent.futures import ThreadPoolExecutor, as_completed, TimeoutError

class SimpleTest:
    def __init__(self):
        self.tests = []

    def add_test(self, test_func):
        self.tests.append(test_func)

    def run_tests_with_timeout(self, timeout_seconds):
        with ThreadPoolExecutor(max_workers=len(self.tests)) as executor:
            future_to_test = {executor.submit(test_func): test_func for test_func in self.tests}
            try:
                for future in as_completed(future_to_test, timeout=timeout_seconds):
                    test_func = future_to_test[future]
                    try:
                        future.result()  # 这将等待任务完成或抛出异常
                        print(f"{test_func.__name__} passed")
                    except Exception as e:
                        print(f"{test_func.__name__} failed: {e}")
            except TimeoutError:
                print("One or more tests timed out.")
                # 注意：这里不能直接中止未完成的测试，但你可以记录它们的状态

# 示例测试用例
def test_example():
    time.sleep(1)  # 模拟一个耗时1秒的测试
    assert 1 + 1 == 2, "Assertion failed in test_example"

def test_failure():
    time.sleep(3)  # 模拟一个耗时3秒并会超时的测试
    assert 1 + 1 == 3, "Assertion failed in test_failure"

# 创建测试框架实例
test_runner = SimpleTest()
# 添加测试用例
test_runner.add_test(test_example)  # 该测试用例应通过
test_runner.add_test(test_failure)  # 该测试用例应超时
# 运行测试用例并设置超时时间为2秒
test_runner.run_tests_with_timeout(2)

代码解释： 当前通过线程池启动，获取当前test_func的长度，作为启动线程池的个数，通过as_completed 去检测当前是否完成，然后对比超时时间，超过了则报错！

运行结果:

test_example passed
One or more tests timed out.

总结

上述还有其他的办法，目前在面试中可以采取这两种！