From 095c712451839c1e6adc5f388180274193c8ad65 Mon Sep 17 00:00:00 2001 From: VsevolodX Date: Wed, 30 Sep 2026 11:25:31 -0700 Subject: [PATCH 01/21] fix(job-wait-loop): Abort and ESC cancel the task running the wait loop, per kernel run_interruptible_loop_async keeps asyncio.current_task() and the BroadcastChannel handler cancels it and aborts the in-flight fetch through a JS AbortController whose signal is passed to the loop body; the sleep between polls is one asyncio.sleep(poll_interval) instead of 0.05 s slices, so a cancel lands at once whether a poll or the sleep is awaited, and a throttled background tab costs one late poll instead of hundreds; CancelledError becomes UserAbortError. The channel is mat3ra_abort_, one per kernel process; each rendered control registers its channel on window.__mat3raAbortChannels keyed by its button, and ESC posts to the channels whose button sits inside .jp-NotebookPanel.jp-mod-current, else to none. Native Python is unchanged (Ctrl+C). Co-Authored-By: Claude Opus 5.5 (1M context) --- .../mat3ra/notebooks_utils/pyodide/runtime.py | 79 ++++++++++--------- tests/py/unit/test_jupyterlite_interrupts.py | 41 +++++++++- 2 files changed, 78 insertions(+), 42 deletions(-) diff --git a/src/py/mat3ra/notebooks_utils/pyodide/runtime.py b/src/py/mat3ra/notebooks_utils/pyodide/runtime.py index efac1ffa6..46a82f91a 100644 --- a/src/py/mat3ra/notebooks_utils/pyodide/runtime.py +++ b/src/py/mat3ra/notebooks_utils/pyodide/runtime.py @@ -14,13 +14,16 @@ HTML = None display = None +# One channel per kernel process: an abort reaches the loops of this kernel only. +ABORT_CHANNEL_NAME = f"mat3ra_abort_{uuid.uuid4().hex}" + class UserAbortError(RuntimeError): pass def display_abort_controls_in_current_cell_output( - channel_name: str = "mat3ra_abort_channel", + channel_name: str = ABORT_CHANNEL_NAME, abort_button_text: str = "Stop polling", ) -> None: """ @@ -55,14 +58,16 @@ def display_abort_controls_in_current_cell_output( (function() {{ const channelName = {channel_name!r}; - // Install ESC broadcaster once per page - if (!window.__mat3raEscapeAbortInstalled) {{ - window.__mat3raEscapeAbortInstalled = true; - const escChannel = new BroadcastChannel(channelName); + // Install ESC broadcaster once per page; ESC aborts the loop of the notebook in focus only + if (!window.__mat3raAbortChannels) {{ + window.__mat3raAbortChannels = new Map(); document.addEventListener("keydown", (event) => {{ - if (event.key === "Escape") {{ - escChannel.postMessage({{ type: "abort", source: "escape" }}); - }} + const notebookPanel = document.querySelector(".jp-NotebookPanel.jp-mod-current"); + window.__mat3raAbortChannels.forEach((buttonChannel, buttonElement) => {{ + if (event.key === "Escape" && notebookPanel?.contains(buttonElement)) {{ + buttonChannel.postMessage({{ type: "abort", source: "escape" }}); + }} + }}); }}, true); }} @@ -71,6 +76,7 @@ def display_abort_controls_in_current_cell_output( const buttonElement = document.getElementById("{element_id}_button"); const statusElement = document.getElementById("{element_id}_status"); if (!buttonElement) return; + window.__mat3raAbortChannels.set(buttonElement, buttonChannel); buttonElement.addEventListener("click", () => {{ buttonChannel.postMessage({{ type: "abort", source: "button" }}); @@ -89,17 +95,22 @@ def display_abort_controls_in_current_cell_output( class BroadcastChannelAbortController(BaseModel): """ WebWorker-side receiver. Works only in pyodide (emscripten). - In regular Python: start() does nothing and is_aborted stays False. + An abort message cancels the task running the loop and aborts the fetch given `fetch_abort_signal`. + In regular Python: start() does nothing and `fetch_abort_signal` stays None. """ - channel_name: str = "mat3ra_abort_channel" - is_aborted: bool = False + channel_name: str = ABORT_CHANNEL_NAME def model_post_init(self, __context: Any) -> None: self._broadcast_channel = None self._on_message_proxy = None + self._fetch_abort_controller = None + + @property + def fetch_abort_signal(self) -> Any: + return getattr(self._fetch_abort_controller, "signal", None) - def start(self) -> None: + def start(self, task: "asyncio.Task[Any]") -> None: if ENVIRONMENT != EnvironmentsEnum.PYODIDE: return if self._broadcast_channel is not None: @@ -109,11 +120,13 @@ def start(self) -> None: from pyodide.ffi import create_proxy # type: ignore self._broadcast_channel = js.BroadcastChannel.new(self.channel_name) + self._fetch_abort_controller = js.AbortController.new() def on_message(event) -> None: message = getattr(event, "data", None) if message and getattr(message, "type", None) == "abort": - self.is_aborted = True + self._fetch_abort_controller.abort() # type: ignore + task.cancel() self._on_message_proxy = create_proxy(on_message) self._broadcast_channel.onmessage = self._on_message_proxy # type: ignore @@ -131,43 +144,33 @@ def stop(self) -> None: async def run_interruptible_loop_async( - loop_body: Callable[[], Awaitable[bool]], + loop_body: Callable[[Any], Awaitable[bool]], poll_interval_seconds: float, *, - channel_name: str = "mat3ra_abort_channel", - check_interval_seconds: float = 0.05, + channel_name: str = ABORT_CHANNEL_NAME, show_controls: bool = True, ) -> None: """ Wraps an async loop around a "poll" function that returns True to continue, False to stop. - loop_body(): - - do one "poll" iteration + loop_body(abort_signal): + - do one "poll" iteration; `abort_signal` aborts its fetch (pyodide), None in regular Python - return True to keep looping, False to stop normally - Between iterations we sleep in small slices so: - - pyodide: ESC/button can be received and stop the loop - - regular Python: yields control (Ctrl+C/Stop works where supported) + pyodide: ESC/button cancels the task running the loop, during a poll or the sleep, raising UserAbortError. + regular Python: Ctrl+C/Stop interrupts the kernel. """ broadcast_channel_abort_controller = BroadcastChannelAbortController(channel_name=channel_name) - broadcast_channel_abort_controller.start() + broadcast_channel_abort_controller.start(asyncio.current_task()) # type: ignore if show_controls and ENVIRONMENT == EnvironmentsEnum.PYODIDE: display_abort_controls_in_current_cell_output(channel_name=channel_name, abort_button_text="Abort") try: - while True: - should_continue = await loop_body() - if not should_continue: - return - - remaining_seconds = float(poll_interval_seconds) - while remaining_seconds > 0: - if broadcast_channel_abort_controller.is_aborted: - raise UserAbortError("Aborted by user.") - await asyncio.sleep(min(check_interval_seconds, remaining_seconds)) - remaining_seconds -= check_interval_seconds - + while await loop_body(broadcast_channel_abort_controller.fetch_abort_signal): + await asyncio.sleep(poll_interval_seconds) + except asyncio.CancelledError: + raise UserAbortError("Aborted by user.") from None finally: broadcast_channel_abort_controller.stop() @@ -176,13 +179,12 @@ def interruptible_polling_loop( poll_interval_kwarg_name: str = "poll_interval", *, default_poll_interval_seconds: float = 10.0, - channel_name: str = "mat3ra_abort_channel", - check_interval_seconds: float = 0.05, + channel_name: str = ABORT_CHANNEL_NAME, show_controls: bool = True, ): """ Turns a poll-step function into an async loop. Wrapped fn returns True to continue, False to stop. - Sleeps in small slices so ESC/Abort (notebooks) or Ctrl+C can raise UserAbortError. + ESC/Abort (notebooks) raises UserAbortError at once, during a poll or the sleep; Ctrl+C natively. Poll interval: kwarg poll_interval_kwarg_name, else default_poll_interval_seconds. """ @@ -191,7 +193,7 @@ def decorator(poll_step_function: Callable[..., Any]) -> Callable[..., Any]: async def wrapped(*args: Any, **kwargs: Any) -> None: poll_interval_seconds = float(kwargs.pop(poll_interval_kwarg_name, default_poll_interval_seconds)) - async def loop_body() -> bool: + async def loop_body(abort_signal: Any) -> bool: result = poll_step_function(*args, **kwargs) should_continue = await result if inspect.isawaitable(result) else result return bool(should_continue) @@ -200,7 +202,6 @@ async def loop_body() -> bool: loop_body, poll_interval_seconds, channel_name=channel_name, - check_interval_seconds=check_interval_seconds, show_controls=show_controls, ) diff --git a/tests/py/unit/test_jupyterlite_interrupts.py b/tests/py/unit/test_jupyterlite_interrupts.py index 9b072e2b1..12178cc40 100644 --- a/tests/py/unit/test_jupyterlite_interrupts.py +++ b/tests/py/unit/test_jupyterlite_interrupts.py @@ -1,4 +1,7 @@ import asyncio +import subprocess +import sys +import time import pytest from mat3ra.notebooks_utils.pyodide.runtime import ( @@ -8,14 +11,18 @@ ) POLL_INTERVAL_SECONDS = 0.01 -CHECK_INTERVAL_SECONDS = 0.005 +ABORT_AFTER_SECONDS = 0.05 +ABORT_DEADLINE_SECONDS = 0.2 +READ_ABORT_CHANNEL_NAME = ( + "from mat3ra.notebooks_utils.pyodide.runtime import ABORT_CHANNEL_NAME; print(ABORT_CHANNEL_NAME)" +) @pytest.mark.asyncio async def test_run_interruptible_loop_async_stops_when_body_returns_false(): call_count = 0 - async def loop_body(): + async def loop_body(abort_signal): nonlocal call_count call_count += 1 return call_count < 3 @@ -24,11 +31,39 @@ async def loop_body(): loop_body, POLL_INTERVAL_SECONDS, show_controls=False, - check_interval_seconds=CHECK_INTERVAL_SECONDS, ) assert call_count == 3 +@pytest.mark.asyncio +@pytest.mark.parametrize( + ("poll_seconds", "poll_interval_seconds"), + [(10.0, POLL_INTERVAL_SECONDS), (0.0, 10.0)], + ids=["during a poll", "during the sleep"], +) +async def test_run_interruptible_loop_async_raises_user_abort_error_when_cancelled(poll_seconds, poll_interval_seconds): + async def loop_body(abort_signal): + await asyncio.sleep(poll_seconds) + return True + + task = asyncio.create_task(run_interruptible_loop_async(loop_body, poll_interval_seconds, show_controls=False)) + await asyncio.sleep(ABORT_AFTER_SECONDS) + started = time.monotonic() + task.cancel() + with pytest.raises(UserAbortError): + await task + assert time.monotonic() - started < ABORT_DEADLINE_SECONDS + + +def test_abort_channel_name_differs_per_kernel(): + channel_names = { + subprocess.run([sys.executable, "-c", READ_ABORT_CHANNEL_NAME], capture_output=True, text=True).stdout + for _ in range(2) + } + assert len(channel_names) == 2 + assert all(channel_name.startswith("mat3ra_abort_") for channel_name in channel_names) + + @pytest.mark.asyncio async def test_interruptible_polling_loop_decorator_returns_coroutine_and_runs_until_false(): call_count = 0 From f63fbdff62e5228a1461a78dfecc7771da655f45 Mon Sep 17 00:00:00 2001 From: VsevolodX Date: Wed, 30 Sep 2026 11:28:27 -0700 Subject: [PATCH 02/21] fix(job-wait-loop): The job status poll never blocks the kernel get_jobs_statuses_by_ids_async(endpoint, job_ids, timeout=30, abort_signal=None) fetches the statuses without holding the event loop: under Pyodide through pyodide.http.pyfetch with the URL built from endpoint.conn.preamble, the endpoint's bearer headers (or its X-Auth-Token headers when there is no access token, as the client does), and the loop's AbortSignal; natively endpoint.list runs in the default executor (loop.run_in_executor, as the repo's mypy hook checks against Python 3.8 where asyncio.to_thread does not exist); both are bounded by asyncio.wait_for(timeout), and an error status raises requests.HTTPError like the client. wait_for_jobs_to_finish_async becomes an async poll step on it, and the polling loop passes it the abort signal; the public call (endpoint, job_ids, poll_interval=) and the notebook call sites are unchanged. Previously the synchronous requests call inside the Pyodide worker held the event loop, so the abort message was never received while a request was in flight, and a hung request blocked forever. Co-Authored-By: Claude Opus 5.5 (1M context) --- src/py/mat3ra/notebooks_utils/api/job.py | 17 +++--- .../notebooks_utils/core/entity/job/api.py | 52 ++++++++++++++++++- .../mat3ra/notebooks_utils/pyodide/runtime.py | 5 +- tests/py/unit/core/entity/test_job_api.py | 18 +++++++ tests/py/unit/test_jupyterlite_interrupts.py | 26 +++++++++- 5 files changed, 108 insertions(+), 10 deletions(-) diff --git a/src/py/mat3ra/notebooks_utils/api/job.py b/src/py/mat3ra/notebooks_utils/api/job.py index 991ca56fe..89187b2c6 100644 --- a/src/py/mat3ra/notebooks_utils/api/job.py +++ b/src/py/mat3ra/notebooks_utils/api/job.py @@ -1,20 +1,24 @@ import datetime from collections import Counter -from typing import List +from typing import Any, List from mat3ra.api_client import JobEndpoints from mat3ra.utils.extra.tabulate import pretty_print -from ..core.entity.job.api import create_job, get_jobs_statuses_by_ids, save_files, submit_jobs +from ..core.entity.job.api import ( + create_job, + get_jobs_statuses_by_ids, + get_jobs_statuses_by_ids_async, + save_files, + submit_jobs, +) from ..pyodide.runtime import interruptible_polling_loop # Contains no external dependencies, only uses the API client, # so can be used in both regular Python and Pyodide environments. -# In regular Python, the interruptible_polling_loop does not actually do anything, -# so the function behaves like a normal polling loop. @interruptible_polling_loop() -def wait_for_jobs_to_finish_async(endpoint: JobEndpoints, job_ids: List[str]) -> bool: +async def wait_for_jobs_to_finish_async(endpoint: JobEndpoints, job_ids: List[str], abort_signal: Any) -> bool: """ Waits for jobs to finish and prints their statuses. A job is considered finished if it is not in "pre-submission", "submitted", or "active" status. @@ -22,8 +26,9 @@ def wait_for_jobs_to_finish_async(endpoint: JobEndpoints, job_ids: List[str]) -> Args: endpoint (JobEndpoints): Job endpoint object from the Exabyte API Client job_ids (list): list of job IDs to wait for + abort_signal: JS AbortSignal of the Abort control, passed in by the polling loop """ - statuses = get_jobs_statuses_by_ids(endpoint, job_ids) + statuses = await get_jobs_statuses_by_ids_async(endpoint, job_ids, abort_signal=abort_signal) counts = Counter(statuses) headers = ["TIME", "SUBMITTED-JOBS", "ACTIVE-JOBS", "FINISHED-JOBS", "ERRORED-JOBS"] now = datetime.datetime.now().strftime("%Y-%m-%d-%H:%M:%S") diff --git a/src/py/mat3ra/notebooks_utils/core/entity/job/api.py b/src/py/mat3ra/notebooks_utils/core/entity/job/api.py index 363c5b2f9..8e4f83297 100644 --- a/src/py/mat3ra/notebooks_utils/core/entity/job/api.py +++ b/src/py/mat3ra/notebooks_utils/core/entity/job/api.py @@ -1,9 +1,15 @@ +import asyncio +import json import re +import urllib.parse import urllib.request -from typing import Any, Dict, Iterable, List, Optional, Union +from typing import Any, Awaitable, Dict, Iterable, List, Optional, Union +import requests from mat3ra.api_client import APIClient, JobEndpoints +from ....primitive.environment import is_pyodide_environment + MATERIALS_SET_ENTITY_CLASS = "Material" @@ -40,6 +46,50 @@ def get_jobs_statuses_by_ids(endpoint: JobEndpoints, job_ids: List[str]) -> List return [job["status"] for job in jobs] +async def _list_jobs_with_fetch(endpoint: JobEndpoints, query: dict, projection: dict, abort_signal: Any) -> List[dict]: + """ + `endpoint.list` through the browser's fetch, which leaves the event loop free while the request is in flight. + Raises `requests.HTTPError` on an error status, as the API client does. + """ + from pyodide.http import pyfetch # type: ignore + + parameters = urllib.parse.urlencode({"query": json.dumps(query), "projection": json.dumps(projection)}) + url = urllib.parse.urljoin(endpoint.conn.preamble, f"{endpoint.name}?{parameters}") + response = await pyfetch(url, headers=endpoint._get_bearer_headers() or endpoint.headers, signal=abort_signal) + if not response.ok: + error_response = requests.Response() + error_response.status_code = response.status + raise requests.HTTPError(f"Error {response.status}.", response=error_response) + return (await response.json())["data"] + + +async def get_jobs_statuses_by_ids_async( + endpoint: JobEndpoints, job_ids: List[str], timeout: float = 30, abort_signal: Any = None +) -> List[str]: + """ + Gets jobs statuses by their IDs without blocking the event loop: through the browser's fetch in pyodide, + in a worker thread otherwise. + + Args: + endpoint (JobEndpoints): Job endpoint object from the Exabyte API Client + job_ids (list): list of job IDs to get the status for + timeout (float): seconds to wait for the response before raising asyncio.TimeoutError + abort_signal: JS AbortSignal that aborts the fetch in pyodide + + Returns: + list: list of job statuses + """ + query, projection = {"_id": {"$in": job_ids}}, {"fields": {"status": 1}} + + def request_jobs() -> Awaitable[List[dict]]: + if is_pyodide_environment(): + return _list_jobs_with_fetch(endpoint, query, projection, abort_signal) + return asyncio.get_running_loop().run_in_executor(None, endpoint.list, query, projection) + + jobs = await asyncio.wait_for(request_jobs(), timeout) + return [job["status"] for job in jobs] + + def _materials_set_reference(materials_set: Dict[str, Any]) -> Dict[str, str]: """ Builds the `_materialsSet` reference a job config expects. diff --git a/src/py/mat3ra/notebooks_utils/pyodide/runtime.py b/src/py/mat3ra/notebooks_utils/pyodide/runtime.py index 46a82f91a..1aaca02fe 100644 --- a/src/py/mat3ra/notebooks_utils/pyodide/runtime.py +++ b/src/py/mat3ra/notebooks_utils/pyodide/runtime.py @@ -183,7 +183,8 @@ def interruptible_polling_loop( show_controls: bool = True, ): """ - Turns a poll-step function into an async loop. Wrapped fn returns True to continue, False to stop. + Turns a poll-step function into an async loop. Wrapped fn takes `abort_signal` for its request and returns + True to continue, False to stop. ESC/Abort (notebooks) raises UserAbortError at once, during a poll or the sleep; Ctrl+C natively. Poll interval: kwarg poll_interval_kwarg_name, else default_poll_interval_seconds. """ @@ -194,7 +195,7 @@ async def wrapped(*args: Any, **kwargs: Any) -> None: poll_interval_seconds = float(kwargs.pop(poll_interval_kwarg_name, default_poll_interval_seconds)) async def loop_body(abort_signal: Any) -> bool: - result = poll_step_function(*args, **kwargs) + result = poll_step_function(*args, abort_signal=abort_signal, **kwargs) should_continue = await result if inspect.isawaitable(result) else result return bool(should_continue) diff --git a/tests/py/unit/core/entity/test_job_api.py b/tests/py/unit/core/entity/test_job_api.py index d676a45b5..f3e4ff116 100644 --- a/tests/py/unit/core/entity/test_job_api.py +++ b/tests/py/unit/core/entity/test_job_api.py @@ -1,3 +1,5 @@ +import asyncio +import threading from typing import Any, Dict, List from unittest.mock import MagicMock @@ -6,6 +8,7 @@ create_job, find_job_for_material, find_job_for_material_with_property, + get_jobs_statuses_by_ids_async, get_kgrid_of_job, get_kgrid_query, ) @@ -251,3 +254,18 @@ def test_find_job_for_material_matches_the_kgrid_of_the_unit(): ) def test_get_kgrid_of_job(unit, expected_kgrid): assert get_kgrid_of_job({"workflow": {"subworkflows": [{"units": [unit]}]}}) == expected_kgrid + + +REQUEST_TIMEOUT_SECONDS = 0.05 +BLOCKED_REQUEST_SECONDS = 1.0 + + +@pytest.mark.asyncio +async def test_get_jobs_statuses_by_ids_async_raises_when_the_request_times_out(): + release_request = threading.Event() + endpoint = MagicMock() + endpoint.list.side_effect = lambda query, projection: release_request.wait(BLOCKED_REQUEST_SECONDS) + + with pytest.raises(asyncio.TimeoutError): + await get_jobs_statuses_by_ids_async(endpoint, [CREATED_JOB["_id"]], timeout=REQUEST_TIMEOUT_SECONDS) + release_request.set() diff --git a/tests/py/unit/test_jupyterlite_interrupts.py b/tests/py/unit/test_jupyterlite_interrupts.py index 12178cc40..e548c7b20 100644 --- a/tests/py/unit/test_jupyterlite_interrupts.py +++ b/tests/py/unit/test_jupyterlite_interrupts.py @@ -1,9 +1,12 @@ import asyncio import subprocess import sys +import threading import time +from unittest.mock import MagicMock import pytest +from mat3ra.notebooks_utils.api.job import wait_for_jobs_to_finish_async from mat3ra.notebooks_utils.pyodide.runtime import ( UserAbortError, interruptible_polling_loop, @@ -13,6 +16,7 @@ POLL_INTERVAL_SECONDS = 0.01 ABORT_AFTER_SECONDS = 0.05 ABORT_DEADLINE_SECONDS = 0.2 +BLOCKED_REQUEST_SECONDS = 1.0 READ_ABORT_CHANNEL_NAME = ( "from mat3ra.notebooks_utils.pyodide.runtime import ABORT_CHANNEL_NAME; print(ABORT_CHANNEL_NAME)" ) @@ -55,6 +59,26 @@ async def loop_body(abort_signal): assert time.monotonic() - started < ABORT_DEADLINE_SECONDS +@pytest.mark.asyncio +async def test_wait_for_jobs_to_finish_async_raises_user_abort_error_while_the_status_request_blocks(): + release_request = threading.Event() + + def list_jobs(query, projection): + release_request.wait(BLOCKED_REQUEST_SECONDS) + return [{"status": "active"}] + + endpoint = MagicMock() + endpoint.list.side_effect = list_jobs + started = time.monotonic() + task = asyncio.create_task(wait_for_jobs_to_finish_async(endpoint, ["job-1"], poll_interval=POLL_INTERVAL_SECONDS)) + await asyncio.sleep(ABORT_AFTER_SECONDS) + task.cancel() + with pytest.raises(UserAbortError): + await task + release_request.set() + assert time.monotonic() - started < ABORT_DEADLINE_SECONDS + + def test_abort_channel_name_differs_per_kernel(): channel_names = { subprocess.run([sys.executable, "-c", READ_ABORT_CHANNEL_NAME], capture_output=True, text=True).stdout @@ -69,7 +93,7 @@ async def test_interruptible_polling_loop_decorator_returns_coroutine_and_runs_u call_count = 0 @interruptible_polling_loop(show_controls=False) - def poll_step(): + def poll_step(abort_signal): nonlocal call_count call_count += 1 return call_count < 2 From 42d7052266de191b60144580d8b2a5a5a25882e3 Mon Sep 17 00:00:00 2001 From: VsevolodX Date: Wed, 30 Sep 2026 11:46:24 -0700 Subject: [PATCH 03/21] test(job-wait-loop): The blocked-request abort test fails instead of hanging when the poll blocks The blocked status request returns a finished job, so an implementation that blocks the event loop ends its loop after one poll and fails with "DID NOT RAISE UserAbortError" in under 2 s; with an active job it looped forever, because on Python 3.11 asyncio.wait_for returns the inner result and drops the cancel when both land in the same tick, which a blocking call always causes. Co-Authored-By: Claude Opus 5.5 (1M context) --- tests/py/unit/test_jupyterlite_interrupts.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/tests/py/unit/test_jupyterlite_interrupts.py b/tests/py/unit/test_jupyterlite_interrupts.py index e548c7b20..a338d56c6 100644 --- a/tests/py/unit/test_jupyterlite_interrupts.py +++ b/tests/py/unit/test_jupyterlite_interrupts.py @@ -65,7 +65,7 @@ async def test_wait_for_jobs_to_finish_async_raises_user_abort_error_while_the_s def list_jobs(query, projection): release_request.wait(BLOCKED_REQUEST_SECONDS) - return [{"status": "active"}] + return [{"status": "finished"}] endpoint = MagicMock() endpoint.list.side_effect = list_jobs From 1acb4a2d7e52442996ed13771e4cf02f9f4a585e Mon Sep 17 00:00:00 2001 From: VsevolodX Date: Wed, 30 Sep 2026 11:51:31 -0700 Subject: [PATCH 04/21] fix(job-wait-loop): A rejected token during the job wait triggers one device login and a retry get_jobs_statuses_by_ids_async catches requests.HTTPError with status 401 from either path (the executor call to endpoint.list natively, the pyfetch path under Pyodide, which raises the same error), awaits auth.reauthenticate(endpoint._auth) and repeats the request once; a second 401 or any other error status raises as before. reauthenticate drops the cached entry for the OIDC URL from the token store (new token_store.delete_token), runs the device login (popup / printed URL as today) and sets the new token on the client's shared AuthContext, which every endpoint reads per request, so the calls after the wait use it too; an abandoned login leaves no stale token for the next authenticate(). The job ids never change, so no job is created twice. Co-Authored-By: Claude Opus 5.5 (1M context) --- src/py/mat3ra/notebooks_utils/auth.py | 14 +++++- .../notebooks_utils/core/entity/job/api.py | 12 ++++- src/py/mat3ra/notebooks_utils/token_store.py | 6 +++ tests/py/unit/core/entity/test_job_api.py | 33 +++++++++++++- tests/py/unit/test_auth_retry.py | 44 +++++++++++++++++++ 5 files changed, 104 insertions(+), 5 deletions(-) create mode 100644 tests/py/unit/test_auth_retry.py diff --git a/src/py/mat3ra/notebooks_utils/auth.py b/src/py/mat3ra/notebooks_utils/auth.py index 433f36490..b43535ac3 100644 --- a/src/py/mat3ra/notebooks_utils/auth.py +++ b/src/py/mat3ra/notebooks_utils/auth.py @@ -13,14 +13,14 @@ import inspect import os -from mat3ra.api_client import ACCESS_TOKEN_ENV_VAR +from mat3ra.api_client import ACCESS_TOKEN_ENV_VAR, AuthContext from .core.api.auth import authenticate_oidc, get_oidc_base_url, store_token_data_in_environment from .io import get_data from .ipython.ui import show_device_flow_popup from .primitive.environment import is_pyodide_environment from .pyodide.api.auth import authenticate_jupyterlite -from .token_store import load_token, save_token +from .token_store import delete_token, load_token, save_token REFRESH_TOKEN_ENV_VAR = "OIDC_REFRESH_TOKEN" @@ -63,3 +63,13 @@ async def authenticate(force=False, globals_dict=None): await authenticate_jupyterlite(data_from_host) elif ACCESS_TOKEN_ENV_VAR not in os.environ or force: await _authenticate_oidc_with_cache(force) + + +async def reauthenticate(auth_context: AuthContext) -> None: + """ + Replaces an access token the platform rejected: drops it from the token cache, runs the device login and sets the + new token on `auth_context`, which every endpoint of the API client reads per request. + """ + await delete_token(get_oidc_base_url()) + await _authenticate_oidc_with_cache(force=True) + auth_context.access_token = os.environ[ACCESS_TOKEN_ENV_VAR] diff --git a/src/py/mat3ra/notebooks_utils/core/entity/job/api.py b/src/py/mat3ra/notebooks_utils/core/entity/job/api.py index 8e4f83297..d5f1eed40 100644 --- a/src/py/mat3ra/notebooks_utils/core/entity/job/api.py +++ b/src/py/mat3ra/notebooks_utils/core/entity/job/api.py @@ -8,6 +8,7 @@ import requests from mat3ra.api_client import APIClient, JobEndpoints +from ....auth import reauthenticate from ....primitive.environment import is_pyodide_environment MATERIALS_SET_ENTITY_CLASS = "Material" @@ -68,7 +69,8 @@ async def get_jobs_statuses_by_ids_async( ) -> List[str]: """ Gets jobs statuses by their IDs without blocking the event loop: through the browser's fetch in pyodide, - in a worker thread otherwise. + in a worker thread otherwise. A rejected access token (401) is replaced through the device login once and the + request repeated. Args: endpoint (JobEndpoints): Job endpoint object from the Exabyte API Client @@ -86,7 +88,13 @@ def request_jobs() -> Awaitable[List[dict]]: return _list_jobs_with_fetch(endpoint, query, projection, abort_signal) return asyncio.get_running_loop().run_in_executor(None, endpoint.list, query, projection) - jobs = await asyncio.wait_for(request_jobs(), timeout) + try: + jobs = await asyncio.wait_for(request_jobs(), timeout) + except requests.HTTPError as error: + if error.response.status_code != 401: + raise + await reauthenticate(endpoint._auth) + jobs = await asyncio.wait_for(request_jobs(), timeout) return [job["status"] for job in jobs] diff --git a/src/py/mat3ra/notebooks_utils/token_store.py b/src/py/mat3ra/notebooks_utils/token_store.py index 73eebd6a8..79a135d5c 100644 --- a/src/py/mat3ra/notebooks_utils/token_store.py +++ b/src/py/mat3ra/notebooks_utils/token_store.py @@ -31,3 +31,9 @@ async def load_token(oidc_url: str) -> Optional[dict]: if not entry or entry.get("expires_at", 0) <= time.time() + _EXPIRY_BUFFER: return None return entry + + +async def delete_token(oidc_url: str) -> None: + cache = await _read() + cache.pop(oidc_url, None) + await _write(cache) diff --git a/tests/py/unit/core/entity/test_job_api.py b/tests/py/unit/core/entity/test_job_api.py index f3e4ff116..dae694db0 100644 --- a/tests/py/unit/core/entity/test_job_api.py +++ b/tests/py/unit/core/entity/test_job_api.py @@ -1,9 +1,11 @@ import asyncio +import contextlib import threading from typing import Any, Dict, List -from unittest.mock import MagicMock +from unittest.mock import AsyncMock, MagicMock import pytest +import requests from mat3ra.notebooks_utils.core.entity.job.api import ( create_job, find_job_for_material, @@ -269,3 +271,32 @@ async def test_get_jobs_statuses_by_ids_async_raises_when_the_request_times_out( with pytest.raises(asyncio.TimeoutError): await get_jobs_statuses_by_ids_async(endpoint, [CREATED_JOB["_id"]], timeout=REQUEST_TIMEOUT_SECONDS) release_request.set() + + +HTTP_ERROR_401 = requests.HTTPError("Error 401.", response=MagicMock(status_code=401)) +HTTP_ERROR_500 = requests.HTTPError("Error 500.", response=MagicMock(status_code=500)) +JOBS_WITH_STATUSES: List[Dict[str, Any]] = [{"status": "active"}, {"status": "finished"}] + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + ("list_results", "expectation", "expected_reauthentications"), + [ + ([HTTP_ERROR_401, JOBS_WITH_STATUSES], contextlib.nullcontext(), 1), + ([HTTP_ERROR_401, HTTP_ERROR_401], pytest.raises(requests.HTTPError), 1), + ([HTTP_ERROR_500], pytest.raises(requests.HTTPError), 0), + ], + ids=["401 once", "401 twice", "500"], +) +async def test_get_jobs_statuses_by_ids_async_reauthenticates_once_on_401( + monkeypatch, list_results, expectation, expected_reauthentications +): + reauthenticate = AsyncMock() + monkeypatch.setattr("mat3ra.notebooks_utils.core.entity.job.api.reauthenticate", reauthenticate) + endpoint = MagicMock() + endpoint.list.side_effect = list_results + + with expectation: + assert await get_jobs_statuses_by_ids_async(endpoint, [CREATED_JOB["_id"]]) == ["active", "finished"] + assert reauthenticate.await_args_list == [((endpoint._auth,),)] * expected_reauthentications + assert endpoint.list.call_count == len(list_results) diff --git a/tests/py/unit/test_auth_retry.py b/tests/py/unit/test_auth_retry.py new file mode 100644 index 000000000..c4c09d686 --- /dev/null +++ b/tests/py/unit/test_auth_retry.py @@ -0,0 +1,44 @@ +import contextlib + +import pytest +from mat3ra.api_client import ACCESS_TOKEN_ENV_VAR, AuthContext +from mat3ra.notebooks_utils import auth, token_store +from mat3ra.notebooks_utils.core.api import token_store as file_token_store + +STALE_TOKEN_DATA = {"access_token": "stale-token", "expires_in": 3600} +NEW_TOKEN_DATA = {"access_token": "new-token", "expires_in": 3600} + + +async def completed_device_login(show_popup): + auth.store_token_data_in_environment(NEW_TOKEN_DATA) + return NEW_TOKEN_DATA + + +async def abandoned_device_login(show_popup): + raise TimeoutError("Timeout waiting for authorization.") + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + ("device_login", "expectation", "expected_access_token", "expected_cached_access_token"), + [ + (completed_device_login, contextlib.nullcontext(), "new-token", "new-token"), + (abandoned_device_login, pytest.raises(TimeoutError), "stale-token", None), + ], + ids=["login completed", "login abandoned"], +) +async def test_reauthenticate( + monkeypatch, tmp_path, device_login, expectation, expected_access_token, expected_cached_access_token +): + monkeypatch.setattr(file_token_store, "_FILE_PATH", str(tmp_path / "oidc_token_cache.json")) + monkeypatch.setattr(auth, "authenticate_oidc", device_login) + monkeypatch.setenv(ACCESS_TOKEN_ENV_VAR, STALE_TOKEN_DATA["access_token"]) + oidc_base_url = auth.get_oidc_base_url() + await token_store.save_token(oidc_base_url, STALE_TOKEN_DATA) + auth_context = AuthContext(access_token=STALE_TOKEN_DATA["access_token"]) + + with expectation: + await auth.reauthenticate(auth_context) + + assert auth_context.access_token == expected_access_token + assert (await file_token_store.read()).get(oidc_base_url, {}).get("access_token") == expected_cached_access_token From 2321863adadbcaf721bf70ea69a3f764aa3eb735 Mon Sep 17 00:00:00 2001 From: VsevolodX Date: Wed, 30 Sep 2026 11:53:53 -0700 Subject: [PATCH 05/21] fix(job-wait-loop): authenticate() checks a cached token before using it and logs in again when the platform rejects it A token loaded from the cache is checked with one authenticated call (APIClient.authenticate(access_token=...).list_accounts(), GET /users/me) before it is put in the environment; on a 401 the cache entry is dropped and the device login runs, so a token that is unexpired by its expires_at but rejected by the platform no longer fails every later call; any other error status raises as before. The cache entry is now dropped in the one login path shared by authenticate(force=True) and reauthenticate, so an abandoned login never leaves the rejected token behind. This covers the stale-token case for synchronous calls, which cannot run the async device login themselves under the deployed Pyodide 0.24.1 (no run_sync, no SharedArrayBuffer); a token that expires during a job wait is handled by the wait loop's own 401 retry. Co-Authored-By: Claude Opus 5.5 (1M context) --- src/py/mat3ra/notebooks_utils/auth.py | 16 ++++++++---- tests/py/unit/test_auth_retry.py | 36 +++++++++++++++++++++++++++ 2 files changed, 47 insertions(+), 5 deletions(-) diff --git a/src/py/mat3ra/notebooks_utils/auth.py b/src/py/mat3ra/notebooks_utils/auth.py index b43535ac3..ba742f865 100644 --- a/src/py/mat3ra/notebooks_utils/auth.py +++ b/src/py/mat3ra/notebooks_utils/auth.py @@ -13,7 +13,8 @@ import inspect import os -from mat3ra.api_client import ACCESS_TOKEN_ENV_VAR, AuthContext +import requests +from mat3ra.api_client import ACCESS_TOKEN_ENV_VAR, APIClient, AuthContext from .core.api.auth import authenticate_oidc, get_oidc_base_url, store_token_data_in_environment from .io import get_data @@ -30,9 +31,15 @@ async def _authenticate_oidc_with_cache(force=False): cached = None if force else await load_token(oidc_url) if cached: - store_token_data_in_environment(cached) - return - + try: + APIClient.authenticate(access_token=cached["access_token"]).list_accounts() + store_token_data_in_environment(cached) + return + except requests.HTTPError as error: + if error.response.status_code != 401: + raise + + await delete_token(oidc_url) token_data = await authenticate_oidc(show_popup=show_device_flow_popup) await save_token(oidc_url, token_data) @@ -70,6 +77,5 @@ async def reauthenticate(auth_context: AuthContext) -> None: Replaces an access token the platform rejected: drops it from the token cache, runs the device login and sets the new token on `auth_context`, which every endpoint of the API client reads per request. """ - await delete_token(get_oidc_base_url()) await _authenticate_oidc_with_cache(force=True) auth_context.access_token = os.environ[ACCESS_TOKEN_ENV_VAR] diff --git a/tests/py/unit/test_auth_retry.py b/tests/py/unit/test_auth_retry.py index c4c09d686..766486fbf 100644 --- a/tests/py/unit/test_auth_retry.py +++ b/tests/py/unit/test_auth_retry.py @@ -1,12 +1,17 @@ import contextlib +import os +from unittest.mock import AsyncMock, MagicMock import pytest +import requests from mat3ra.api_client import ACCESS_TOKEN_ENV_VAR, AuthContext from mat3ra.notebooks_utils import auth, token_store from mat3ra.notebooks_utils.core.api import token_store as file_token_store STALE_TOKEN_DATA = {"access_token": "stale-token", "expires_in": 3600} NEW_TOKEN_DATA = {"access_token": "new-token", "expires_in": 3600} +HTTP_ERROR_401 = requests.HTTPError("Error 401.", response=MagicMock(status_code=401)) +HTTP_ERROR_500 = requests.HTTPError("Error 500.", response=MagicMock(status_code=500)) async def completed_device_login(show_popup): @@ -42,3 +47,34 @@ async def test_reauthenticate( assert auth_context.access_token == expected_access_token assert (await file_token_store.read()).get(oidc_base_url, {}).get("access_token") == expected_cached_access_token + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + ("accounts_side_effect", "expectation", "expected_access_token", "expected_logins"), + [ + (HTTP_ERROR_401, contextlib.nullcontext(), "new-token", 1), + (None, contextlib.nullcontext(), "stale-token", 0), + (HTTP_ERROR_500, pytest.raises(requests.HTTPError), None, 0), + ], + ids=["cached token rejected", "cached token accepted", "platform error"], +) +async def test_authenticate_validates_the_cached_token( + monkeypatch, tmp_path, accounts_side_effect, expectation, expected_access_token, expected_logins +): + monkeypatch.setattr(file_token_store, "_FILE_PATH", str(tmp_path / "oidc_token_cache.json")) + device_login = AsyncMock(side_effect=completed_device_login) + monkeypatch.setattr(auth, "authenticate_oidc", device_login) + api_client = MagicMock() + api_client.authenticate.return_value.list_accounts.side_effect = accounts_side_effect + monkeypatch.setattr(auth, "APIClient", api_client) + monkeypatch.setenv(ACCESS_TOKEN_ENV_VAR, STALE_TOKEN_DATA["access_token"]) + monkeypatch.delenv(ACCESS_TOKEN_ENV_VAR) + await token_store.save_token(auth.get_oidc_base_url(), STALE_TOKEN_DATA) + + with expectation: + await auth.authenticate(globals_dict={}) + + assert os.environ.get(ACCESS_TOKEN_ENV_VAR) == expected_access_token + assert device_login.await_count == expected_logins + api_client.authenticate.assert_called_once_with(access_token=STALE_TOKEN_DATA["access_token"]) From ba85f794825fe51209f9d07ce4b9eda66feb0718 Mon Sep 17 00:00:00 2001 From: VsevolodX Date: Wed, 30 Sep 2026 12:21:41 -0700 Subject: [PATCH 06/21] fix(job-wait-loop): ESC aborts the loops of the notebook in focus; an abort racing a response still stops the loop - ESC finds the panel holding the focus and posts to the channel on each abort button in it (or to the only button on the page); the page-global channel map is gone. - is_aborted is checked after each poll, so an abort that asyncio.wait_for drops in the response's tick stops the loop instead of failing on the spent AbortSignal. - An abandoned device login also clears OIDC_ACCESS_TOKEN, so re-running authenticate() logs in again. - A 401 re-login runs only for bearer-token clients; with X-Auth headers the 401 is raised. - The fetch branch of the status request is tested with a fake pyodide.http; the subprocess uuid test is dropped. - Unused get_jobs_statuses_by_ids removed; DEFAULT_STATUS_TIMEOUT_SECONDS; abort_signal keyword-only. Co-Authored-By: Claude Opus 5.5 (1M context) --- src/py/mat3ra/notebooks_utils/api/job.py | 13 ++-- src/py/mat3ra/notebooks_utils/auth.py | 1 + .../notebooks_utils/core/entity/job/api.py | 30 ++++------ .../mat3ra/notebooks_utils/pyodide/runtime.py | 32 ++++++---- tests/py/unit/core/entity/test_job_api.py | 60 +++++++++++++++++-- tests/py/unit/test_auth_retry.py | 10 ++-- tests/py/unit/test_jupyterlite_interrupts.py | 14 ----- 7 files changed, 95 insertions(+), 65 deletions(-) diff --git a/src/py/mat3ra/notebooks_utils/api/job.py b/src/py/mat3ra/notebooks_utils/api/job.py index 89187b2c6..68cda7eaf 100644 --- a/src/py/mat3ra/notebooks_utils/api/job.py +++ b/src/py/mat3ra/notebooks_utils/api/job.py @@ -5,20 +5,16 @@ from mat3ra.api_client import JobEndpoints from mat3ra.utils.extra.tabulate import pretty_print -from ..core.entity.job.api import ( - create_job, - get_jobs_statuses_by_ids, - get_jobs_statuses_by_ids_async, - save_files, - submit_jobs, -) +from ..core.entity.job.api import create_job, get_jobs_statuses_by_ids_async, save_files, submit_jobs from ..pyodide.runtime import interruptible_polling_loop # Contains no external dependencies, only uses the API client, # so can be used in both regular Python and Pyodide environments. @interruptible_polling_loop() -async def wait_for_jobs_to_finish_async(endpoint: JobEndpoints, job_ids: List[str], abort_signal: Any) -> bool: +async def wait_for_jobs_to_finish_async( + endpoint: JobEndpoints, job_ids: List[str], *, abort_signal: Any = None +) -> bool: """ Waits for jobs to finish and prints their statuses. A job is considered finished if it is not in "pre-submission", "submitted", or "active" status. @@ -47,7 +43,6 @@ async def wait_for_jobs_to_finish_async(endpoint: JobEndpoints, job_ids: List[st __all__ = [ "create_job", - "get_jobs_statuses_by_ids", "save_files", "submit_jobs", "wait_for_jobs_to_finish_async", diff --git a/src/py/mat3ra/notebooks_utils/auth.py b/src/py/mat3ra/notebooks_utils/auth.py index ba742f865..c2e5b0f69 100644 --- a/src/py/mat3ra/notebooks_utils/auth.py +++ b/src/py/mat3ra/notebooks_utils/auth.py @@ -40,6 +40,7 @@ async def _authenticate_oidc_with_cache(force=False): raise await delete_token(oidc_url) + os.environ.pop(ACCESS_TOKEN_ENV_VAR, None) token_data = await authenticate_oidc(show_popup=show_device_flow_popup) await save_token(oidc_url, token_data) diff --git a/src/py/mat3ra/notebooks_utils/core/entity/job/api.py b/src/py/mat3ra/notebooks_utils/core/entity/job/api.py index d5f1eed40..30ab0043d 100644 --- a/src/py/mat3ra/notebooks_utils/core/entity/job/api.py +++ b/src/py/mat3ra/notebooks_utils/core/entity/job/api.py @@ -12,6 +12,7 @@ from ....primitive.environment import is_pyodide_environment MATERIALS_SET_ENTITY_CLASS = "Material" +DEFAULT_STATUS_TIMEOUT_SECONDS = 30 def save_files(job_id: str, job_endpoint: JobEndpoints, filename_on_cloud: str, filename_on_disk: str) -> None: @@ -32,25 +33,10 @@ def save_files(job_id: str, job_endpoint: JobEndpoints, filename_on_cloud: str, outp.write(server_response.read()) -def get_jobs_statuses_by_ids(endpoint: JobEndpoints, job_ids: List[str]) -> List[str]: - """ - Gets jobs statues by their IDs. - - Args: - endpoint (JobEndpoints): Job endpoint object from the Exabyte API Client - job_ids (list): list of job IDs to get the status for - - Returns: - list: list of job statuses - """ - jobs = endpoint.list({"_id": {"$in": job_ids}}, {"fields": {"status": 1}}) - return [job["status"] for job in jobs] - - async def _list_jobs_with_fetch(endpoint: JobEndpoints, query: dict, projection: dict, abort_signal: Any) -> List[dict]: """ `endpoint.list` through the browser's fetch, which leaves the event loop free while the request is in flight. - Raises `requests.HTTPError` on an error status, as the API client does. + Raises `requests.HTTPError` on an error status. """ from pyodide.http import pyfetch # type: ignore @@ -65,12 +51,16 @@ async def _list_jobs_with_fetch(endpoint: JobEndpoints, query: dict, projection: async def get_jobs_statuses_by_ids_async( - endpoint: JobEndpoints, job_ids: List[str], timeout: float = 30, abort_signal: Any = None + endpoint: JobEndpoints, + job_ids: List[str], + timeout: float = DEFAULT_STATUS_TIMEOUT_SECONDS, + abort_signal: Any = None, ) -> List[str]: """ Gets jobs statuses by their IDs without blocking the event loop: through the browser's fetch in pyodide, - in a worker thread otherwise. A rejected access token (401) is replaced through the device login once and the - request repeated. + in a worker thread otherwise. Natively, a request that times out or is cancelled keeps its worker thread until the + API client's own timeout. A rejected access token (401) is replaced through the device login once and the request + repeated. Args: endpoint (JobEndpoints): Job endpoint object from the Exabyte API Client @@ -91,7 +81,7 @@ def request_jobs() -> Awaitable[List[dict]]: try: jobs = await asyncio.wait_for(request_jobs(), timeout) except requests.HTTPError as error: - if error.response.status_code != 401: + if error.response.status_code != 401 or not endpoint._auth.access_token: raise await reauthenticate(endpoint._auth) jobs = await asyncio.wait_for(request_jobs(), timeout) diff --git a/src/py/mat3ra/notebooks_utils/pyodide/runtime.py b/src/py/mat3ra/notebooks_utils/pyodide/runtime.py index 1aaca02fe..826b48f5a 100644 --- a/src/py/mat3ra/notebooks_utils/pyodide/runtime.py +++ b/src/py/mat3ra/notebooks_utils/pyodide/runtime.py @@ -44,6 +44,7 @@ def display_abort_controls_in_current_cell_output(
+ >{abort_button_text}""" - Press ESC to abort + display( + HTML( + f""" +
+ {button_html if show_button else ""} + {abort_hint_text}
@@ -59,18 +63,16 @@ def display_abort_controls_in_current_cell_output( (function() {{ const channelName = {channel_name!r}; - // Install ESC broadcaster once per page; ESC aborts the loops of the notebook in focus, - // or the only loop on the page when the focus is outside any notebook + // Install ESC broadcaster once per page if (!window.__mat3raEscapeAbortByPanelInstalled) {{ window.__mat3raEscapeAbortByPanelInstalled = true; document.addEventListener("keydown", (event) => {{ if (event.key !== "Escape") return; const notebookPanel = document.activeElement?.closest(".jp-NotebookPanel"); - const abortButtons = (notebookPanel || document) - .querySelectorAll("button[data-mat3ra-abort-channel]"); - if (!notebookPanel && abortButtons.length !== 1) return; - abortButtons.forEach((abortButton) => {{ - const escapeChannel = new BroadcastChannel(abortButton.dataset.mat3raAbortChannel); + const abortHints = (notebookPanel || document).querySelectorAll("[data-mat3ra-abort-channel]"); + if (!notebookPanel && abortHints.length !== 1) return; + abortHints.forEach((abortHint) => {{ + const escapeChannel = new BroadcastChannel(abortHint.dataset.mat3raAbortChannel); escapeChannel.postMessage({{ type: "abort", source: "escape" }}); escapeChannel.close(); }}); @@ -157,7 +159,9 @@ async def run_interruptible_loop_async( *, channel_name: str = ABORT_CHANNEL_NAME, show_controls: bool = True, + show_button: bool = True, abort_button_text: str = "Abort", + abort_hint_text: str = "Press ESC to abort", ) -> None: """ Wraps an async loop around a "poll" function that returns True to continue, False to stop. @@ -173,7 +177,12 @@ async def run_interruptible_loop_async( broadcast_channel_abort_controller.start(asyncio.current_task()) # type: ignore if show_controls and ENVIRONMENT == EnvironmentsEnum.PYODIDE: - display_abort_controls_in_current_cell_output(channel_name=channel_name, abort_button_text=abort_button_text) + display_abort_controls_in_current_cell_output( + channel_name=channel_name, + abort_button_text=abort_button_text, + abort_hint_text=abort_hint_text, + show_button=show_button, + ) try: while await loop_body(broadcast_channel_abort_controller.fetch_abort_signal): diff --git a/tests/py/unit/core/entity/test_job_api.py b/tests/py/unit/core/entity/test_job_api.py index e47dd6d44..e0cf1c5eb 100644 --- a/tests/py/unit/core/entity/test_job_api.py +++ b/tests/py/unit/core/entity/test_job_api.py @@ -1,6 +1,5 @@ import asyncio import contextlib -import sys import threading from types import SimpleNamespace from typing import Any, Dict, List @@ -342,7 +341,7 @@ async def test_get_jobs_statuses_by_ids_async_reauthenticates_once_on_401( ) async def test_list_jobs_with_fetch(monkeypatch, access_token, response, expectation, expected_headers): pyfetch = AsyncMock(return_value=response) - monkeypatch.setitem(sys.modules, "pyodide.http", SimpleNamespace(pyfetch=pyfetch)) + monkeypatch.setattr("mat3ra.notebooks_utils.core.entity.job.api.pyfetch", pyfetch) endpoint = JobEndpoints(*JOB_ENDPOINT_ARGUMENTS, auth=AuthContext(access_token=access_token)) with expectation: From 1f82b9362edd4eee9fa1062d94d6ec0969cd1b60 Mon Sep 17 00:00:00 2001 From: VsevolodX Date: Wed, 7 Oct 2026 15:19:06 -0700 Subject: [PATCH 10/21] fix(job-wait-loop): the status check takes headers and auth from the API client's public endpoint API mat3ra-api-client 2026.10.7 removed BaseEndpoint._get_bearer_headers and froze endpoint.headers at construction, so the browser fetch failed with AttributeError on the first status poll and the retry after a re-login sent the expired token. The fetch now sends endpoint.get_request_headers() (the endpoint headers merged with the auth context's current token) and the 401 path reads endpoint.auth; nothing in src/ reaches into api-client privates. The endpoint fake in the 401 test is spec'd on JobEndpoints, so private access fails there too. Needs the api-client change that adds get_request_headers() and auth. Co-Authored-By: Claude Opus 5.5 (1M context) --- src/py/mat3ra/notebooks_utils/core/entity/job/api.py | 6 +++--- tests/py/unit/core/entity/test_job_api.py | 11 ++++++----- 2 files changed, 9 insertions(+), 8 deletions(-) diff --git a/src/py/mat3ra/notebooks_utils/core/entity/job/api.py b/src/py/mat3ra/notebooks_utils/core/entity/job/api.py index 387d62119..c52d5ca4b 100644 --- a/src/py/mat3ra/notebooks_utils/core/entity/job/api.py +++ b/src/py/mat3ra/notebooks_utils/core/entity/job/api.py @@ -45,7 +45,7 @@ async def _list_jobs_with_fetch(endpoint: JobEndpoints, query: dict, projection: """ parameters = urllib.parse.urlencode({"query": json.dumps(query), "projection": json.dumps(projection)}) url = urllib.parse.urljoin(endpoint.conn.preamble, f"{endpoint.name}?{parameters}") - response = await pyfetch(url, headers=endpoint._get_bearer_headers() or endpoint.headers, signal=abort_signal) + response = await pyfetch(url, headers=endpoint.get_request_headers(), signal=abort_signal) if not response.ok: error_response = requests.Response() error_response.status_code = response.status @@ -85,9 +85,9 @@ def request_jobs() -> Awaitable[List[dict]]: try: jobs = await asyncio.wait_for(request_jobs(), timeout) except requests.HTTPError as error: - if error.response.status_code != 401 or not endpoint._auth.access_token: + if error.response.status_code != 401 or not endpoint.auth.access_token: raise - await reauthenticate(endpoint._auth) + await reauthenticate(endpoint.auth) jobs = await asyncio.wait_for(request_jobs(), timeout) return [job["status"] for job in jobs] diff --git a/tests/py/unit/core/entity/test_job_api.py b/tests/py/unit/core/entity/test_job_api.py index e0cf1c5eb..aa028a533 100644 --- a/tests/py/unit/core/entity/test_job_api.py +++ b/tests/py/unit/core/entity/test_job_api.py @@ -298,13 +298,13 @@ async def test_get_jobs_statuses_by_ids_async_reauthenticates_once_on_401( ): reauthenticate = AsyncMock() monkeypatch.setattr("mat3ra.notebooks_utils.core.entity.job.api.reauthenticate", reauthenticate) - endpoint = MagicMock() - endpoint._auth.access_token = access_token + endpoint = MagicMock(spec=JobEndpoints) + endpoint.auth.access_token = access_token endpoint.list.side_effect = list_results with expectation: assert await get_jobs_statuses_by_ids_async(endpoint, [CREATED_JOB["_id"]]) == ["active", "finished"] - assert reauthenticate.await_args_list == [((endpoint._auth,),)] * expected_reauthentications + assert reauthenticate.await_args_list == [((endpoint.auth,),)] * expected_reauthentications assert endpoint.list.call_count == len(list_results) @@ -317,7 +317,7 @@ async def test_get_jobs_statuses_by_ids_async_reauthenticates_once_on_401( "?query=%7B%22_id%22%3A+%7B%22%24in%22%3A+%5B%22job-1%22%5D%7D%7D" "&projection=%7B%22fields%22%3A+%7B%22status%22%3A+1%7D%7D" ) -BEARER_HEADERS = {"Authorization": f"Bearer {ACCESS_TOKEN}"} +BEARER_HEADERS = {"Authorization": f"Bearer {ACCESS_TOKEN}", "Content-Type": "application/json"} X_AUTH_HEADERS = {"X-Account-Id": OWNER_ID, "X-Auth-Token": AUTH_TOKEN, "Content-Type": "application/json"} ABORT_SIGNAL = "abort-signal" FETCH_RESPONSE_OK = SimpleNamespace(ok=True, status=200, json=AsyncMock(return_value={"data": JOBS_WITH_STATUSES})) @@ -342,7 +342,8 @@ async def test_get_jobs_statuses_by_ids_async_reauthenticates_once_on_401( async def test_list_jobs_with_fetch(monkeypatch, access_token, response, expectation, expected_headers): pyfetch = AsyncMock(return_value=response) monkeypatch.setattr("mat3ra.notebooks_utils.core.entity.job.api.pyfetch", pyfetch) - endpoint = JobEndpoints(*JOB_ENDPOINT_ARGUMENTS, auth=AuthContext(access_token=access_token)) + auth_context = AuthContext(access_token=access_token, account_id=OWNER_ID, auth_token=AUTH_TOKEN) + endpoint = JobEndpoints(*JOB_ENDPOINT_ARGUMENTS, auth=auth_context) with expectation: assert await _list_jobs_with_fetch(endpoint, JOBS_QUERY, STATUS_PROJECTION, ABORT_SIGNAL) == JOBS_WITH_STATUSES From 0907bafe1d2a72894b1406b8cd88f07fd7d5eb7e Mon Sep 17 00:00:00 2001 From: VsevolodX Date: Wed, 7 Oct 2026 15:20:26 -0700 Subject: [PATCH 11/21] fix(job-wait-loop): a failed status check prints one line and retries at the next interval A status request that timed out after 30 s, failed at the network or got a 5xx ended the whole job wait, hours in, while the jobs kept running. The wait now prints the time and the error with "retrying" and polls again at the next interval, indefinitely. A 4xx other than 401 still raises (the request itself is wrong); a 401 still re-logs in once. asyncio.TimeoutError is listed next to OSError because on Python 3.10 it is not an OSError. Abort and ESC still win: the retry returns to the polling loop, which raises UserAbortError when an abort arrived, whatever exception the aborted fetch produced. Co-Authored-By: Claude Opus 5.5 (1M context) --- src/py/mat3ra/notebooks_utils/api/job.py | 10 ++++++- tests/py/unit/test_jupyterlite_interrupts.py | 31 +++++++++++++++++++- 2 files changed, 39 insertions(+), 2 deletions(-) diff --git a/src/py/mat3ra/notebooks_utils/api/job.py b/src/py/mat3ra/notebooks_utils/api/job.py index 68cda7eaf..c4852645b 100644 --- a/src/py/mat3ra/notebooks_utils/api/job.py +++ b/src/py/mat3ra/notebooks_utils/api/job.py @@ -1,7 +1,9 @@ +import asyncio import datetime from collections import Counter from typing import Any, List +import requests from mat3ra.api_client import JobEndpoints from mat3ra.utils.extra.tabulate import pretty_print @@ -24,7 +26,13 @@ async def wait_for_jobs_to_finish_async( job_ids (list): list of job IDs to wait for abort_signal: JS AbortSignal of the Abort control, passed in by the polling loop """ - statuses = await get_jobs_statuses_by_ids_async(endpoint, job_ids, abort_signal=abort_signal) + try: + statuses = await get_jobs_statuses_by_ids_async(endpoint, job_ids, abort_signal=abort_signal) + except (asyncio.TimeoutError, OSError) as error: + if isinstance(error, requests.HTTPError) and error.response.status_code < 500: + raise + print(f"{datetime.datetime.now():%Y-%m-%d-%H:%M:%S} status check failed: {error!r}, retrying") + return True counts = Counter(statuses) headers = ["TIME", "SUBMITTED-JOBS", "ACTIVE-JOBS", "FINISHED-JOBS", "ERRORED-JOBS"] now = datetime.datetime.now().strftime("%Y-%m-%d-%H:%M:%S") diff --git a/tests/py/unit/test_jupyterlite_interrupts.py b/tests/py/unit/test_jupyterlite_interrupts.py index 8067058b7..26b241376 100644 --- a/tests/py/unit/test_jupyterlite_interrupts.py +++ b/tests/py/unit/test_jupyterlite_interrupts.py @@ -2,9 +2,10 @@ import contextlib import threading import time -from unittest.mock import MagicMock +from unittest.mock import AsyncMock, MagicMock import pytest +import requests from mat3ra.notebooks_utils.api.job import wait_for_jobs_to_finish_async from mat3ra.notebooks_utils.core.api.auth import _poll_for_token_data from mat3ra.notebooks_utils.pyodide.runtime import ( @@ -76,6 +77,34 @@ def list_jobs(query, projection): assert time.monotonic() - started < ABORT_DEADLINE_SECONDS +HTTP_ERROR_503 = requests.HTTPError("Error 503.", response=MagicMock(status_code=503)) +HTTP_ERROR_403 = requests.HTTPError("Error 403.", response=MagicMock(status_code=403)) +NETWORK_ERRORS = [requests.ConnectionError("Connection refused."), OSError("Failed to fetch")] + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + ("status_results", "expectation", "expected_retries"), + [ + ([asyncio.TimeoutError(), ["finished"]], contextlib.nullcontext(), 1), + ([*NETWORK_ERRORS, ["finished"]], contextlib.nullcontext(), 2), + ([HTTP_ERROR_503, ["finished"]], contextlib.nullcontext(), 1), + ([HTTP_ERROR_403], pytest.raises(requests.HTTPError), 0), + ], + ids=["timeout", "network error", "503", "403"], +) +async def test_wait_for_jobs_to_finish_async_retries_a_failed_status_check( + monkeypatch, capsys, status_results, expectation, expected_retries +): + get_statuses = AsyncMock(side_effect=status_results) + monkeypatch.setattr("mat3ra.notebooks_utils.api.job.get_jobs_statuses_by_ids_async", get_statuses) + + with expectation: + await wait_for_jobs_to_finish_async(MagicMock(), ["job-1"], poll_interval=POLL_INTERVAL_SECONDS) + assert get_statuses.await_count == len(status_results) + assert capsys.readouterr().out.count("retrying") == expected_retries + + TOKEN_DATA = {"access_token": "new-token", "expires_in": 3600} PENDING_TOKEN_RESPONSE = MagicMock(status_code=400, json=MagicMock(return_value={"error": "authorization_pending"})) AUTHORIZED_TOKEN_RESPONSE = MagicMock(status_code=200, json=MagicMock(return_value=TOKEN_DATA)) From 6582f4999ad51447e9a1e2aa46267b8b8670269e Mon Sep 17 00:00:00 2001 From: VsevolodX Date: Wed, 7 Oct 2026 15:21:23 -0700 Subject: [PATCH 12/21] fix(job-wait-loop): a refused device login stops at once with the error code Every non-200 token response counted as "still pending", so a login the user denied, or a device code the provider expired, kept polling silently until the device code's lifetime ran out (up to 10 minutes) and then reported a timeout. Only authorization_pending and slow_down keep the polling going now; any other error (access_denied, expired_token, ...) raises "Device login failed: ." at once, from the native request and the browser fetch alike. The exception is not an OSError, so a re-login inside the job wait that is refused ends the wait instead of being retried. Co-Authored-By: Claude Opus 5.5 (1M context) --- src/py/mat3ra/notebooks_utils/core/api/auth.py | 11 +++++++++-- tests/py/unit/test_jupyterlite_interrupts.py | 11 ++++++++++- 2 files changed, 19 insertions(+), 3 deletions(-) diff --git a/src/py/mat3ra/notebooks_utils/core/api/auth.py b/src/py/mat3ra/notebooks_utils/core/api/auth.py index d983db3a0..df7d991a0 100644 --- a/src/py/mat3ra/notebooks_utils/core/api/auth.py +++ b/src/py/mat3ra/notebooks_utils/core/api/auth.py @@ -61,9 +61,16 @@ def store_token_data_in_environment(token_data: dict) -> None: os.environ[REFRESH_TOKEN_ENV_VAR] = token_data["refresh_token"] +def _get_token_data(status_code: int, response_data: dict) -> dict: + """Token data of a token response, empty while the login is pending; raises when the login was refused.""" + if status_code != 200 and response_data.get("error") not in ("authorization_pending", "slow_down"): + raise Exception(f"Device login failed: {response_data.get('error')}.") + return response_data if status_code == 200 else {} + + def _request_token_data(token_url: str, form_data: dict) -> dict: response = requests.post(token_url, data=form_data, headers=FORM_HEADERS, timeout=TOKEN_REQUEST_TIMEOUT_SECONDS) - return response.json() if response.status_code == 200 else {} + return _get_token_data(response.status_code, response.json()) async def _request_token_data_with_fetch(token_url: str, form_data: dict, abort_signal: Any) -> dict: @@ -72,7 +79,7 @@ async def _request_token_data_with_fetch(token_url: str, form_data: dict, abort_ """ body = urllib.parse.urlencode(form_data, doseq=True) response = await pyfetch(token_url, method="POST", body=body, headers=FORM_HEADERS, signal=abort_signal) - return await response.json() if response.status == 200 else {} + return _get_token_data(response.status, await response.json()) async def _poll_for_token_data( diff --git a/tests/py/unit/test_jupyterlite_interrupts.py b/tests/py/unit/test_jupyterlite_interrupts.py index 26b241376..3df7ae5e7 100644 --- a/tests/py/unit/test_jupyterlite_interrupts.py +++ b/tests/py/unit/test_jupyterlite_interrupts.py @@ -108,6 +108,8 @@ async def test_wait_for_jobs_to_finish_async_retries_a_failed_status_check( TOKEN_DATA = {"access_token": "new-token", "expires_in": 3600} PENDING_TOKEN_RESPONSE = MagicMock(status_code=400, json=MagicMock(return_value={"error": "authorization_pending"})) AUTHORIZED_TOKEN_RESPONSE = MagicMock(status_code=200, json=MagicMock(return_value=TOKEN_DATA)) +SLOW_DOWN_TOKEN_RESPONSE = MagicMock(status_code=400, json=MagicMock(return_value={"error": "slow_down"})) +REFUSED_TOKEN_RESPONSE = MagicMock(status_code=400, json=MagicMock(return_value={"error": "access_denied"})) DEVICE_FLOW_ARGUMENTS = ("https://platform.mat3ra.com/oidc", "client-1", "device-code-1") EXPIRES_IN_SECONDS = 600 @@ -118,8 +120,15 @@ async def test_wait_for_jobs_to_finish_async_retries_a_failed_status_check( [ ([PENDING_TOKEN_RESPONSE, AUTHORIZED_TOKEN_RESPONSE], EXPIRES_IN_SECONDS, contextlib.nullcontext(), TOKEN_DATA), ([PENDING_TOKEN_RESPONSE], POLL_INTERVAL_SECONDS, pytest.raises(Exception, match="Timeout"), None), + ( + [SLOW_DOWN_TOKEN_RESPONSE, AUTHORIZED_TOKEN_RESPONSE], + EXPIRES_IN_SECONDS, + contextlib.nullcontext(), + TOKEN_DATA, + ), + ([REFUSED_TOKEN_RESPONSE], EXPIRES_IN_SECONDS, pytest.raises(Exception, match="access_denied"), None), ], - ids=["authorized on the 2nd poll", "device code expired"], + ids=["authorized on the 2nd poll", "device code expired", "slow down", "login refused"], ) async def test_poll_for_token_data(monkeypatch, token_responses, expires_in_seconds, expectation, expected_token_data): monkeypatch.setattr("requests.post", MagicMock(side_effect=token_responses)) From 27c08c005337b043704cbeaf6febbb18aabeb79e Mon Sep 17 00:00:00 2001 From: VsevolodX Date: Wed, 7 Oct 2026 15:27:59 -0700 Subject: [PATCH 13/21] fix(job-wait-loop): the retry and refused-login tests carry one row per case and no extra column The retry test is named after the function and counts the "retrying" lines from the number of failures; the device-login expiry row now polls slow_down responses, so slow_down staying pending needs no row of its own. Same cases, fewer lines. Co-Authored-By: Claude Opus 5.5 (1M context) --- tests/py/unit/test_jupyterlite_interrupts.py | 27 +++++++------------- 1 file changed, 9 insertions(+), 18 deletions(-) diff --git a/tests/py/unit/test_jupyterlite_interrupts.py b/tests/py/unit/test_jupyterlite_interrupts.py index 3df7ae5e7..83fd7f618 100644 --- a/tests/py/unit/test_jupyterlite_interrupts.py +++ b/tests/py/unit/test_jupyterlite_interrupts.py @@ -79,30 +79,27 @@ def list_jobs(query, projection): HTTP_ERROR_503 = requests.HTTPError("Error 503.", response=MagicMock(status_code=503)) HTTP_ERROR_403 = requests.HTTPError("Error 403.", response=MagicMock(status_code=403)) -NETWORK_ERRORS = [requests.ConnectionError("Connection refused."), OSError("Failed to fetch")] @pytest.mark.asyncio @pytest.mark.parametrize( - ("status_results", "expectation", "expected_retries"), + ("status_results", "expectation"), [ - ([asyncio.TimeoutError(), ["finished"]], contextlib.nullcontext(), 1), - ([*NETWORK_ERRORS, ["finished"]], contextlib.nullcontext(), 2), - ([HTTP_ERROR_503, ["finished"]], contextlib.nullcontext(), 1), - ([HTTP_ERROR_403], pytest.raises(requests.HTTPError), 0), + ([asyncio.TimeoutError(), ["finished"]], contextlib.nullcontext()), + ([requests.ConnectionError(), OSError("Failed to fetch"), ["finished"]], contextlib.nullcontext()), + ([HTTP_ERROR_503, ["finished"]], contextlib.nullcontext()), + ([HTTP_ERROR_403], pytest.raises(requests.HTTPError)), ], ids=["timeout", "network error", "503", "403"], ) -async def test_wait_for_jobs_to_finish_async_retries_a_failed_status_check( - monkeypatch, capsys, status_results, expectation, expected_retries -): +async def test_wait_for_jobs_to_finish_async(monkeypatch, capsys, status_results, expectation): get_statuses = AsyncMock(side_effect=status_results) monkeypatch.setattr("mat3ra.notebooks_utils.api.job.get_jobs_statuses_by_ids_async", get_statuses) with expectation: await wait_for_jobs_to_finish_async(MagicMock(), ["job-1"], poll_interval=POLL_INTERVAL_SECONDS) assert get_statuses.await_count == len(status_results) - assert capsys.readouterr().out.count("retrying") == expected_retries + assert capsys.readouterr().out.count("retrying") == len(status_results) - 1 TOKEN_DATA = {"access_token": "new-token", "expires_in": 3600} @@ -119,16 +116,10 @@ async def test_wait_for_jobs_to_finish_async_retries_a_failed_status_check( ("token_responses", "expires_in_seconds", "expectation", "expected_token_data"), [ ([PENDING_TOKEN_RESPONSE, AUTHORIZED_TOKEN_RESPONSE], EXPIRES_IN_SECONDS, contextlib.nullcontext(), TOKEN_DATA), - ([PENDING_TOKEN_RESPONSE], POLL_INTERVAL_SECONDS, pytest.raises(Exception, match="Timeout"), None), - ( - [SLOW_DOWN_TOKEN_RESPONSE, AUTHORIZED_TOKEN_RESPONSE], - EXPIRES_IN_SECONDS, - contextlib.nullcontext(), - TOKEN_DATA, - ), + ([SLOW_DOWN_TOKEN_RESPONSE], POLL_INTERVAL_SECONDS, pytest.raises(Exception, match="Timeout"), None), ([REFUSED_TOKEN_RESPONSE], EXPIRES_IN_SECONDS, pytest.raises(Exception, match="access_denied"), None), ], - ids=["authorized on the 2nd poll", "device code expired", "slow down", "login refused"], + ids=["authorized on the 2nd poll", "device code expired while slowed down", "login refused"], ) async def test_poll_for_token_data(monkeypatch, token_responses, expires_in_seconds, expectation, expected_token_data): monkeypatch.setattr("requests.post", MagicMock(side_effect=token_responses)) From c98adfc7de4a5527a3b76d475766969449797896 Mon Sep 17 00:00:00 2001 From: VsevolodX Date: Wed, 7 Oct 2026 15:31:20 -0700 Subject: [PATCH 14/21] fix(job-wait-loop): a network failure while the status response body is read is retried too Pyodide's pyfetch turns a failed request into OSError, but reading the body afterwards (response.json()) lets a raw JsException through, which the job wait's retry does not catch, so a connection dropped mid-body ended the wait. _list_jobs_with_fetch now converts it to OSError with the browser's message, as pyfetch itself does. An abort during the read takes the same path and still ends the wait through the polling loop's abort state. Co-Authored-By: Claude Opus 5.5 (1M context) --- src/py/mat3ra/notebooks_utils/core/entity/job/api.py | 7 ++++++- tests/py/unit/core/entity/test_job_api.py | 11 ++++++++++- 2 files changed, 16 insertions(+), 2 deletions(-) diff --git a/src/py/mat3ra/notebooks_utils/core/entity/job/api.py b/src/py/mat3ra/notebooks_utils/core/entity/job/api.py index c52d5ca4b..949169567 100644 --- a/src/py/mat3ra/notebooks_utils/core/entity/job/api.py +++ b/src/py/mat3ra/notebooks_utils/core/entity/job/api.py @@ -12,8 +12,10 @@ from ....primitive.environment import is_pyodide_environment try: + from pyodide.ffi import JsException # type: ignore from pyodide.http import pyfetch # type: ignore except ImportError: + JsException = None pyfetch = None MATERIALS_SET_ENTITY_CLASS = "Material" @@ -50,7 +52,10 @@ async def _list_jobs_with_fetch(endpoint: JobEndpoints, query: dict, projection: error_response = requests.Response() error_response.status_code = response.status raise requests.HTTPError(f"Error {response.status}.", response=error_response) - return (await response.json())["data"] + try: + return (await response.json())["data"] + except JsException as error: + raise OSError(error.message) from None async def get_jobs_statuses_by_ids_async( diff --git a/tests/py/unit/core/entity/test_job_api.py b/tests/py/unit/core/entity/test_job_api.py index aa028a533..807bdf713 100644 --- a/tests/py/unit/core/entity/test_job_api.py +++ b/tests/py/unit/core/entity/test_job_api.py @@ -324,6 +324,13 @@ async def test_get_jobs_statuses_by_ids_async_reauthenticates_once_on_401( FETCH_RESPONSE_401 = SimpleNamespace(ok=False, status=401) +class FakeJsException(Exception): + message = "TypeError: network error" + + +FETCH_RESPONSE_BODY_FAILED = SimpleNamespace(ok=True, status=200, json=AsyncMock(side_effect=FakeJsException())) + + @pytest.mark.asyncio @pytest.mark.parametrize( ("access_token", "response", "expectation", "expected_headers"), @@ -336,12 +343,14 @@ async def test_get_jobs_statuses_by_ids_async_reauthenticates_once_on_401( pytest.raises(requests.HTTPError, check=lambda error: error.response.status_code == 401), BEARER_HEADERS, ), + (ACCESS_TOKEN, FETCH_RESPONSE_BODY_FAILED, pytest.raises(OSError, match="network error"), BEARER_HEADERS), ], - ids=["bearer token", "X-Auth headers", "401"], + ids=["bearer token", "X-Auth headers", "401", "network error while reading the body"], ) async def test_list_jobs_with_fetch(monkeypatch, access_token, response, expectation, expected_headers): pyfetch = AsyncMock(return_value=response) monkeypatch.setattr("mat3ra.notebooks_utils.core.entity.job.api.pyfetch", pyfetch) + monkeypatch.setattr("mat3ra.notebooks_utils.core.entity.job.api.JsException", FakeJsException) auth_context = AuthContext(access_token=access_token, account_id=OWNER_ID, auth_token=AUTH_TOKEN) endpoint = JobEndpoints(*JOB_ENDPOINT_ARGUMENTS, auth=auth_context) From 2af535e595df42e638fcb3120a1033c826329553 Mon Sep 17 00:00:00 2001 From: VsevolodX Date: Wed, 7 Oct 2026 15:48:21 -0700 Subject: [PATCH 15/21] fix(job-wait-loop): a server error from the token endpoint keeps the device login polling Since the refused-login fix both token paths parsed every non-200 body, so a 5xx during the device login ended it: an HTML error page from the ingress raised JSONDecodeError, a JSON body without "error" raised "Device login failed: None.", and the approval the user had already given was lost. At 3fce4da2 such a response was simply polled again. A 5xx is polled again now without reading its body, the same rule as the job status check; any other non-200 that is not authorization_pending / slow_down raises with the HTTP status and the error code. The pending codes are a module constant. Co-Authored-By: Claude Opus 5.5 (1M context) --- src/py/mat3ra/notebooks_utils/core/api/auth.py | 18 ++++++++++++------ tests/py/unit/test_jupyterlite_interrupts.py | 4 +++- 2 files changed, 15 insertions(+), 7 deletions(-) diff --git a/src/py/mat3ra/notebooks_utils/core/api/auth.py b/src/py/mat3ra/notebooks_utils/core/api/auth.py index df7d991a0..a19e1ba60 100644 --- a/src/py/mat3ra/notebooks_utils/core/api/auth.py +++ b/src/py/mat3ra/notebooks_utils/core/api/auth.py @@ -18,6 +18,7 @@ REFRESH_TOKEN_ENV_VAR = "OIDC_REFRESH_TOKEN" TOKEN_REQUEST_TIMEOUT_SECONDS = 10 FORM_HEADERS = {"Content-Type": "application/x-www-form-urlencoded"} +PENDING_LOGIN_ERRORS = ("authorization_pending", "slow_down") def get_oidc_base_url() -> str: @@ -62,15 +63,20 @@ def store_token_data_in_environment(token_data: dict) -> None: def _get_token_data(status_code: int, response_data: dict) -> dict: - """Token data of a token response, empty while the login is pending; raises when the login was refused.""" - if status_code != 200 and response_data.get("error") not in ("authorization_pending", "slow_down"): - raise Exception(f"Device login failed: {response_data.get('error')}.") - return response_data if status_code == 200 else {} + """ + Token data of a token response, empty while the login is pending or after a server error (5xx), which is polled + again; raises with the status and the error code when the login was refused. + """ + if status_code == 200: + return response_data + if status_code >= 500 or response_data.get("error") in PENDING_LOGIN_ERRORS: + return {} + raise Exception(f"Device login failed ({status_code}): {response_data.get('error')}.") def _request_token_data(token_url: str, form_data: dict) -> dict: response = requests.post(token_url, data=form_data, headers=FORM_HEADERS, timeout=TOKEN_REQUEST_TIMEOUT_SECONDS) - return _get_token_data(response.status_code, response.json()) + return _get_token_data(response.status_code, response.json() if response.status_code < 500 else {}) async def _request_token_data_with_fetch(token_url: str, form_data: dict, abort_signal: Any) -> dict: @@ -79,7 +85,7 @@ async def _request_token_data_with_fetch(token_url: str, form_data: dict, abort_ """ body = urllib.parse.urlencode(form_data, doseq=True) response = await pyfetch(token_url, method="POST", body=body, headers=FORM_HEADERS, signal=abort_signal) - return _get_token_data(response.status, await response.json()) + return _get_token_data(response.status, await response.json() if response.status < 500 else {}) async def _poll_for_token_data( diff --git a/tests/py/unit/test_jupyterlite_interrupts.py b/tests/py/unit/test_jupyterlite_interrupts.py index 83fd7f618..3e293ef66 100644 --- a/tests/py/unit/test_jupyterlite_interrupts.py +++ b/tests/py/unit/test_jupyterlite_interrupts.py @@ -107,6 +107,7 @@ async def test_wait_for_jobs_to_finish_async(monkeypatch, capsys, status_results AUTHORIZED_TOKEN_RESPONSE = MagicMock(status_code=200, json=MagicMock(return_value=TOKEN_DATA)) SLOW_DOWN_TOKEN_RESPONSE = MagicMock(status_code=400, json=MagicMock(return_value={"error": "slow_down"})) REFUSED_TOKEN_RESPONSE = MagicMock(status_code=400, json=MagicMock(return_value={"error": "access_denied"})) +SERVER_ERROR_RESPONSE = MagicMock(status_code=502, json=MagicMock(side_effect=ValueError("not JSON"))) DEVICE_FLOW_ARGUMENTS = ("https://platform.mat3ra.com/oidc", "client-1", "device-code-1") EXPIRES_IN_SECONDS = 600 @@ -118,8 +119,9 @@ async def test_wait_for_jobs_to_finish_async(monkeypatch, capsys, status_results ([PENDING_TOKEN_RESPONSE, AUTHORIZED_TOKEN_RESPONSE], EXPIRES_IN_SECONDS, contextlib.nullcontext(), TOKEN_DATA), ([SLOW_DOWN_TOKEN_RESPONSE], POLL_INTERVAL_SECONDS, pytest.raises(Exception, match="Timeout"), None), ([REFUSED_TOKEN_RESPONSE], EXPIRES_IN_SECONDS, pytest.raises(Exception, match="access_denied"), None), + ([SERVER_ERROR_RESPONSE, AUTHORIZED_TOKEN_RESPONSE], EXPIRES_IN_SECONDS, contextlib.nullcontext(), TOKEN_DATA), ], - ids=["authorized on the 2nd poll", "device code expired while slowed down", "login refused"], + ids=["authorized on the 2nd poll", "slowed down until expiry", "login refused", "authorized after a 502"], ) async def test_poll_for_token_data(monkeypatch, token_responses, expires_in_seconds, expectation, expected_token_data): monkeypatch.setattr("requests.post", MagicMock(side_effect=token_responses)) From 8dfdd64f4e46f0f43df5071b05e70931924d864f Mon Sep 17 00:00:00 2001 From: VsevolodX Date: Wed, 7 Oct 2026 15:49:34 -0700 Subject: [PATCH 16/21] fix(job-wait-loop): tests pin that an abort beats the retry and that a replaced token reaches the fetch Two properties the retry and the public api-client API rely on had no test: removing the polling loop's abort check, or sending the endpoint's frozen headers again, left the suite green. One test aborts the loop while the status check raises the OSError an aborted fetch produces in Pyodide and expects UserAbortError with no second poll. The fetch rows now build the endpoint with a stale token and replace it on the shared auth context before the request, as the re-login does, so frozen headers fail them. Co-Authored-By: Claude Opus 5.5 (1M context) --- tests/py/unit/core/entity/test_job_api.py | 4 +++- tests/py/unit/test_jupyterlite_interrupts.py | 12 ++++++++++++ 2 files changed, 15 insertions(+), 1 deletion(-) diff --git a/tests/py/unit/core/entity/test_job_api.py b/tests/py/unit/core/entity/test_job_api.py index 807bdf713..e304bc7e0 100644 --- a/tests/py/unit/core/entity/test_job_api.py +++ b/tests/py/unit/core/entity/test_job_api.py @@ -351,8 +351,10 @@ async def test_list_jobs_with_fetch(monkeypatch, access_token, response, expecta pyfetch = AsyncMock(return_value=response) monkeypatch.setattr("mat3ra.notebooks_utils.core.entity.job.api.pyfetch", pyfetch) monkeypatch.setattr("mat3ra.notebooks_utils.core.entity.job.api.JsException", FakeJsException) - auth_context = AuthContext(access_token=access_token, account_id=OWNER_ID, auth_token=AUTH_TOKEN) + stale_access_token = "stale-access-token" if access_token else None + auth_context = AuthContext(access_token=stale_access_token, account_id=OWNER_ID, auth_token=AUTH_TOKEN) endpoint = JobEndpoints(*JOB_ENDPOINT_ARGUMENTS, auth=auth_context) + auth_context.access_token = access_token with expectation: assert await _list_jobs_with_fetch(endpoint, JOBS_QUERY, STATUS_PROJECTION, ABORT_SIGNAL) == JOBS_WITH_STATUSES diff --git a/tests/py/unit/test_jupyterlite_interrupts.py b/tests/py/unit/test_jupyterlite_interrupts.py index 3e293ef66..ed322178b 100644 --- a/tests/py/unit/test_jupyterlite_interrupts.py +++ b/tests/py/unit/test_jupyterlite_interrupts.py @@ -9,6 +9,7 @@ from mat3ra.notebooks_utils.api.job import wait_for_jobs_to_finish_async from mat3ra.notebooks_utils.core.api.auth import _poll_for_token_data from mat3ra.notebooks_utils.pyodide.runtime import ( + BroadcastChannelAbortController, UserAbortError, interruptible_polling_loop, run_interruptible_loop_async, @@ -102,6 +103,17 @@ async def test_wait_for_jobs_to_finish_async(monkeypatch, capsys, status_results assert capsys.readouterr().out.count("retrying") == len(status_results) - 1 +@pytest.mark.asyncio +async def test_wait_for_jobs_to_finish_async_raises_user_abort_error_after_an_aborted_fetch(monkeypatch): + monkeypatch.setattr(BroadcastChannelAbortController, "start", lambda self, task: setattr(self, "is_aborted", True)) + get_statuses = AsyncMock(side_effect=[OSError("The user aborted a request.")]) + monkeypatch.setattr("mat3ra.notebooks_utils.api.job.get_jobs_statuses_by_ids_async", get_statuses) + + with pytest.raises(UserAbortError): + await wait_for_jobs_to_finish_async(MagicMock(), ["job-1"], poll_interval=10.0) + assert get_statuses.await_count == 1 + + TOKEN_DATA = {"access_token": "new-token", "expires_in": 3600} PENDING_TOKEN_RESPONSE = MagicMock(status_code=400, json=MagicMock(return_value={"error": "authorization_pending"})) AUTHORIZED_TOKEN_RESPONSE = MagicMock(status_code=200, json=MagicMock(return_value=TOKEN_DATA)) From 633e99e2075e91aec4ef05cfe55bc0489fbef128 Mon Sep 17 00:00:00 2001 From: VsevolodX Date: Wed, 7 Oct 2026 15:49:59 -0700 Subject: [PATCH 17/21] fix(job-wait-loop): the docstrings state the retry rule; the timestamp is formatted once wait_for_jobs_to_finish_async says which failures are retried (timeout, network, 5xx) and why a 4xx raises; _list_jobs_with_fetch says it raises OSError on a network failure, during the body read too. The wait computes its timestamp once and uses it for the retry line and the table row. Co-Authored-By: Claude Opus 5.5 (1M context) --- src/py/mat3ra/notebooks_utils/api/job.py | 6 ++++-- src/py/mat3ra/notebooks_utils/core/entity/job/api.py | 2 +- 2 files changed, 5 insertions(+), 3 deletions(-) diff --git a/src/py/mat3ra/notebooks_utils/api/job.py b/src/py/mat3ra/notebooks_utils/api/job.py index c4852645b..4a03a793d 100644 --- a/src/py/mat3ra/notebooks_utils/api/job.py +++ b/src/py/mat3ra/notebooks_utils/api/job.py @@ -20,22 +20,24 @@ async def wait_for_jobs_to_finish_async( """ Waits for jobs to finish and prints their statuses. A job is considered finished if it is not in "pre-submission", "submitted", or "active" status. + A status check that times out, fails at the network or gets a 5xx is retried at the next interval; a 4xx raises, + because the request itself is wrong (a 401 only after one new login). Args: endpoint (JobEndpoints): Job endpoint object from the Exabyte API Client job_ids (list): list of job IDs to wait for abort_signal: JS AbortSignal of the Abort control, passed in by the polling loop """ + now = datetime.datetime.now().strftime("%Y-%m-%d-%H:%M:%S") try: statuses = await get_jobs_statuses_by_ids_async(endpoint, job_ids, abort_signal=abort_signal) except (asyncio.TimeoutError, OSError) as error: if isinstance(error, requests.HTTPError) and error.response.status_code < 500: raise - print(f"{datetime.datetime.now():%Y-%m-%d-%H:%M:%S} status check failed: {error!r}, retrying") + print(f"{now} status check failed: {error!r}, retrying") return True counts = Counter(statuses) headers = ["TIME", "SUBMITTED-JOBS", "ACTIVE-JOBS", "FINISHED-JOBS", "ERRORED-JOBS"] - now = datetime.datetime.now().strftime("%Y-%m-%d-%H:%M:%S") row = [ now, counts.get("submitted", 0) + counts.get("queued", 0), diff --git a/src/py/mat3ra/notebooks_utils/core/entity/job/api.py b/src/py/mat3ra/notebooks_utils/core/entity/job/api.py index 949169567..61e4ed13b 100644 --- a/src/py/mat3ra/notebooks_utils/core/entity/job/api.py +++ b/src/py/mat3ra/notebooks_utils/core/entity/job/api.py @@ -43,7 +43,7 @@ def save_files(job_id: str, job_endpoint: JobEndpoints, filename_on_cloud: str, async def _list_jobs_with_fetch(endpoint: JobEndpoints, query: dict, projection: dict, abort_signal: Any) -> List[dict]: """ `endpoint.list` through the browser's fetch, which leaves the event loop free while the request is in flight. - Raises `requests.HTTPError` on an error status. + Raises `requests.HTTPError` on an error status and `OSError` when the network fails, while the body is read too. """ parameters = urllib.parse.urlencode({"query": json.dumps(query), "projection": json.dumps(projection)}) url = urllib.parse.urljoin(endpoint.conn.preamble, f"{endpoint.name}?{parameters}") From 5ca0f2e0b6713178c05c40cf2acbe45359339aff Mon Sep 17 00:00:00 2001 From: VsevolodX Date: Wed, 7 Oct 2026 19:33:01 -0700 Subject: [PATCH 18/21] fix(job-wait-loop): keep only what the retry, the device login and the current token need The fetch sends the endpoint headers merged with its auth context's current headers, and the 401 path reads endpoint.auth. A status check that times out, fails at the network or gets a 5xx prints one line and polls again; a 4xx raises. The token poll keeps going on a 5xx (body not read) or a pending/slow_down error, and raises with the status and the error code otherwise; that decision moves into the poll step, so no helper or constant. The body-read JsException conversion, the docstring additions and the pin tests are removed; one parametrized test per behaviour remains. Co-Authored-By: Claude Opus 5.5 (1M context) --- src/py/mat3ra/notebooks_utils/api/job.py | 6 +-- .../mat3ra/notebooks_utils/core/api/auth.py | 28 ++++------- .../notebooks_utils/core/entity/job/api.py | 11 ++--- tests/py/unit/core/entity/test_job_api.py | 17 ++----- tests/py/unit/test_jupyterlite_interrupts.py | 49 ++++++------------- 5 files changed, 31 insertions(+), 80 deletions(-) diff --git a/src/py/mat3ra/notebooks_utils/api/job.py b/src/py/mat3ra/notebooks_utils/api/job.py index 4a03a793d..ad8d96d73 100644 --- a/src/py/mat3ra/notebooks_utils/api/job.py +++ b/src/py/mat3ra/notebooks_utils/api/job.py @@ -20,24 +20,22 @@ async def wait_for_jobs_to_finish_async( """ Waits for jobs to finish and prints their statuses. A job is considered finished if it is not in "pre-submission", "submitted", or "active" status. - A status check that times out, fails at the network or gets a 5xx is retried at the next interval; a 4xx raises, - because the request itself is wrong (a 401 only after one new login). Args: endpoint (JobEndpoints): Job endpoint object from the Exabyte API Client job_ids (list): list of job IDs to wait for abort_signal: JS AbortSignal of the Abort control, passed in by the polling loop """ - now = datetime.datetime.now().strftime("%Y-%m-%d-%H:%M:%S") try: statuses = await get_jobs_statuses_by_ids_async(endpoint, job_ids, abort_signal=abort_signal) except (asyncio.TimeoutError, OSError) as error: if isinstance(error, requests.HTTPError) and error.response.status_code < 500: raise - print(f"{now} status check failed: {error!r}, retrying") + print(f"status check failed: {error!r}, retrying") return True counts = Counter(statuses) headers = ["TIME", "SUBMITTED-JOBS", "ACTIVE-JOBS", "FINISHED-JOBS", "ERRORED-JOBS"] + now = datetime.datetime.now().strftime("%Y-%m-%d-%H:%M:%S") row = [ now, counts.get("submitted", 0) + counts.get("queued", 0), diff --git a/src/py/mat3ra/notebooks_utils/core/api/auth.py b/src/py/mat3ra/notebooks_utils/core/api/auth.py index a19e1ba60..1b88b5e21 100644 --- a/src/py/mat3ra/notebooks_utils/core/api/auth.py +++ b/src/py/mat3ra/notebooks_utils/core/api/auth.py @@ -18,7 +18,6 @@ REFRESH_TOKEN_ENV_VAR = "OIDC_REFRESH_TOKEN" TOKEN_REQUEST_TIMEOUT_SECONDS = 10 FORM_HEADERS = {"Content-Type": "application/x-www-form-urlencoded"} -PENDING_LOGIN_ERRORS = ("authorization_pending", "slow_down") def get_oidc_base_url() -> str: @@ -62,30 +61,18 @@ def store_token_data_in_environment(token_data: dict) -> None: os.environ[REFRESH_TOKEN_ENV_VAR] = token_data["refresh_token"] -def _get_token_data(status_code: int, response_data: dict) -> dict: - """ - Token data of a token response, empty while the login is pending or after a server error (5xx), which is polled - again; raises with the status and the error code when the login was refused. - """ - if status_code == 200: - return response_data - if status_code >= 500 or response_data.get("error") in PENDING_LOGIN_ERRORS: - return {} - raise Exception(f"Device login failed ({status_code}): {response_data.get('error')}.") - - -def _request_token_data(token_url: str, form_data: dict) -> dict: +def _request_token_data(token_url: str, form_data: dict) -> tuple: response = requests.post(token_url, data=form_data, headers=FORM_HEADERS, timeout=TOKEN_REQUEST_TIMEOUT_SECONDS) - return _get_token_data(response.status_code, response.json() if response.status_code < 500 else {}) + return response.status_code, response.json() if response.status_code < 500 else {} -async def _request_token_data_with_fetch(token_url: str, form_data: dict, abort_signal: Any) -> dict: +async def _request_token_data_with_fetch(token_url: str, form_data: dict, abort_signal: Any) -> tuple: """ `_request_token_data` through the browser's fetch, which leaves the event loop free while the request is in flight. """ body = urllib.parse.urlencode(form_data, doseq=True) response = await pyfetch(token_url, method="POST", body=body, headers=FORM_HEADERS, signal=abort_signal) - return _get_token_data(response.status, await response.json() if response.status < 500 else {}) + return response.status, await response.json() if response.status < 500 else {} async def _poll_for_token_data( @@ -108,7 +95,7 @@ async def _poll_for_token_data( deadline_seconds = time.time() + expires_in_seconds token_data: dict = {} - def request_token_data(abort_signal: Any) -> Awaitable[dict]: + def request_token_data(abort_signal: Any) -> Awaitable[tuple]: if is_pyodide_environment(): return _request_token_data_with_fetch(token_url, form_data, abort_signal) return asyncio.get_running_loop().run_in_executor(None, _request_token_data, token_url, form_data) @@ -116,7 +103,10 @@ def request_token_data(abort_signal: Any) -> Awaitable[dict]: async def poll_step(abort_signal: Any) -> bool: if time.time() >= deadline_seconds: raise Exception("Timeout waiting for authorization.") - token_data.update(await asyncio.wait_for(request_token_data(abort_signal), TOKEN_REQUEST_TIMEOUT_SECONDS)) + status, response_data = await asyncio.wait_for(request_token_data(abort_signal), TOKEN_REQUEST_TIMEOUT_SECONDS) + if status != 200 and status < 500 and response_data.get("error") not in ("authorization_pending", "slow_down"): + raise Exception(f"Device login failed ({status}): {response_data.get('error')}.") + token_data.update(response_data if status == 200 else {}) return not token_data await run_interruptible_loop_async( diff --git a/src/py/mat3ra/notebooks_utils/core/entity/job/api.py b/src/py/mat3ra/notebooks_utils/core/entity/job/api.py index 61e4ed13b..7f616e068 100644 --- a/src/py/mat3ra/notebooks_utils/core/entity/job/api.py +++ b/src/py/mat3ra/notebooks_utils/core/entity/job/api.py @@ -12,10 +12,8 @@ from ....primitive.environment import is_pyodide_environment try: - from pyodide.ffi import JsException # type: ignore from pyodide.http import pyfetch # type: ignore except ImportError: - JsException = None pyfetch = None MATERIALS_SET_ENTITY_CLASS = "Material" @@ -43,19 +41,16 @@ def save_files(job_id: str, job_endpoint: JobEndpoints, filename_on_cloud: str, async def _list_jobs_with_fetch(endpoint: JobEndpoints, query: dict, projection: dict, abort_signal: Any) -> List[dict]: """ `endpoint.list` through the browser's fetch, which leaves the event loop free while the request is in flight. - Raises `requests.HTTPError` on an error status and `OSError` when the network fails, while the body is read too. + Raises `requests.HTTPError` on an error status. """ parameters = urllib.parse.urlencode({"query": json.dumps(query), "projection": json.dumps(projection)}) url = urllib.parse.urljoin(endpoint.conn.preamble, f"{endpoint.name}?{parameters}") - response = await pyfetch(url, headers=endpoint.get_request_headers(), signal=abort_signal) + response = await pyfetch(url, headers={**endpoint.headers, **endpoint.auth.get_headers()}, signal=abort_signal) if not response.ok: error_response = requests.Response() error_response.status_code = response.status raise requests.HTTPError(f"Error {response.status}.", response=error_response) - try: - return (await response.json())["data"] - except JsException as error: - raise OSError(error.message) from None + return (await response.json())["data"] async def get_jobs_statuses_by_ids_async( diff --git a/tests/py/unit/core/entity/test_job_api.py b/tests/py/unit/core/entity/test_job_api.py index e304bc7e0..a194ce9bf 100644 --- a/tests/py/unit/core/entity/test_job_api.py +++ b/tests/py/unit/core/entity/test_job_api.py @@ -298,7 +298,7 @@ async def test_get_jobs_statuses_by_ids_async_reauthenticates_once_on_401( ): reauthenticate = AsyncMock() monkeypatch.setattr("mat3ra.notebooks_utils.core.entity.job.api.reauthenticate", reauthenticate) - endpoint = MagicMock(spec=JobEndpoints) + endpoint = MagicMock() endpoint.auth.access_token = access_token endpoint.list.side_effect = list_results @@ -324,13 +324,6 @@ async def test_get_jobs_statuses_by_ids_async_reauthenticates_once_on_401( FETCH_RESPONSE_401 = SimpleNamespace(ok=False, status=401) -class FakeJsException(Exception): - message = "TypeError: network error" - - -FETCH_RESPONSE_BODY_FAILED = SimpleNamespace(ok=True, status=200, json=AsyncMock(side_effect=FakeJsException())) - - @pytest.mark.asyncio @pytest.mark.parametrize( ("access_token", "response", "expectation", "expected_headers"), @@ -343,18 +336,14 @@ class FakeJsException(Exception): pytest.raises(requests.HTTPError, check=lambda error: error.response.status_code == 401), BEARER_HEADERS, ), - (ACCESS_TOKEN, FETCH_RESPONSE_BODY_FAILED, pytest.raises(OSError, match="network error"), BEARER_HEADERS), ], - ids=["bearer token", "X-Auth headers", "401", "network error while reading the body"], + ids=["bearer token", "X-Auth headers", "401"], ) async def test_list_jobs_with_fetch(monkeypatch, access_token, response, expectation, expected_headers): pyfetch = AsyncMock(return_value=response) monkeypatch.setattr("mat3ra.notebooks_utils.core.entity.job.api.pyfetch", pyfetch) - monkeypatch.setattr("mat3ra.notebooks_utils.core.entity.job.api.JsException", FakeJsException) - stale_access_token = "stale-access-token" if access_token else None - auth_context = AuthContext(access_token=stale_access_token, account_id=OWNER_ID, auth_token=AUTH_TOKEN) + auth_context = AuthContext(access_token=access_token, account_id=OWNER_ID, auth_token=AUTH_TOKEN) endpoint = JobEndpoints(*JOB_ENDPOINT_ARGUMENTS, auth=auth_context) - auth_context.access_token = access_token with expectation: assert await _list_jobs_with_fetch(endpoint, JOBS_QUERY, STATUS_PROJECTION, ABORT_SIGNAL) == JOBS_WITH_STATUSES diff --git a/tests/py/unit/test_jupyterlite_interrupts.py b/tests/py/unit/test_jupyterlite_interrupts.py index ed322178b..fa394eba7 100644 --- a/tests/py/unit/test_jupyterlite_interrupts.py +++ b/tests/py/unit/test_jupyterlite_interrupts.py @@ -2,14 +2,13 @@ import contextlib import threading import time -from unittest.mock import AsyncMock, MagicMock +from unittest.mock import MagicMock import pytest import requests from mat3ra.notebooks_utils.api.job import wait_for_jobs_to_finish_async from mat3ra.notebooks_utils.core.api.auth import _poll_for_token_data from mat3ra.notebooks_utils.pyodide.runtime import ( - BroadcastChannelAbortController, UserAbortError, interruptible_polling_loop, run_interruptible_loop_async, @@ -78,48 +77,28 @@ def list_jobs(query, projection): assert time.monotonic() - started < ABORT_DEADLINE_SECONDS -HTTP_ERROR_503 = requests.HTTPError("Error 503.", response=MagicMock(status_code=503)) -HTTP_ERROR_403 = requests.HTTPError("Error 403.", response=MagicMock(status_code=403)) - - @pytest.mark.asyncio @pytest.mark.parametrize( - ("status_results", "expectation"), + ("status_error", "expectation"), [ - ([asyncio.TimeoutError(), ["finished"]], contextlib.nullcontext()), - ([requests.ConnectionError(), OSError("Failed to fetch"), ["finished"]], contextlib.nullcontext()), - ([HTTP_ERROR_503, ["finished"]], contextlib.nullcontext()), - ([HTTP_ERROR_403], pytest.raises(requests.HTTPError)), + (asyncio.TimeoutError(), contextlib.nullcontext()), + (OSError("Failed to fetch"), contextlib.nullcontext()), + (requests.HTTPError(response=MagicMock(status_code=503)), contextlib.nullcontext()), + (requests.HTTPError(response=MagicMock(status_code=403)), pytest.raises(requests.HTTPError)), ], - ids=["timeout", "network error", "503", "403"], + ids=["timeout", "network", "503", "403"], ) -async def test_wait_for_jobs_to_finish_async(monkeypatch, capsys, status_results, expectation): - get_statuses = AsyncMock(side_effect=status_results) - monkeypatch.setattr("mat3ra.notebooks_utils.api.job.get_jobs_statuses_by_ids_async", get_statuses) - +async def test_wait_for_jobs_to_finish_async(status_error, expectation): + endpoint = MagicMock(list=MagicMock(side_effect=[status_error, [{"status": "finished"}]])) with expectation: - await wait_for_jobs_to_finish_async(MagicMock(), ["job-1"], poll_interval=POLL_INTERVAL_SECONDS) - assert get_statuses.await_count == len(status_results) - assert capsys.readouterr().out.count("retrying") == len(status_results) - 1 - - -@pytest.mark.asyncio -async def test_wait_for_jobs_to_finish_async_raises_user_abort_error_after_an_aborted_fetch(monkeypatch): - monkeypatch.setattr(BroadcastChannelAbortController, "start", lambda self, task: setattr(self, "is_aborted", True)) - get_statuses = AsyncMock(side_effect=[OSError("The user aborted a request.")]) - monkeypatch.setattr("mat3ra.notebooks_utils.api.job.get_jobs_statuses_by_ids_async", get_statuses) - - with pytest.raises(UserAbortError): - await wait_for_jobs_to_finish_async(MagicMock(), ["job-1"], poll_interval=10.0) - assert get_statuses.await_count == 1 + await wait_for_jobs_to_finish_async(endpoint, ["job-1"], poll_interval=POLL_INTERVAL_SECONDS) TOKEN_DATA = {"access_token": "new-token", "expires_in": 3600} PENDING_TOKEN_RESPONSE = MagicMock(status_code=400, json=MagicMock(return_value={"error": "authorization_pending"})) AUTHORIZED_TOKEN_RESPONSE = MagicMock(status_code=200, json=MagicMock(return_value=TOKEN_DATA)) -SLOW_DOWN_TOKEN_RESPONSE = MagicMock(status_code=400, json=MagicMock(return_value={"error": "slow_down"})) +SERVER_ERROR_RESPONSE = MagicMock(status_code=502, json=MagicMock(side_effect=ValueError)) REFUSED_TOKEN_RESPONSE = MagicMock(status_code=400, json=MagicMock(return_value={"error": "access_denied"})) -SERVER_ERROR_RESPONSE = MagicMock(status_code=502, json=MagicMock(side_effect=ValueError("not JSON"))) DEVICE_FLOW_ARGUMENTS = ("https://platform.mat3ra.com/oidc", "client-1", "device-code-1") EXPIRES_IN_SECONDS = 600 @@ -129,11 +108,11 @@ async def test_wait_for_jobs_to_finish_async_raises_user_abort_error_after_an_ab ("token_responses", "expires_in_seconds", "expectation", "expected_token_data"), [ ([PENDING_TOKEN_RESPONSE, AUTHORIZED_TOKEN_RESPONSE], EXPIRES_IN_SECONDS, contextlib.nullcontext(), TOKEN_DATA), - ([SLOW_DOWN_TOKEN_RESPONSE], POLL_INTERVAL_SECONDS, pytest.raises(Exception, match="Timeout"), None), - ([REFUSED_TOKEN_RESPONSE], EXPIRES_IN_SECONDS, pytest.raises(Exception, match="access_denied"), None), + ([PENDING_TOKEN_RESPONSE], POLL_INTERVAL_SECONDS, pytest.raises(Exception, match="Timeout"), None), ([SERVER_ERROR_RESPONSE, AUTHORIZED_TOKEN_RESPONSE], EXPIRES_IN_SECONDS, contextlib.nullcontext(), TOKEN_DATA), + ([REFUSED_TOKEN_RESPONSE], EXPIRES_IN_SECONDS, pytest.raises(Exception, match=r"\(400\): access_denied"), None), ], - ids=["authorized on the 2nd poll", "slowed down until expiry", "login refused", "authorized after a 502"], + ids=["authorized on the 2nd poll", "device code expired", "authorized after a 502", "login refused"], ) async def test_poll_for_token_data(monkeypatch, token_responses, expires_in_seconds, expectation, expected_token_data): monkeypatch.setattr("requests.post", MagicMock(side_effect=token_responses)) From d8e90f491bbd3ccc61260cf3581406776e7cdba9 Mon Sep 17 00:00:00 2001 From: VsevolodX Date: Wed, 7 Oct 2026 20:07:56 -0700 Subject: [PATCH 19/21] test(job-wait-loop): one parametrized test per behaviour - Abort and ESC raise UserAbortError while the status request or the device-login token request is in flight. - The status request goes through the browser fetch (URL, current bearer headers, abort signal) and is bounded by the timeout. - During the wait: a timeout or a 5xx retries, a 4xx raises, a 401 re-logs in once and retries once. - authenticate() re-logs in when the cached or environment token is rejected, and keeps a valid one. - The device login keeps polling on pending and 5xx, and raises when it is refused. Pin tests, per-line rows and duplicate rows are removed. Co-Authored-By: Claude Opus 5.5 (1M context) --- tests/py/unit/core/entity/test_job_api.py | 88 ++------------- tests/py/unit/test_auth_retry.py | 95 ++++++---------- tests/py/unit/test_jupyterlite_interrupts.py | 109 ++++--------------- 3 files changed, 62 insertions(+), 230 deletions(-) diff --git a/tests/py/unit/core/entity/test_job_api.py b/tests/py/unit/core/entity/test_job_api.py index a194ce9bf..fc3c575e9 100644 --- a/tests/py/unit/core/entity/test_job_api.py +++ b/tests/py/unit/core/entity/test_job_api.py @@ -1,15 +1,10 @@ import asyncio -import contextlib -import threading -from types import SimpleNamespace from typing import Any, Dict, List -from unittest.mock import AsyncMock, MagicMock +from unittest.mock import MagicMock import pytest -import requests from mat3ra.api_client import AuthContext, JobEndpoints from mat3ra.notebooks_utils.core.entity.job.api import ( - _list_jobs_with_fetch, create_job, find_job_for_material, find_job_for_material_with_property, @@ -263,88 +258,23 @@ def test_get_kgrid_of_job(unit, expected_kgrid): REQUEST_TIMEOUT_SECONDS = 0.05 BLOCKED_REQUEST_SECONDS = 1.0 - - -@pytest.mark.asyncio -async def test_get_jobs_statuses_by_ids_async_raises_when_the_request_times_out(): - release_request = threading.Event() - endpoint = MagicMock() - endpoint.list.side_effect = lambda query, projection: release_request.wait(BLOCKED_REQUEST_SECONDS) - - with pytest.raises(asyncio.TimeoutError): - await get_jobs_statuses_by_ids_async(endpoint, [CREATED_JOB["_id"]], timeout=REQUEST_TIMEOUT_SECONDS) - release_request.set() - - -HTTP_ERROR_401 = requests.HTTPError("Error 401.", response=MagicMock(status_code=401)) -HTTP_ERROR_500 = requests.HTTPError("Error 500.", response=MagicMock(status_code=500)) -JOBS_WITH_STATUSES: List[Dict[str, Any]] = [{"status": "active"}, {"status": "finished"}] ACCESS_TOKEN = "access-token-1" - - -@pytest.mark.asyncio -@pytest.mark.parametrize( - ("access_token", "list_results", "expectation", "expected_reauthentications"), - [ - (ACCESS_TOKEN, [HTTP_ERROR_401, JOBS_WITH_STATUSES], contextlib.nullcontext(), 1), - (ACCESS_TOKEN, [HTTP_ERROR_401, HTTP_ERROR_401], pytest.raises(requests.HTTPError), 1), - (ACCESS_TOKEN, [HTTP_ERROR_500], pytest.raises(requests.HTTPError), 0), - (None, [HTTP_ERROR_401], pytest.raises(requests.HTTPError), 0), - ], - ids=["401 once", "401 twice", "500", "401 with X-Auth headers"], -) -async def test_get_jobs_statuses_by_ids_async_reauthenticates_once_on_401( - monkeypatch, access_token, list_results, expectation, expected_reauthentications -): - reauthenticate = AsyncMock() - monkeypatch.setattr("mat3ra.notebooks_utils.core.entity.job.api.reauthenticate", reauthenticate) - endpoint = MagicMock() - endpoint.auth.access_token = access_token - endpoint.list.side_effect = list_results - - with expectation: - assert await get_jobs_statuses_by_ids_async(endpoint, [CREATED_JOB["_id"]]) == ["active", "finished"] - assert reauthenticate.await_args_list == [((endpoint.auth,),)] * expected_reauthentications - assert endpoint.list.call_count == len(list_results) - - -AUTH_TOKEN = "auth-token-1" -JOB_ENDPOINT_ARGUMENTS = ("platform.mat3ra.com", 443, OWNER_ID, AUTH_TOKEN, "2018-10-01", True) -JOBS_QUERY: Dict[str, Any] = {"_id": {"$in": [CREATED_JOB["_id"]]}} -STATUS_PROJECTION: Dict[str, Any] = {"fields": {"status": 1}} +ABORT_SIGNAL = "abort-signal" JOBS_FETCH_URL = ( "https://platform.mat3ra.com:443/api/2018-10-01/jobs" "?query=%7B%22_id%22%3A+%7B%22%24in%22%3A+%5B%22job-1%22%5D%7D%7D" "&projection=%7B%22fields%22%3A+%7B%22status%22%3A+1%7D%7D" ) BEARER_HEADERS = {"Authorization": f"Bearer {ACCESS_TOKEN}", "Content-Type": "application/json"} -X_AUTH_HEADERS = {"X-Account-Id": OWNER_ID, "X-Auth-Token": AUTH_TOKEN, "Content-Type": "application/json"} -ABORT_SIGNAL = "abort-signal" -FETCH_RESPONSE_OK = SimpleNamespace(ok=True, status=200, json=AsyncMock(return_value={"data": JOBS_WITH_STATUSES})) -FETCH_RESPONSE_401 = SimpleNamespace(ok=False, status=401) @pytest.mark.asyncio -@pytest.mark.parametrize( - ("access_token", "response", "expectation", "expected_headers"), - [ - (ACCESS_TOKEN, FETCH_RESPONSE_OK, contextlib.nullcontext(), BEARER_HEADERS), - (None, FETCH_RESPONSE_OK, contextlib.nullcontext(), X_AUTH_HEADERS), - ( - ACCESS_TOKEN, - FETCH_RESPONSE_401, - pytest.raises(requests.HTTPError, check=lambda error: error.response.status_code == 401), - BEARER_HEADERS, - ), - ], - ids=["bearer token", "X-Auth headers", "401"], -) -async def test_list_jobs_with_fetch(monkeypatch, access_token, response, expectation, expected_headers): - pyfetch = AsyncMock(return_value=response) +async def test_get_jobs_statuses_by_ids_async(monkeypatch): + pyfetch = MagicMock(side_effect=lambda *args, **kwargs: asyncio.sleep(BLOCKED_REQUEST_SECONDS)) monkeypatch.setattr("mat3ra.notebooks_utils.core.entity.job.api.pyfetch", pyfetch) - auth_context = AuthContext(access_token=access_token, account_id=OWNER_ID, auth_token=AUTH_TOKEN) - endpoint = JobEndpoints(*JOB_ENDPOINT_ARGUMENTS, auth=auth_context) + monkeypatch.setattr("mat3ra.notebooks_utils.core.entity.job.api.is_pyodide_environment", lambda: True) + endpoint = JobEndpoints("platform.mat3ra.com", 443, OWNER_ID, None, auth=AuthContext(access_token=ACCESS_TOKEN)) - with expectation: - assert await _list_jobs_with_fetch(endpoint, JOBS_QUERY, STATUS_PROJECTION, ABORT_SIGNAL) == JOBS_WITH_STATUSES - pyfetch.assert_awaited_once_with(JOBS_FETCH_URL, headers=expected_headers, signal=ABORT_SIGNAL) + with pytest.raises(asyncio.TimeoutError): + await get_jobs_statuses_by_ids_async(endpoint, [CREATED_JOB["_id"]], REQUEST_TIMEOUT_SECONDS, ABORT_SIGNAL) + pyfetch.assert_called_once_with(JOBS_FETCH_URL, headers=BEARER_HEADERS, signal=ABORT_SIGNAL) diff --git a/tests/py/unit/test_auth_retry.py b/tests/py/unit/test_auth_retry.py index 6ec6b9229..cd45fcf0d 100644 --- a/tests/py/unit/test_auth_retry.py +++ b/tests/py/unit/test_auth_retry.py @@ -1,18 +1,22 @@ +import asyncio import contextlib import os -from unittest.mock import AsyncMock, MagicMock +from unittest.mock import MagicMock import pytest import requests from mat3ra.api_client import ACCESS_TOKEN_ENV_VAR, AuthContext from mat3ra.notebooks_utils import auth, token_store +from mat3ra.notebooks_utils.api.job import wait_for_jobs_to_finish_async from mat3ra.notebooks_utils.core.api import token_store as file_token_store -STALE_TOKEN_DATA = {"access_token": "stale-token", "expires_in": 3600} +POLL_INTERVAL_SECONDS = 0.01 +CACHED_TOKEN_DATA = {"access_token": "cached-token", "expires_in": 3600} NEW_TOKEN_DATA = {"access_token": "new-token", "expires_in": 3600} -HTTP_ERROR_401 = requests.HTTPError("Error 401.", response=MagicMock(status_code=401)) -HTTP_ERROR_500 = requests.HTTPError("Error 500.", response=MagicMock(status_code=500)) -MISSING_DATA_ERROR = KeyError("data") +FINISHED_JOBS = [{"status": "finished"}] +HTTP_ERROR_401 = requests.HTTPError(response=MagicMock(status_code=401)) +HTTP_ERROR_403 = requests.HTTPError(response=MagicMock(status_code=403)) +HTTP_ERROR_503 = requests.HTTPError(response=MagicMock(status_code=503)) async def completed_device_login(show_popup): @@ -20,81 +24,48 @@ async def completed_device_login(show_popup): return NEW_TOKEN_DATA -async def abandoned_device_login(show_popup): - raise TimeoutError("Timeout waiting for authorization.") - - -async def store_token_in_cache(monkeypatch): - await token_store.save_token(auth.get_oidc_base_url(), STALE_TOKEN_DATA) - - -async def store_token_in_environment(monkeypatch): - monkeypatch.setenv(ACCESS_TOKEN_ENV_VAR, STALE_TOKEN_DATA["access_token"]) - - @pytest.mark.asyncio @pytest.mark.parametrize( - ("device_login", "expectation", "expected_access_token", "expected_stored_access_token"), + ("status_results", "expectation", "expected_access_token"), [ - (completed_device_login, contextlib.nullcontext(), "new-token", "new-token"), - (abandoned_device_login, pytest.raises(TimeoutError), "stale-token", None), + ([asyncio.TimeoutError(), FINISHED_JOBS], contextlib.nullcontext(), "cached-token"), + ([HTTP_ERROR_503, FINISHED_JOBS], contextlib.nullcontext(), "cached-token"), + ([HTTP_ERROR_403, FINISHED_JOBS], pytest.raises(requests.HTTPError), "cached-token"), + ([HTTP_ERROR_401, FINISHED_JOBS], contextlib.nullcontext(), "new-token"), + ([HTTP_ERROR_401, HTTP_ERROR_401, FINISHED_JOBS], pytest.raises(requests.HTTPError), "new-token"), ], - ids=["login completed", "login abandoned"], ) -async def test_reauthenticate( - monkeypatch, tmp_path, device_login, expectation, expected_access_token, expected_stored_access_token -): +async def test_wait_for_jobs_to_finish_async(monkeypatch, tmp_path, status_results, expectation, expected_access_token): monkeypatch.setattr(file_token_store, "_FILE_PATH", str(tmp_path / "oidc_token_cache.json")) - monkeypatch.setattr(auth, "authenticate_oidc", device_login) - monkeypatch.setattr(auth, "APIClient", MagicMock()) - monkeypatch.setenv(ACCESS_TOKEN_ENV_VAR, STALE_TOKEN_DATA["access_token"]) - oidc_base_url = auth.get_oidc_base_url() - await token_store.save_token(oidc_base_url, STALE_TOKEN_DATA) - auth_context = AuthContext(access_token=STALE_TOKEN_DATA["access_token"]) + monkeypatch.setattr(auth, "authenticate_oidc", completed_device_login) + monkeypatch.setenv(ACCESS_TOKEN_ENV_VAR, CACHED_TOKEN_DATA["access_token"]) + auth_context = AuthContext(access_token=CACHED_TOKEN_DATA["access_token"]) + endpoint = MagicMock(auth=auth_context, list=MagicMock(side_effect=status_results)) with expectation: - await auth.reauthenticate(auth_context) - + await wait_for_jobs_to_finish_async(endpoint, ["job-1"], poll_interval=POLL_INTERVAL_SECONDS) assert auth_context.access_token == expected_access_token - assert os.environ.get(ACCESS_TOKEN_ENV_VAR) == expected_stored_access_token - assert (await file_token_store.read()).get(oidc_base_url, {}).get("access_token") == expected_stored_access_token @pytest.mark.asyncio @pytest.mark.parametrize( - ("store_token", "accounts_side_effect", "expectation", "expected_access_token", "expected_logins"), + ("environment_token", "token_check_error", "checked_token", "expected_token"), [ - (store_token_in_cache, HTTP_ERROR_401, contextlib.nullcontext(), "new-token", 1), - (store_token_in_cache, MISSING_DATA_ERROR, contextlib.nullcontext(), "new-token", 1), - (store_token_in_environment, HTTP_ERROR_401, contextlib.nullcontext(), "new-token", 1), - (store_token_in_cache, None, contextlib.nullcontext(), "stale-token", 0), - (store_token_in_environment, None, contextlib.nullcontext(), "stale-token", 0), - (store_token_in_cache, HTTP_ERROR_500, pytest.raises(requests.HTTPError), None, 0), - ], - ids=[ - "cached token rejected", - "cached token rejected in a 200 response body", - "environment token rejected", - "cached token accepted", - "environment token accepted", - "platform error", + ("", HTTP_ERROR_401, "cached-token", "new-token"), + ("environment-token", KeyError("data"), "environment-token", "new-token"), + ("", None, "cached-token", "cached-token"), ], ) -async def test_authenticate( - monkeypatch, tmp_path, store_token, accounts_side_effect, expectation, expected_access_token, expected_logins -): +async def test_authenticate(monkeypatch, tmp_path, environment_token, token_check_error, checked_token, expected_token): monkeypatch.setattr(file_token_store, "_FILE_PATH", str(tmp_path / "oidc_token_cache.json")) - device_login = AsyncMock(side_effect=completed_device_login) - monkeypatch.setattr(auth, "authenticate_oidc", device_login) + monkeypatch.setattr(auth, "authenticate_oidc", completed_device_login) api_client = MagicMock() - api_client.authenticate.return_value.list_accounts.side_effect = accounts_side_effect + api_client.authenticate.return_value.list_accounts.side_effect = token_check_error monkeypatch.setattr(auth, "APIClient", api_client) - monkeypatch.delenv(ACCESS_TOKEN_ENV_VAR, raising=False) - await store_token(monkeypatch) + monkeypatch.setenv(ACCESS_TOKEN_ENV_VAR, environment_token) + await token_store.save_token(auth.get_oidc_base_url(), CACHED_TOKEN_DATA) - with expectation: - await auth.authenticate(globals_dict={}) + await auth.authenticate(globals_dict={}) - assert os.environ.get(ACCESS_TOKEN_ENV_VAR) == expected_access_token - assert device_login.await_count == expected_logins - api_client.authenticate.assert_called_once_with(access_token=STALE_TOKEN_DATA["access_token"]) + assert os.environ[ACCESS_TOKEN_ENV_VAR] == expected_token + api_client.authenticate.assert_called_once_with(access_token=checked_token) diff --git a/tests/py/unit/test_jupyterlite_interrupts.py b/tests/py/unit/test_jupyterlite_interrupts.py index fa394eba7..8f56ef64c 100644 --- a/tests/py/unit/test_jupyterlite_interrupts.py +++ b/tests/py/unit/test_jupyterlite_interrupts.py @@ -1,11 +1,9 @@ import asyncio import contextlib import threading -import time from unittest.mock import MagicMock import pytest -import requests from mat3ra.notebooks_utils.api.job import wait_for_jobs_to_finish_async from mat3ra.notebooks_utils.core.api.auth import _poll_for_token_data from mat3ra.notebooks_utils.pyodide.runtime import ( @@ -16,8 +14,13 @@ POLL_INTERVAL_SECONDS = 0.01 ABORT_AFTER_SECONDS = 0.05 -ABORT_DEADLINE_SECONDS = 0.2 BLOCKED_REQUEST_SECONDS = 1.0 +TOKEN_DATA = {"access_token": "new-token", "expires_in": 3600} +PENDING_TOKEN_RESPONSE = MagicMock(status_code=400, json=MagicMock(return_value={"error": "authorization_pending"})) +AUTHORIZED_TOKEN_RESPONSE = MagicMock(status_code=200, json=MagicMock(return_value=TOKEN_DATA)) +SERVER_ERROR_RESPONSE = MagicMock(status_code=502, json=MagicMock(side_effect=ValueError)) +REFUSED_TOKEN_RESPONSE = MagicMock(status_code=400, json=MagicMock(return_value={"error": "access_denied"})) +DEVICE_FLOW_ARGUMENTS = ("https://platform.mat3ra.com/oidc", "client-1", "device-code-1", POLL_INTERVAL_SECONDS, 600) @pytest.mark.asyncio @@ -39,111 +42,39 @@ async def loop_body(abort_signal): @pytest.mark.asyncio @pytest.mark.parametrize( - ("poll_seconds", "poll_interval_seconds"), - [(10.0, POLL_INTERVAL_SECONDS), (0.0, 10.0)], - ids=["during a poll", "during the sleep"], + "start_polling", + [ + lambda blocked_request: wait_for_jobs_to_finish_async(MagicMock(list=blocked_request), ["job-1"]), + lambda blocked_request: _poll_for_token_data(*DEVICE_FLOW_ARGUMENTS), + ], ) -async def test_run_interruptible_loop_async_raises_user_abort_error_when_cancelled(poll_seconds, poll_interval_seconds): - async def loop_body(abort_signal): - await asyncio.sleep(poll_seconds) - return True - - task = asyncio.create_task(run_interruptible_loop_async(loop_body, poll_interval_seconds, show_controls=False)) - await asyncio.sleep(ABORT_AFTER_SECONDS) - started = time.monotonic() - task.cancel() - with pytest.raises(UserAbortError): - await task - assert time.monotonic() - started < ABORT_DEADLINE_SECONDS - - -@pytest.mark.asyncio -async def test_wait_for_jobs_to_finish_async_raises_user_abort_error_while_the_status_request_blocks(): +async def test_abort_raises_user_abort_error_while_a_request_is_in_flight(monkeypatch, start_polling): release_request = threading.Event() - def list_jobs(query, projection): + def blocked_request(*args, **kwargs): release_request.wait(BLOCKED_REQUEST_SECONDS) - return [{"status": "finished"}] - endpoint = MagicMock() - endpoint.list.side_effect = list_jobs - started = time.monotonic() - task = asyncio.create_task(wait_for_jobs_to_finish_async(endpoint, ["job-1"], poll_interval=POLL_INTERVAL_SECONDS)) + monkeypatch.setattr("requests.post", blocked_request) + task = asyncio.create_task(start_polling(blocked_request)) await asyncio.sleep(ABORT_AFTER_SECONDS) task.cancel() with pytest.raises(UserAbortError): await task release_request.set() - assert time.monotonic() - started < ABORT_DEADLINE_SECONDS @pytest.mark.asyncio @pytest.mark.parametrize( - ("status_error", "expectation"), + ("token_responses", "expectation"), [ - (asyncio.TimeoutError(), contextlib.nullcontext()), - (OSError("Failed to fetch"), contextlib.nullcontext()), - (requests.HTTPError(response=MagicMock(status_code=503)), contextlib.nullcontext()), - (requests.HTTPError(response=MagicMock(status_code=403)), pytest.raises(requests.HTTPError)), + ([PENDING_TOKEN_RESPONSE, SERVER_ERROR_RESPONSE, AUTHORIZED_TOKEN_RESPONSE], contextlib.nullcontext()), + ([REFUSED_TOKEN_RESPONSE], pytest.raises(Exception, match=r"\(400\): access_denied")), ], - ids=["timeout", "network", "503", "403"], ) -async def test_wait_for_jobs_to_finish_async(status_error, expectation): - endpoint = MagicMock(list=MagicMock(side_effect=[status_error, [{"status": "finished"}]])) - with expectation: - await wait_for_jobs_to_finish_async(endpoint, ["job-1"], poll_interval=POLL_INTERVAL_SECONDS) - - -TOKEN_DATA = {"access_token": "new-token", "expires_in": 3600} -PENDING_TOKEN_RESPONSE = MagicMock(status_code=400, json=MagicMock(return_value={"error": "authorization_pending"})) -AUTHORIZED_TOKEN_RESPONSE = MagicMock(status_code=200, json=MagicMock(return_value=TOKEN_DATA)) -SERVER_ERROR_RESPONSE = MagicMock(status_code=502, json=MagicMock(side_effect=ValueError)) -REFUSED_TOKEN_RESPONSE = MagicMock(status_code=400, json=MagicMock(return_value={"error": "access_denied"})) -DEVICE_FLOW_ARGUMENTS = ("https://platform.mat3ra.com/oidc", "client-1", "device-code-1") -EXPIRES_IN_SECONDS = 600 - - -@pytest.mark.asyncio -@pytest.mark.parametrize( - ("token_responses", "expires_in_seconds", "expectation", "expected_token_data"), - [ - ([PENDING_TOKEN_RESPONSE, AUTHORIZED_TOKEN_RESPONSE], EXPIRES_IN_SECONDS, contextlib.nullcontext(), TOKEN_DATA), - ([PENDING_TOKEN_RESPONSE], POLL_INTERVAL_SECONDS, pytest.raises(Exception, match="Timeout"), None), - ([SERVER_ERROR_RESPONSE, AUTHORIZED_TOKEN_RESPONSE], EXPIRES_IN_SECONDS, contextlib.nullcontext(), TOKEN_DATA), - ([REFUSED_TOKEN_RESPONSE], EXPIRES_IN_SECONDS, pytest.raises(Exception, match=r"\(400\): access_denied"), None), - ], - ids=["authorized on the 2nd poll", "device code expired", "authorized after a 502", "login refused"], -) -async def test_poll_for_token_data(monkeypatch, token_responses, expires_in_seconds, expectation, expected_token_data): +async def test_poll_for_token_data(monkeypatch, token_responses, expectation): monkeypatch.setattr("requests.post", MagicMock(side_effect=token_responses)) - with expectation: - token_data = await _poll_for_token_data(*DEVICE_FLOW_ARGUMENTS, POLL_INTERVAL_SECONDS, expires_in_seconds) - assert token_data == expected_token_data - - -@pytest.mark.asyncio -@pytest.mark.parametrize( - ("request_seconds", "poll_interval_seconds", "response"), - [(BLOCKED_REQUEST_SECONDS, POLL_INTERVAL_SECONDS, AUTHORIZED_TOKEN_RESPONSE), (0.0, 10.0, PENDING_TOKEN_RESPONSE)], - ids=["during the token request", "during the sleep"], -) -async def test_poll_for_token_data_raises_user_abort(monkeypatch, request_seconds, poll_interval_seconds, response): - release_request = threading.Event() - - def post_token_request(*args, **kwargs): - release_request.wait(request_seconds) - return response - - monkeypatch.setattr("requests.post", post_token_request) - started = time.monotonic() - task = asyncio.create_task(_poll_for_token_data(*DEVICE_FLOW_ARGUMENTS, poll_interval_seconds, EXPIRES_IN_SECONDS)) - await asyncio.sleep(ABORT_AFTER_SECONDS) - task.cancel() - with pytest.raises(UserAbortError): - await task - release_request.set() - assert time.monotonic() - started < ABORT_DEADLINE_SECONDS + assert await _poll_for_token_data(*DEVICE_FLOW_ARGUMENTS) == TOKEN_DATA @pytest.mark.asyncio From 9a45cce357c37eef05f694196befb3224b298629 Mon Sep 17 00:00:00 2001 From: VsevolodX Date: Wed, 7 Oct 2026 20:08:05 -0700 Subject: [PATCH 20/21] fix(job-wait-loop): drop the unused abort_button_text parameter and the docstring lines about internals - run_interruptible_loop_async no longer takes abort_button_text; no caller passed it, so the button keeps the "Abort" text it had on main. - Docstrings no longer explain the worker thread's lifetime, the abort receiver's message handling, or how the re-login reaches the endpoints. No behaviour change. Co-Authored-By: Claude Opus 5.5 (1M context) --- src/py/mat3ra/notebooks_utils/auth.py | 3 +-- src/py/mat3ra/notebooks_utils/core/entity/job/api.py | 4 +--- src/py/mat3ra/notebooks_utils/pyodide/runtime.py | 5 +---- 3 files changed, 3 insertions(+), 9 deletions(-) diff --git a/src/py/mat3ra/notebooks_utils/auth.py b/src/py/mat3ra/notebooks_utils/auth.py index 965fa8d0b..ec5fd1f9e 100644 --- a/src/py/mat3ra/notebooks_utils/auth.py +++ b/src/py/mat3ra/notebooks_utils/auth.py @@ -78,8 +78,7 @@ async def authenticate(force=False, globals_dict=None): async def reauthenticate(auth_context: AuthContext) -> None: """ - Replaces an access token the platform rejected: drops it from the token cache, runs the device login and sets the - new token on `auth_context`, which every endpoint of the API client reads per request. + Replaces an access token the platform rejected with a new one from the device login, set on `auth_context`. """ await _authenticate_oidc_with_cache(force=True) auth_context.access_token = os.environ[ACCESS_TOKEN_ENV_VAR] diff --git a/src/py/mat3ra/notebooks_utils/core/entity/job/api.py b/src/py/mat3ra/notebooks_utils/core/entity/job/api.py index 7f616e068..ca8b7440f 100644 --- a/src/py/mat3ra/notebooks_utils/core/entity/job/api.py +++ b/src/py/mat3ra/notebooks_utils/core/entity/job/api.py @@ -61,9 +61,7 @@ async def get_jobs_statuses_by_ids_async( ) -> List[str]: """ Gets jobs statuses by their IDs without blocking the event loop: through the browser's fetch in pyodide, - in a worker thread otherwise. Natively, a request that times out or is cancelled keeps its worker thread until the - API client's own timeout. A rejected access token (401) is replaced through the device login once and the request - repeated. + in a worker thread otherwise. A rejected access token (401) is replaced through the device login once. Args: endpoint (JobEndpoints): Job endpoint object from the Exabyte API Client diff --git a/src/py/mat3ra/notebooks_utils/pyodide/runtime.py b/src/py/mat3ra/notebooks_utils/pyodide/runtime.py index 1053c9535..2e60a8906 100644 --- a/src/py/mat3ra/notebooks_utils/pyodide/runtime.py +++ b/src/py/mat3ra/notebooks_utils/pyodide/runtime.py @@ -102,8 +102,6 @@ def display_abort_controls_in_current_cell_output( class BroadcastChannelAbortController(BaseModel): """ WebWorker-side receiver. Works only in pyodide (emscripten). - An abort message sets `is_aborted`, cancels the task running the loop and aborts the fetch given - `fetch_abort_signal`. In regular Python: start() does nothing, `is_aborted` stays False and `fetch_abort_signal` stays None. """ @@ -160,7 +158,6 @@ async def run_interruptible_loop_async( channel_name: str = ABORT_CHANNEL_NAME, show_controls: bool = True, show_button: bool = True, - abort_button_text: str = "Abort", abort_hint_text: str = "Press ESC to abort", ) -> None: """ @@ -179,7 +176,7 @@ async def run_interruptible_loop_async( if show_controls and ENVIRONMENT == EnvironmentsEnum.PYODIDE: display_abort_controls_in_current_cell_output( channel_name=channel_name, - abort_button_text=abort_button_text, + abort_button_text="Abort", abort_hint_text=abort_hint_text, show_button=show_button, ) From 894e58eb4700b8f65e5218b102df224d21a8428e Mon Sep 17 00:00:00 2001 From: VsevolodX Date: Wed, 7 Oct 2026 20:08:41 -0700 Subject: [PATCH 21/21] fix(job-wait-loop): require mat3ra-api-client 2026.10.8.post0, the first with a per-request token The status fetch reads `endpoint.auth`, and the retry after a 401 relies on request() merging the current auth headers. Both first ship in 2026.10.8.post0. The floor is set in pyproject.toml and in both JupyterLite package lists in config.yml. Co-Authored-By: Claude Opus 5.5 (1M context) --- config.yml | 4 ++-- pyproject.toml | 2 +- 2 files changed, 3 insertions(+), 3 deletions(-) diff --git a/config.yml b/config.yml index 5a5c5477d..7b88d0f05 100644 --- a/config.yml +++ b/config.yml @@ -66,7 +66,7 @@ notebooks: - ipywidgets - tabulate==0.9.0 - mat3ra-utils - - mat3ra-api-client + - mat3ra-api-client>=2026.10.8.post0 - name: dataframe packages_pyodide: - jinja2 @@ -79,7 +79,7 @@ notebooks: - emfs:/drive/packages/pydantic-2.7.1-py3-none-any.whl - mat3ra-standata - mat3ra-ide - - mat3ra-api-client + - mat3ra-api-client>=2026.10.8.post0 - requests - mat3ra-ade - mat3ra-mode diff --git a/pyproject.toml b/pyproject.toml index 0f4f23014..2840c8d43 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -24,7 +24,7 @@ utils_standata = [ "mat3ra-standata", ] api = [ - "mat3ra-api-client", + "mat3ra-api-client>=2026.10.8.post0", ] materials = [ # ase >=3.25.0 is required for supercell to be generated,