From efc0e5355ac22129e6b2f9c3ca5ee6abf388df9b Mon Sep 17 00:00:00 2001 From: Hayden <154503486+groupthinking@users.noreply.github.com> Date: Sun, 2 Aug 2026 09:41:49 -0500 Subject: [PATCH 1/4] perf: read local image bytes off the event loop in vision providers Azure `_prepare_image_input` and Google `analyze_image` read local image files with a synchronous `open().read()` inside `async def`, blocking the event loop for the duration of the disk read. Every other coroutine on the loop stalls until the read completes. Both now delegate to a module-level `_read_file_bytes` helper via `asyncio.to_thread`, matching the fix already merged for the AWS Rekognition sibling in #1205. This completes that cross-provider work so all three providers share one contract. The URL branches are untouched: Azure returns None so the SDK fetches the URL itself, and Google still sets `image.source.image_uri`. Also drops three now-redundant function-local `import asyncio` statements in azure_vision.py, made dead by the new module-level import. The lazy Azure SDK imports beside them are left in place. Regression tests assert off-loop execution by thread identity rather than wall-clock timing, which is flaky under CI load. Both new tests fail against the unpatched sources with "read on the event loop thread". Refs #1232 Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- .../cloud_ai/providers/azure_vision.py | 18 ++-- .../cloud_ai/providers/google_cloud.py | 11 ++- tests/unit/test_azure_vision_provider.py | 65 ++++++++++++++ tests/unit/test_google_cloud_provider.py | 90 +++++++++++++++++++ 4 files changed, 172 insertions(+), 12 deletions(-) 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..6f61cf6c8 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,66 @@ 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 _ThreadRecordingOpen: + """Wrap ``builtins.open`` and record which thread opened a target path. + + Off-loop execution is asserted by *thread identity* rather than elapsed + wall-clock time, which is flaky on loaded CI runners. Only calls for the + target path are recorded 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): + if str(file) == self._target: + self.threads.append(threading.get_ident()) + return self._real_open(file, *args, **kwargs) + + +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 open 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..b6a48d1f4 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,91 @@ 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 _ThreadRecordingOpen: + """Wrap ``builtins.open`` and record which thread opened a target path. + + Off-loop execution is asserted by *thread identity* rather than elapsed + wall-clock time, which is flaky on loaded CI runners. Only calls for the + target path are recorded 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): + if str(file) == self._target: + self.threads.append(threading.get_ident()) + return self._real_open(file, *args, **kwargs) + + +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 open 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" From 86371fc43e7ee25c3915293ea322884e637c5693 Mon Sep 17 00:00:00 2001 From: Claude Date: Sun, 2 Aug 2026 14:51:26 +0000 Subject: [PATCH 2/4] test(vision): record read-thread, not open-thread, in off-loop guards Copilot review on #1233 flagged that _ThreadRecordingOpen recorded the thread that called open(), not the thread that performed handle.read(). A regression offloading only open() while reading bytes back on the event loop would still pass, so the test did not prove #1232's required property. Wrap the returned handle in _ThreadRecordingHandle and record the calling thread on read() instead. Behaviour-preservation tests (URL branch, missing file) are unchanged; all off-loop guards still pass, and the recorder now fails on an open-offloaded/read-on-loop regression. Co-Authored-By: Claude Opus 4.8 (1M context) Claude-Session: https://claude.ai/code/session_01A6r6TyHudJkb6mn4HUNF5Y --- tests/unit/test_azure_vision_provider.py | 40 +++++++++++++++++++----- tests/unit/test_google_cloud_provider.py | 40 +++++++++++++++++++----- 2 files changed, 66 insertions(+), 14 deletions(-) diff --git a/tests/unit/test_azure_vision_provider.py b/tests/unit/test_azure_vision_provider.py index 6f61cf6c8..98d6c2d7b 100644 --- a/tests/unit/test_azure_vision_provider.py +++ b/tests/unit/test_azure_vision_provider.py @@ -1203,13 +1203,38 @@ def test_cost_scales_with_analysis_types(self): # =========================================================================== +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 opened a target path. + """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. Only calls for the - target path are recorded so unrelated ``open`` traffic (logging, coverage) - cannot contaminate the result. + 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): @@ -1218,9 +1243,10 @@ def __init__(self, target): self.threads: list[int] = [] def __call__(self, file, *args, **kwargs): + handle = self._real_open(file, *args, **kwargs) if str(file) == self._target: - self.threads.append(threading.get_ident()) - return self._real_open(file, *args, **kwargs) + return _ThreadRecordingHandle(handle, self.threads) + return handle class TestAzureVisionImageReadOffEventLoop: @@ -1235,7 +1261,7 @@ async def test_local_file_read_runs_on_worker_thread(self, tmp_path): result = await provider._prepare_image_input(str(img_file)) assert result == b"\xff\xd8\xff\xe0" - assert recorder.threads, "expected the provider to open the local image file" + 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" diff --git a/tests/unit/test_google_cloud_provider.py b/tests/unit/test_google_cloud_provider.py index b6a48d1f4..ae159f2a8 100644 --- a/tests/unit/test_google_cloud_provider.py +++ b/tests/unit/test_google_cloud_provider.py @@ -1124,13 +1124,38 @@ def test_zero_duration_zero_cost(self): # =========================================================================== +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 opened a target path. + """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. Only calls for the - target path are recorded so unrelated ``open`` traffic (logging, coverage) - cannot contaminate the result. + 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): @@ -1139,9 +1164,10 @@ def __init__(self, target): self.threads: list[int] = [] def __call__(self, file, *args, **kwargs): + handle = self._real_open(file, *args, **kwargs) if str(file) == self._target: - self.threads.append(threading.get_ident()) - return self._real_open(file, *args, **kwargs) + return _ThreadRecordingHandle(handle, self.threads) + return handle class TestGoogleCloudImageReadOffEventLoop: @@ -1183,7 +1209,7 @@ async def test_local_file_read_runs_on_worker_thread(self, tmp_path): 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 open the local image file" + 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" From e0961b2918545d1146f6e6bb67f8091715df60d4 Mon Sep 17 00:00:00 2001 From: Claude Date: Sun, 2 Aug 2026 14:56:57 +0000 Subject: [PATCH 3/4] test(google): pin missing-file CloudAIError wrapper contract MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit CodeRabbit review on #1233 flagged that TestGoogleCloudImageReadOffEventLoop had no missing-file test, while GoogleCloudAI.analyze_image catches FileNotFoundError in its broad `except Exception` and re-raises CloudAIError — unlike Azure's private _prepare_image_input, which propagates FileNotFoundError. Add a Google test asserting the CloudAIError wrapper so moving the read off the event loop cannot silently change how a missing local image is reported. Closes the coverage gap; documents that the two providers differ at the tested surface (public analyze_image vs private helper). Co-Authored-By: Claude Opus 4.8 (1M context) Claude-Session: https://claude.ai/code/session_01A6r6TyHudJkb6mn4HUNF5Y --- tests/unit/test_google_cloud_provider.py | 20 ++++++++++++++++++++ 1 file changed, 20 insertions(+) diff --git a/tests/unit/test_google_cloud_provider.py b/tests/unit/test_google_cloud_provider.py index ae159f2a8..03fa00ccf 100644 --- a/tests/unit/test_google_cloud_provider.py +++ b/tests/unit/test_google_cloud_provider.py @@ -1231,3 +1231,23 @@ async def test_http_url_performs_no_disk_read(self, tmp_path): 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): + await provider.analyze_image( + str(missing), [AnalysisType.LABEL_DETECTION] + ) From ff3bf2581e4d4fe3bc0dd068f9d52f92d5fb610d Mon Sep 17 00:00:00 2001 From: Hayden <154503486+groupthinking@users.noreply.github.com> Date: Sun, 2 Aug 2026 10:07:37 -0500 Subject: [PATCH 4/4] test(google): assert FileNotFoundError survives in the exception chain Strengthen the wrapper-contract guard added in e0961b29: asserting only pytest.raises(CloudAIError) would still pass if the underlying cause were swallowed or the message went generic. Also assert the original FileNotFoundError is preserved on __context__ (the provider re-raises without 'from e', so chaining is implicit) and that the path error text reaches the caller. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- tests/unit/test_google_cloud_provider.py | 9 ++++++++- 1 file changed, 8 insertions(+), 1 deletion(-) diff --git a/tests/unit/test_google_cloud_provider.py b/tests/unit/test_google_cloud_provider.py index 03fa00ccf..22d50820b 100644 --- a/tests/unit/test_google_cloud_provider.py +++ b/tests/unit/test_google_cloud_provider.py @@ -1247,7 +1247,14 @@ async def test_missing_local_file_wrapped_in_cloud_ai_error(self, tmp_path): mock_vision = self._vision_modules() with self._patched_modules(mock_vision): - with pytest.raises(CloudAIError): + 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)