Class TestNoUncaughtExceptions

tests/load/test_stress_engine.py:152–181

批量提交不同规模的请求，引擎不得抛出未处理异常。

class TestNoUncaughtExceptions:
    """批量提交不同规模的请求，引擎不得抛出未处理异常。"""

    @pytest.mark.parametrize("n_requests", [10, 50, 200])
    def test_sequential_no_exception(self, isolated_engine, n_requests: int):
        """顺序提交 n_requests 次，无任何异常。"""
        payload = dict(_HIGH_PRI_PAYLOAD)
        for _ in range(n_requests):
            result = isolated_engine.submit_request(dict(payload))
            assert isinstance(result, dict), "submit_request 应返回字典"

    @pytest.mark.parametrize("n_threads,per_thread", [(4, 10), (8, 25), (16, 12)])
    def test_concurrent_no_exception(self, isolated_engine, n_threads: int, per_thread: int):
        """多线程并发提交，无任何未捕获异常。"""
        results, errors = _submit_concurrent(isolated_engine, _HIGH_PRI_PAYLOAD, n_threads, per_thread)
        assert len(errors) == 0, f"发现未捕获异常: {errors[:3]}"
        assert len(results) == n_threads * per_thread

    @pytest.mark.parametrize("data_type", list(DATA_TYPES.keys()))
    def test_all_data_types_no_exception(self, isolated_engine, data_type: str):
        """所有数据类型各提交 20 次，无任何异常。"""
        payload = {
            "data_type": data_type,
            "data_size": 100,
            "priority": 5,
            "max_delay": 600,
        }
        for _ in range(20):
            result = isolated_engine.submit_request(dict(payload))
            assert isinstance(result, dict)


# ---------------------------------------------------------------------------