Class TestConcurrentSubmitCounting

tests/test_concurrency.py:133–226

多线程并发提交 —— 计数器自洽验证。

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

class TestConcurrentSubmitCounting:
    """多线程并发提交 —— 计数器自洽验证。"""

    def test_no_exceptions_raised(self, isolated_engine):
        """并发提交不得抛出异常。"""
        _, errors = _concurrent_submit(isolated_engine, n_threads=5, per_thread=10)
        assert errors == [], f"并发提交出现异常: {errors}"

    def test_total_requests_equals_total_submitted(self, isolated_engine):
        """并发提交 N*M 次后，total_requests 增量应恰好为 N*M。"""
        n_threads, per_thread = 5, 20
        before = isolated_engine.stats["total_requests"]
        _, errors = _concurrent_submit(isolated_engine, n_threads, per_thread)
        assert errors == [], f"并发出现异常: {errors}"
        assert isolated_engine.stats["total_requests"] == before + n_threads * per_thread

    def test_all_results_returned(self, isolated_engine):
        """每次 submit_request 都应返回结果（不会静默丢失）。"""
        n_threads, per_thread = 4, 10
        results, errors = _concurrent_submit(isolated_engine, n_threads, per_thread)
        assert errors == []
        assert len(results) == n_threads * per_thread

    def test_results_are_dicts(self, isolated_engine):
        """每个返回值都应是字典。"""
        results, errors = _concurrent_submit(isolated_engine, n_threads=3, per_thread=5)
        assert errors == []
        for r in results:
            assert isinstance(r, dict), f"返回值应为 dict，实际为 {type(r)}"

    def test_results_have_id_field(self, isolated_engine):
        """每个结果字典应包含 id 字段。"""
        results, errors = _concurrent_submit(isolated_engine, n_threads=3, per_thread=5)
        assert errors == []
        for r in results:
            assert "id" in r or r.get("status") == "error", (
                f"结果缺少 id 字段: {r}"
            )

    def test_results_have_status_field(self, isolated_engine):
        """每个结果字典应包含 status 字段。"""
        results, errors = _concurrent_submit(isolated_engine, n_threads=3, per_thread=5)
        assert errors == []
        for r in results:
            assert "status" in r, f"结果缺少 status 字段: {r}"

    def test_no_duplicate_ids(self, isolated_engine):
        """并发提交产生的请求 ID 应唯一（若结果含 id）。"""
        results, errors = _concurrent_submit(isolated_engine, n_threads=4, per_thread=10)
        assert errors == []
        ids = [r.get("id") for r in results if r.get("id") is not None]
        assert len(ids) == len(set(ids)), "出现重复 id"

    def test_request_list_consistent_with_counter(self, isolated_engine):
        """transmission_requests + request_history 长度等于 total_requests 增量。"""
        n_threads, per_thread = 4, 15
        before_total = isolated_engine.stats["total_requests"]
        before_lists = (