diff --git a/src/youtube_extension/integrations/cloud_ai/providers/azure_vision.py b/src/youtube_extension/integrations/cloud_ai/providers/azure_vision.py index 9f7cf296d..b05e4c82f 100644 --- a/src/youtube_extension/integrations/cloud_ai/providers/azure_vision.py +++ b/src/youtube_extension/integrations/cloud_ai/providers/azure_vision.py @@ -8,6 +8,7 @@ - Custom Vision (if configured) """ +import asyncio import logging from datetime import datetime from typing import Any, Optional @@ -29,6 +30,12 @@ logger = logging.getLogger(__name__) +def _read_file_bytes(path: str) -> bytes: + """Read a file's bytes. Module-level so it can run in a worker thread.""" + with open(path, 'rb') as handle: + return handle.read() + + class AzureVision(BaseCloudAI): """Microsoft Azure AI Vision integration.""" @@ -251,9 +258,8 @@ async def _prepare_image_input(self, image_url: str) -> Optional[bytes]: # For URL input, Azure can analyze directly return None else: - # For local files, read content - with open(image_url, 'rb') as image_file: - return image_file.read() + # Local file - read off the event loop + return await asyncio.to_thread(_read_file_bytes, image_url) async def _await_ocr_call(self, deadline: float, func: Any, *args: Any, **kwargs: Any) -> Any: """Run a blocking Azure SDK call in a worker thread, bounded by a @@ -263,8 +269,6 @@ async def _await_ocr_call(self, deadline: float, func: Any, *args: Any, **kwargs ``asyncio.wait_for`` enforces the remaining budget so neither a slow read nor a slow poll can run past the OCR timeout. Raises ``CloudAIError`` on expiry.""" - import asyncio - remaining = deadline - asyncio.get_running_loop().time() if remaining <= 0: raise CloudAIError("Azure OCR operation timed out") @@ -277,8 +281,6 @@ async def _await_ocr_call(self, deadline: float, func: Any, *args: Any, **kwargs async def _perform_ocr(self, image_url: str, image_stream: Optional[bytes]) -> dict[str, Any]: """Perform OCR using Azure Read API.""" - import asyncio - from azure.cognitiveservices.vision.computervision.models import ( OperationStatusCodes, ) @@ -322,8 +324,6 @@ async def _perform_ocr(self, image_url: str, image_stream: Optional[bytes]) -> d async def _perform_ocr_stream(self, image_stream) -> dict[str, Any]: """Perform OCR on image stream.""" - import asyncio - from azure.cognitiveservices.vision.computervision.models import ( OperationStatusCodes, ) diff --git a/src/youtube_extension/integrations/cloud_ai/providers/google_cloud.py b/src/youtube_extension/integrations/cloud_ai/providers/google_cloud.py index 9200d25ee..98e6de8c3 100644 --- a/src/youtube_extension/integrations/cloud_ai/providers/google_cloud.py +++ b/src/youtube_extension/integrations/cloud_ai/providers/google_cloud.py @@ -31,6 +31,12 @@ logger = logging.getLogger(__name__) +def _read_file_bytes(path: str) -> bytes: + """Read a file's bytes. Module-level so it can run in a worker thread.""" + with open(path, 'rb') as handle: + return handle.read() + + class GoogleCloudAI(BaseCloudAI): """Google Cloud Video Intelligence and Vision API integration.""" @@ -181,9 +187,8 @@ async def analyze_image(self, image_url: str, if image_url.startswith(('http://', 'https://')): image.source.image_uri = image_url else: - # For local files - with open(image_url, 'rb') as image_file: - image.content = image_file.read() + # Local file - read off the event loop + image.content = await asyncio.to_thread(_read_file_bytes, image_url) # Prepare features features = self._prepare_vision_features(analysis_types) diff --git a/tests/unit/test_azure_vision_provider.py b/tests/unit/test_azure_vision_provider.py index 4b6839796..98d6c2d7b 100644 --- a/tests/unit/test_azure_vision_provider.py +++ b/tests/unit/test_azure_vision_provider.py @@ -2,7 +2,9 @@ from __future__ import annotations +import builtins import sys +import threading import types as _types from pathlib import Path from unittest.mock import AsyncMock, MagicMock, patch @@ -1194,3 +1196,92 @@ def test_cost_scales_with_analysis_types(self): cost1 = provider.estimate_cost(60.0, [AnalysisType.OCR]) cost2 = provider.estimate_cost(60.0, [AnalysisType.OCR, AnalysisType.FACE_DETECTION]) assert cost2 > cost1 + + +# =========================================================================== +# Local image reads must not block the event loop +# =========================================================================== + + +class _ThreadRecordingHandle: + """File-object proxy that records the calling thread on every ``read``.""" + + def __init__(self, handle, threads): + self._handle = handle + self._threads = threads + + def read(self, *args, **kwargs): + self._threads.append(threading.get_ident()) + return self._handle.read(*args, **kwargs) + + def __enter__(self): + self._handle.__enter__() + return self + + def __exit__(self, *exc_info): + return self._handle.__exit__(*exc_info) + + def __getattr__(self, name): + return getattr(self._handle, name) + + +class _ThreadRecordingOpen: + """Wrap ``builtins.open`` and record which thread *reads* a target path. + + Off-loop execution is asserted by *thread identity* rather than elapsed + wall-clock time, which is flaky on loaded CI runners. The recording hooks + ``read()`` on the returned handle rather than ``open()`` itself, so a + regression that opens the file on a worker thread but reads its bytes back + on the event loop is still caught. Only the target path is wrapped so + unrelated ``open`` traffic (logging, coverage) cannot contaminate the + result. + """ + + def __init__(self, target): + self._real_open = builtins.open + self._target = str(target) + self.threads: list[int] = [] + + def __call__(self, file, *args, **kwargs): + handle = self._real_open(file, *args, **kwargs) + if str(file) == self._target: + return _ThreadRecordingHandle(handle, self.threads) + return handle + + +class TestAzureVisionImageReadOffEventLoop: + async def test_local_file_read_runs_on_worker_thread(self, tmp_path): + provider = _make_provider() + img_file = tmp_path / "frame.jpg" + img_file.write_bytes(b"\xff\xd8\xff\xe0") + recorder = _ThreadRecordingOpen(img_file) + loop_thread = threading.get_ident() + + with patch("builtins.open", recorder): + result = await provider._prepare_image_input(str(img_file)) + + assert result == b"\xff\xd8\xff\xe0" + assert recorder.threads, "expected the provider to read the local image file" + assert loop_thread not in recorder.threads, ( + "local image bytes were read on the event loop thread; the read must " + "be offloaded to a worker thread" + ) + + async def test_http_url_performs_no_disk_read(self, tmp_path): + """The URL branch must remain untouched: Azure fetches it directly.""" + provider = _make_provider() + decoy = tmp_path / "unused.jpg" + decoy.write_bytes(b"\x00") + recorder = _ThreadRecordingOpen(decoy) + + with patch("builtins.open", recorder): + result = await provider._prepare_image_input("https://example.com/img.jpg") + + assert result is None + assert recorder.threads == [] + + async def test_missing_file_still_raises_file_not_found(self, tmp_path): + """Offloading must not swallow or re-wrap I/O errors.""" + provider = _make_provider() + with pytest.raises(FileNotFoundError): + await provider._prepare_image_input(str(tmp_path / "does-not-exist.jpg")) diff --git a/tests/unit/test_google_cloud_provider.py b/tests/unit/test_google_cloud_provider.py index 0cef59252..22d50820b 100644 --- a/tests/unit/test_google_cloud_provider.py +++ b/tests/unit/test_google_cloud_provider.py @@ -3,7 +3,9 @@ from __future__ import annotations import asyncio +import builtins import sys +import threading import types as _types from pathlib import Path from unittest.mock import AsyncMock, MagicMock, patch @@ -1115,3 +1117,144 @@ def test_zero_duration_zero_cost(self): provider = _make_provider() cost = provider.estimate_cost(0.0, [AnalysisType.LABEL_DETECTION]) assert cost == pytest.approx(0.0) + + +# =========================================================================== +# Local image reads must not block the event loop +# =========================================================================== + + +class _ThreadRecordingHandle: + """File-object proxy that records the calling thread on every ``read``.""" + + def __init__(self, handle, threads): + self._handle = handle + self._threads = threads + + def read(self, *args, **kwargs): + self._threads.append(threading.get_ident()) + return self._handle.read(*args, **kwargs) + + def __enter__(self): + self._handle.__enter__() + return self + + def __exit__(self, *exc_info): + return self._handle.__exit__(*exc_info) + + def __getattr__(self, name): + return getattr(self._handle, name) + + +class _ThreadRecordingOpen: + """Wrap ``builtins.open`` and record which thread *reads* a target path. + + Off-loop execution is asserted by *thread identity* rather than elapsed + wall-clock time, which is flaky on loaded CI runners. The recording hooks + ``read()`` on the returned handle rather than ``open()`` itself, so a + regression that opens the file on a worker thread but reads its bytes back + on the event loop is still caught. Only the target path is wrapped so + unrelated ``open`` traffic (logging, coverage) cannot contaminate the + result. + """ + + def __init__(self, target): + self._real_open = builtins.open + self._target = str(target) + self.threads: list[int] = [] + + def __call__(self, file, *args, **kwargs): + handle = self._real_open(file, *args, **kwargs) + if str(file) == self._target: + return _ThreadRecordingHandle(handle, self.threads) + return handle + + +class TestGoogleCloudImageReadOffEventLoop: + def _vision_modules(self): + mock_image_instance = MagicMock() + mock_image_instance.source = MagicMock() + mock_image_cls = MagicMock(return_value=mock_image_instance) + mock_feature_type = MagicMock() + mock_feature_type.LABEL_DETECTION = "LABEL_DETECTION" + mock_feature_cls = MagicMock() + mock_feature_cls.Type = mock_feature_type + mock_vision = MagicMock() + mock_vision.Image = mock_image_cls + mock_vision.Feature = mock_feature_cls + return mock_vision + + def _patched_modules(self, mock_vision): + return patch.dict("sys.modules", { + "google": _types.ModuleType("google"), + "google.cloud": _types.ModuleType("google.cloud"), + "google.cloud.vision": mock_vision, + }) + + def _client(self): + client = AsyncMock() + client.annotate_image = AsyncMock(return_value=_make_vision_response()) + return client + + async def test_local_file_read_runs_on_worker_thread(self, tmp_path): + provider = _make_provider() + provider._vision_client = self._client() + img_file = tmp_path / "frame.jpg" + img_file.write_bytes(b"\x89PNG\r\n") + mock_vision = self._vision_modules() + recorder = _ThreadRecordingOpen(img_file) + loop_thread = threading.get_ident() + + with self._patched_modules(mock_vision), patch("builtins.open", recorder): + await provider.analyze_image(str(img_file), [AnalysisType.LABEL_DETECTION]) + + assert mock_vision.Image.return_value.content == b"\x89PNG\r\n" + assert recorder.threads, "expected the provider to read the local image file" + assert loop_thread not in recorder.threads, ( + "local image bytes were read on the event loop thread; the read must " + "be offloaded to a worker thread" + ) + + async def test_http_url_performs_no_disk_read(self, tmp_path): + """The URI branch must remain untouched: Vision fetches it directly.""" + provider = _make_provider() + provider._vision_client = self._client() + decoy = tmp_path / "unused.jpg" + decoy.write_bytes(b"\x00") + mock_vision = self._vision_modules() + recorder = _ThreadRecordingOpen(decoy) + + with self._patched_modules(mock_vision), patch("builtins.open", recorder): + await provider.analyze_image( + "https://example.com/img.jpg", [AnalysisType.LABEL_DETECTION] + ) + + assert recorder.threads == [] + assert mock_vision.Image.return_value.source.image_uri == "https://example.com/img.jpg" + + async def test_missing_local_file_wrapped_in_cloud_ai_error(self, tmp_path): + """A missing local image surfaces as ``CloudAIError`` from the public API. + + Unlike Azure's private ``_prepare_image_input`` (which propagates + ``FileNotFoundError``), Google's public ``analyze_image`` catches it in + its broad ``except Exception`` and re-raises as ``CloudAIError``. Pin + that wrapper contract so moving the read off the loop cannot silently + alter how a missing file is reported. + """ + provider = _make_provider() + provider._vision_client = self._client() + missing = tmp_path / "does-not-exist.jpg" + mock_vision = self._vision_modules() + + with self._patched_modules(mock_vision): + with pytest.raises(CloudAIError) as exc_info: + await provider.analyze_image( + str(missing), [AnalysisType.LABEL_DETECTION] + ) + + # The provider re-raises without ``from e``, so the original error is + # carried on ``__context__`` (implicit chaining), not ``__cause__``. + assert isinstance(exc_info.value.__context__, FileNotFoundError), ( + "offloading must preserve the underlying I/O error in the exception chain" + ) + assert "No such file or directory" in str(exc_info.value)