fix: prevent isolated request cancellation from aborting tests (#1407)

Author: Guovin

tests/test_speed_dns_errors.py

@@ -1,12 +1,47 @@
 import asyncio
 import socket
 import unittest
+from contextlib import asynccontextmanager
+from types import SimpleNamespace
+from unittest.mock import AsyncMock, patch
 
 from aiohttp import ClientSession, TCPConnector
 
 from utils.speed import create_speed_test_session, _install_aiohttp_dns_error_filter
 
 
+class MemoryArtifactWriter:
+    def __init__(self, *_args, **_kwargs):
+        self.count = 0
+
+    def write(self, _record):
+        self.count += 1
+
+    def close(self):
+        pass
+
+
+@asynccontextmanager
+async def speed_session(_concurrency):
+    yield object()
+
+
+def speed_test_config():
+    return SimpleNamespace(
+        open_ipv6=False,
+        speed_test_mode="full",
+        open_full_speed_test=True,
+        open_filter_resolution=False,
+        open_stream_screenshot=False,
+        performance_settings=SimpleNamespace(
+            speed_test_concurrency=1,
+            probe_concurrency=1,
+        ),
+        speed_test_target=1,
+        speed_test_timeout=1,
+    )
+
+
 class FailingResolver:
     async def resolve(self, host, port=0, family=socket.AF_INET):
         await asyncio.sleep(0.02)
@@ -59,3 +94,64 @@ async def test_unrelated_asyncio_exception_uses_previous_handler(self):
         loop.call_exception_handler(context)
 
         self.assertEqual(contexts, [context])
+
+    async def test_isolated_request_cancellation_does_not_cancel_batch(self):
+        from utils.channel import test_speed
+
+        channel_data = {
+            "Test": {
+                "Channel": [{
+                    "url": "http://unresolvable.invalid/live",
+                    "host": "unresolvable.invalid",
+                    "resolution": None,
+                    "ipv_type": "ipv4",
+                    "origin": "subscribe",
+                }]
+            }
+        }
+        with (
+            patch("utils.channel.config", speed_test_config()),
+            patch("utils.channel.ArtifactWriter", MemoryArtifactWriter),
+            patch("utils.channel.create_speed_test_session", speed_session),
+            patch("utils.channel.get_speed", AsyncMock(side_effect=asyncio.CancelledError)),
+            patch("utils.channel.mark_url_bad"),
+            patch("utils.channel.mark_url_good"),
+        ):
+            result = await test_speed(channel_data)
+
+        item = result["Test"]["Channel"][0]
+        self.assertEqual(item["test_status"], "request_error")
+        self.assertEqual(item["error_type"], "CancelledError")
+
+    async def test_real_batch_cancellation_still_propagates(self):
+        from utils.channel import test_speed
+
+        started = asyncio.Event()
+
+        async def wait_for_cancellation(*_args, **_kwargs):
+            started.set()
+            await asyncio.Event().wait()
+
+        channel_data = {
+            "Test": {
+                "Channel": [{
+                    "url": "http://example.invalid/live",
+                    "host": "example.invalid",
+                    "resolution": None,
+                    "ipv_type": "ipv4",
+                    "origin": "subscribe",
+                }]
+            }
+        }
+
+        with (
+            patch("utils.channel.config", speed_test_config()),
+            patch("utils.channel.ArtifactWriter", MemoryArtifactWriter),
+            patch("utils.channel.create_speed_test_session", speed_session),
+            patch("utils.channel.get_speed", wait_for_cancellation),
+        ):
+            task = asyncio.create_task(test_speed(channel_data))
+            await started.wait()
+            task.cancel()
+            with self.assertRaises(asyncio.CancelledError):
+                await task

utils/channel.py

@@ -1006,6 +1006,20 @@ async def worker():
                                 http_semaphore=http_semaphore,
                                 probe_semaphore=probe_semaphore,
                             )
+                    except asyncio.CancelledError:
+                        task = asyncio.current_task()
+                        if task is not None and task.cancelling():
+                            raise
+                        result = {
+                            "speed": 0,
+                            "delay": -1,
+                            "resolution": None,
+                            "fps": None,
+                            "video_codec": None,
+                            "audio_codec": None,
+                            "test_status": "request_error",
+                            "error_type": "CancelledError",
+                        }
                     except TimeoutError:
                         result = {
                             "speed": 0,