Class TestConcurrentErrorPaths

tests/test_concurrency.py:261–344

并发触发错误路径 —— 确保 RLock 覆盖错误流。

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

class TestConcurrentErrorPaths:
    """并发触发错误路径 —— 确保 RLock 覆盖错误流。"""

    def test_concurrent_nonexistent_satellite_no_exception(self, isolated_engine):
        """并发指定不存在的卫星，每次都应以 status='error' 结果返回，不抛异常。"""
        bad_payload = {
            "data_type": "TASK_CMD",
            "data_size": 50,
            "priority": 5,
            "max_delay": 600,
            "satellite_id": "NONEXISTENT_SAT_CONCURRENT_TEST",
        }
        results = []
        errors = []
        barrier = threading.Barrier(6)

        def worker():
            barrier.wait()
            for _ in range(5):
                try:
                    r = isolated_engine.submit_request(dict(bad_payload))
                    results.append(r)
                except Exception as exc:
                    errors.append(exc)

        threads = [threading.Thread(target=worker) for _ in range(6)]
        for t in threads:
            t.start()
        for t in threads:
            t.join(timeout=30)

        assert errors == [], f"不存在卫星的并发调用出现异常: {errors}"
        for r in results:
            assert r.get("status") == "error"

    def test_rejected_count_incremented_for_missing_satellite(self, isolated_engine):
        """不存在卫星请求应计入 rejected_requests。"""
        bad_payload = dict(_VALID_PAYLOAD)
        bad_payload["satellite_id"] = "NO_SUCH_SAT_XYZ"
        n = 5
        before = isolated_engine.stats["rejected_requests"]
        for _ in range(n):
            isolated_engine.submit_request(dict(bad_payload))
        assert isolated_engine.stats["rejected_requests"] == before + n

    def test_mixed_valid_and_invalid_concurrent(self, isolated_engine):
        """合法请求与非法卫星请求混合并发，计数器仍自洽。"""
        good = dict(_VALID_PAYLOAD)
        bad = dict(_VALID_PAYLOAD)
        bad["satellite_id"] = "MIXED_BAD_SAT_99"

        results = []
        errors = []
        barrier = threading.Barrier(4)

        def good_worker():
            barrier.wait()
            for _ in range(5):