diff --git a/src/winml/modelkit/session/ep_registry.py b/src/winml/modelkit/session/ep_registry.py index 47810ad59..5426518d1 100644 --- a/src/winml/modelkit/session/ep_registry.py +++ b/src/winml/modelkit/session/ep_registry.py @@ -11,7 +11,8 @@ from __future__ import annotations import logging -from typing import TYPE_CHECKING, cast +import os +from typing import TYPE_CHECKING, Any, cast if TYPE_CHECKING: @@ -20,6 +21,197 @@ logger = logging.getLogger(__name__) + +def _ep_download_timeout_default() -> int: + """Read ``WINMLCLI_EP_DOWNLOAD_TIMEOUT`` (seconds) or fall back to 5 minutes. + + Lets users on slow networks raise the cap without code changes. Falls back + to the default when the env var is unset, empty, or non-integer. + """ + raw = os.environ.get("WINMLCLI_EP_DOWNLOAD_TIMEOUT") + if not raw: + return 5 * 60 + try: + return int(raw) + except ValueError: + logger.warning("Invalid WINMLCLI_EP_DOWNLOAD_TIMEOUT=%r; using default 300s.", raw) + return 5 * 60 + + +# Evaluated once at module import. Changing WINMLCLI_EP_DOWNLOAD_TIMEOUT +# after import does NOT take effect for the running process; tests that need +# a different value should monkeypatch ep_registry.EP_DOWNLOAD_TIMEOUT_SECONDS +# directly. +EP_DOWNLOAD_TIMEOUT_SECONDS = _ep_download_timeout_default() + + +class _NoopBar: + """No-op stand-in for tqdm when the optional dependency is missing. + + Exposes the attribute (``n``) and methods (``refresh``, ``close``) that + ``_ensure_provider_ready`` touches, so the helper can stay branch-free. + """ + + def __init__(self) -> None: + self.n = 0 + + def refresh(self) -> None: + return None + + def close(self) -> None: + return None + + +def _make_progress_bar() -> Any: + """Return a tqdm bar if tqdm is installed, else a silent no-op stand-in. + + tqdm is a dev-only optional dep in this package, so production installs + without it must still complete EP downloads — they just lose the live bar. + The pre-download Console notice is emitted by the caller and is unaffected. + + Format: ``Downloading... ████████████░░░░░░ 62%`` + """ + try: + from tqdm import tqdm + except ImportError: + return _NoopBar() + return tqdm( + total=100, + bar_format="Downloading... {bar} {percentage:3.0f}%", + ascii="░█", + leave=True, + ) + + +def _parse_ep_metadata_from_path(library_path: str) -> tuple[str, str]: + r"""Best-effort ``(version, package_family_name)`` from an EP's install path. + + WinML's ``ExecutionProvider`` handle sometimes returns empty ``version`` / + ``package_family_name`` even after the EP is Ready. When the EP is delivered + as an MSIX package its ``library_path`` lives under ``WindowsApps`` in a + folder named with the full package identity:: + + ...\\WindowsApps\\____\\... + + e.g. ``MicrosoftCorporationII.WinML.Intel.OpenVINO.EP.1.8_1.8.79.0_x64__8wekyb3d8bbwe`` + yields version ``1.8.79.0`` and package family name + ``MicrosoftCorporationII.WinML.Intel.OpenVINO.EP.1.8_8wekyb3d8bbwe`` (the + package family name is ``_``). + + Returns ``("", "")`` when the path is empty or does not match this layout. + """ + import re + from itertools import pairwise + from pathlib import PurePath + + if not library_path: + return "", "" + + parts = PurePath(library_path).parts + pkg_folder = next( + (child for parent, child in pairwise(parts) if parent.lower() == "windowsapps"), + "", + ) + # Full MSIX package name: Name_Version_Arch_ResourceId_PublisherId (ResourceId + # is usually empty, giving the doubled "__" before the publisher id). + segments = pkg_folder.split("_") + if len(segments) < 5: + return "", "" + + name, version, publisher = segments[0], segments[1], segments[-1] + # Guard against unexpected folder shapes: version must be dotted-numeric. + if not re.fullmatch(r"\d+(\.\d+)*", version): + version = "" + package_family_name = f"{name}_{publisher}" if name and publisher else "" + return version, package_family_name + + +def _ensure_provider_ready(provider: Any) -> None: + """Ensure an EP is ready, showing a tqdm progress bar when downloading. + + Providers already in the ``Ready`` state take the synchronous fast path so + cached EPs do not flash a 0-100% bar. Otherwise drives a tqdm bar from + ``ensure_ready_async``'s ``on_progress`` callback (cumulative fraction + 0.0-1.0, per windowsml docs) and waits for the ``on_complete`` callback + via a threading.Event with a ``EP_DOWNLOAD_TIMEOUT_SECONDS`` timeout. On + timeout the async op is cancelled and ``TimeoutError`` is raised. + """ + import threading + + from windowsml import EpReadyState + + if provider.ready_state == EpReadyState.Ready: + provider.ensure_ready() + return + + # Lazy-import to keep ep_registry import cheap (rich pulls in pygments etc.); + # this branch only runs on the cold "EP needs download" path. + from ..utils.console import get_console + + console = get_console() + console.print(f"[WinML] Installing Execution Provider: [bold]{provider.name}[/bold]") + + bar = _make_progress_bar() + done = threading.Event() + + def _on_progress(fraction: float) -> None: + # Native ops may fire a stale on_progress after on_complete; once done + # is set the main thread owns bar.n (forces it to 100 and closes the + # bar), so silently drop late callbacks instead of clobbering 100 with + # an earlier fraction or writing to a closed bar. + if done.is_set(): + return + bar.n = max(0, min(100, int(fraction * 100))) + bar.refresh() + + op = None + success = False + try: + op = provider.ensure_ready_async(on_complete=done.set, on_progress=_on_progress) + if not done.wait(timeout=EP_DOWNLOAD_TIMEOUT_SECONDS): + op.cancel() + raise TimeoutError( + f"EP {provider.name!r} download did not complete within " + f"{EP_DOWNLOAD_TIMEOUT_SECONDS}s; cancelled." + ) + # Surface any native failure (raises OSError on error). + op.get_status() + # Success: providers usually fire on_progress(1.0) before on_complete, + # but force the bar to 100 in case they didn't. + bar.n = 100 + bar.refresh() + success = True + finally: + bar.close() + if op is not None: + op.close() + if not success: + # Failure-path notice — kept in finally so it fires for every + # non-success exit (launch failure, timeout, get_status OSError). + # Printed after bar.close() so it appears below the bar's last frame. + console.print(f"[red]❌ Failed to download {provider.name} EP[/red]") + console.print("Try:") + console.print(" 1. Check your internet connection") + console.print(" 2. Troubleshoot: https://aka.ms/winmlcli/ep-errors") + + console.print(f"{provider.name} EP installed successfully.") + + # The native handle sometimes reports empty version/PFN even once Ready; + # fall back to parsing them from the MSIX install path. Skip a line entirely + # when its value can't be determined rather than printing a blank field. + version = provider.version + package_family_name = provider.package_family_name + if not version or not package_family_name: + parsed_version, parsed_pfn = _parse_ep_metadata_from_path(provider.library_path) + version = version or parsed_version + package_family_name = package_family_name or parsed_pfn + if version: + console.print(f"- Version: {version}", soft_wrap=True) + if package_family_name: + # soft_wrap so long package family names aren't hard-wrapped mid-string. + console.print(f"- Package Family Name: {package_family_name}", soft_wrap=True) + + # Singleton instance _winml_ep_registry: WinMLEPRegistry | None = None @@ -80,9 +272,19 @@ def _load_ep_catalog(self) -> None: with EpCatalog() as catalog: for provider in catalog.find_all_providers(): try: - provider.ensure_ready() + _ensure_provider_ready(provider) + except OSError as e: + # windowsml maps native HRESULT failures to OSError; surface + # winerror so the HRESULT is grep-able in logs. + logger.info( + "Failed to ensure EP %s is ready: %s (winerror=%s)", + provider.name, + e, + getattr(e, "winerror", None), + ) + continue except Exception as e: - logger.debug("Failed to ensure EP %s is ready: %s", provider.name, e) + logger.info("Failed to ensure EP %s is ready: %s", provider.name, e) continue if provider.library_path == "": continue diff --git a/tests/unit/session/test_ep_registry.py b/tests/unit/session/test_ep_registry.py new file mode 100644 index 000000000..f7c4b15c8 --- /dev/null +++ b/tests/unit/session/test_ep_registry.py @@ -0,0 +1,514 @@ +# ------------------------------------------------------------------------- +# Copyright (c) Microsoft Corporation. All rights reserved. +# Licensed under the MIT License. +# -------------------------------------------------------------------------- +"""Tests for ep_registry module helpers.""" + +from __future__ import annotations + +import logging +import sys +import types +from unittest.mock import MagicMock + +import pytest + + +def _install_fake_windowsml(monkeypatch: pytest.MonkeyPatch) -> types.SimpleNamespace: + """Inject a fake ``windowsml`` module exposing only what the helper needs.""" + + class _EpReadyState: + Ready = 0 + NotReady = 1 + NotPresent = 2 + + fake = types.ModuleType("windowsml") + fake.EpReadyState = _EpReadyState # type: ignore[attr-defined] + monkeypatch.setitem(sys.modules, "windowsml", fake) + return types.SimpleNamespace(EpReadyState=_EpReadyState) + + +def _install_fake_tqdm(monkeypatch: pytest.MonkeyPatch) -> MagicMock: + """Inject a fake ``tqdm.tqdm``. Helper writes ``bar.n`` directly + refresh().""" + + fake_bar = MagicMock() + fake_bar.n = 0 + + tqdm_mod = types.ModuleType("tqdm") + tqdm_mod.tqdm = MagicMock(return_value=fake_bar) # type: ignore[attr-defined] + monkeypatch.setitem(sys.modules, "tqdm", tqdm_mod) + return fake_bar + + +def test_ensure_provider_ready_skips_progress_when_already_ready( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """Ready providers take the sync fast path and skip the async/progress flow.""" + from winml.modelkit.session import ep_registry + + ns = _install_fake_windowsml(monkeypatch) + provider = MagicMock() + provider.ready_state = ns.EpReadyState.Ready + + ep_registry._ensure_provider_ready(provider) + + provider.ensure_ready.assert_called_once_with() + provider.ensure_ready_async.assert_not_called() + + +def test_ensure_provider_ready_drives_progress_bar(monkeypatch: pytest.MonkeyPatch) -> None: + """NotReady providers go through ensure_ready_async; on_progress drives a tqdm bar.""" + from winml.modelkit.session import ep_registry + + ns = _install_fake_windowsml(monkeypatch) + fake_bar = _install_fake_tqdm(monkeypatch) + + op = MagicMock() + + def fake_ensure_async(on_complete=None, on_progress=None): + # Simulate cumulative-fraction progress callbacks, then completion. + for fraction in (0.0, 0.25, 0.5, 1.0): + on_progress(fraction) + on_complete() + return op + + provider = MagicMock() + provider.name = "FakeEP" + provider.ready_state = ns.EpReadyState.NotPresent + provider.ensure_ready_async.side_effect = fake_ensure_async + + ep_registry._ensure_provider_ready(provider) + + provider.ensure_ready.assert_not_called() + provider.ensure_ready_async.assert_called_once() + op.get_status.assert_called_once_with() + op.cancel.assert_not_called() + op.close.assert_called_once_with() + fake_bar.close.assert_called_once_with() + # Success path forces bar.n to 100 even though the last fraction was 1.0. + assert fake_bar.n == 100 + fake_bar.refresh.assert_called() + + +def test_ensure_provider_ready_ignores_stale_progress_after_complete( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """A stale on_progress fired after on_complete must NOT clobber bar.n. + + The native layer can fire a late progress callback from its own thread + after on_complete; once `done` is set, the main thread owns bar.n and the + callback should drop instead of writing an earlier fraction back over the + forced 100 (or worse, writing to an already-closed bar).""" + from winml.modelkit.session import ep_registry + + ns = _install_fake_windowsml(monkeypatch) + fake_bar = _install_fake_tqdm(monkeypatch) + + op = MagicMock() + saved_progress: list = [] + + def fake_ensure_async(on_complete=None, on_progress=None): + on_progress(0.5) + on_complete() + saved_progress.append(on_progress) # Fire a stale callback after. + return op + + provider = MagicMock() + provider.name = "FakeEP" + provider.ready_state = ns.EpReadyState.NotPresent + provider.ensure_ready_async.side_effect = fake_ensure_async + + ep_registry._ensure_provider_ready(provider) + + # Fire the stale callback as if it arrived after _ensure_provider_ready + # already set bar.n = 100 and closed the bar. + assert fake_bar.n == 100 + saved_progress[0](0.62) + # Stale callback must have been dropped — bar.n stays at 100. + assert fake_bar.n == 100 + + +def test_ensure_provider_ready_warns_before_download( + monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str] +) -> None: + """A yellow notice is printed to the stderr Console before download + so users know the wait is expected.""" + from winml.modelkit.session import ep_registry + + ns = _install_fake_windowsml(monkeypatch) + _install_fake_tqdm(monkeypatch) + + op = MagicMock() + + def fake_ensure_async(on_complete=None, on_progress=None): + on_complete() + return op + + provider = MagicMock() + provider.name = "FakeEP" + provider.ready_state = ns.EpReadyState.NotPresent + provider.ensure_ready_async.side_effect = fake_ensure_async + + ep_registry._ensure_provider_ready(provider) + + err = capsys.readouterr().err + assert "[WinML] Installing Execution Provider" in err + assert "FakeEP" in err + + +def test_ensure_provider_ready_forces_bar_to_100_on_success_without_progress( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """If the async op completes successfully without ever firing on_progress, + the success path forces the bar to 100 so the final render shows full.""" + from winml.modelkit.session import ep_registry + + ns = _install_fake_windowsml(monkeypatch) + fake_bar = _install_fake_tqdm(monkeypatch) + + op = MagicMock() + + def fake_ensure_async(on_complete=None, on_progress=None): + on_complete() # No progress firings, but completes immediately. + return op + + provider = MagicMock() + provider.name = "FakeEP" + provider.ready_state = ns.EpReadyState.NotReady + provider.ensure_ready_async.side_effect = fake_ensure_async + + ep_registry._ensure_provider_ready(provider) + + assert fake_bar.n == 100 + fake_bar.close.assert_called_once_with() + op.close.assert_called_once_with() + + +def test_ensure_provider_ready_times_out_and_cancels( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """When on_complete never fires within the timeout, cancel and raise TimeoutError.""" + from winml.modelkit.session import ep_registry + + ns = _install_fake_windowsml(monkeypatch) + fake_bar = _install_fake_tqdm(monkeypatch) + + # Shrink the timeout so the test runs in milliseconds, not minutes. + monkeypatch.setattr(ep_registry, "EP_DOWNLOAD_TIMEOUT_SECONDS", 0.05) + + op = MagicMock() + provider = MagicMock() + provider.name = "SlowEP" + provider.ready_state = ns.EpReadyState.NotPresent + # ensure_ready_async returns op but never calls on_complete -> times out. + provider.ensure_ready_async.return_value = op + + with pytest.raises(TimeoutError, match="SlowEP"): + ep_registry._ensure_provider_ready(provider) + + op.cancel.assert_called_once_with() + op.close.assert_called_once_with() + fake_bar.close.assert_called_once_with() + # Bar must NOT be force-filled on timeout — it should reflect where the + # download stalled (here: 0 because no progress callbacks ever fired). + assert fake_bar.n == 0 + + +def test_ensure_provider_ready_surfaces_get_status_error( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """A failure surfaced by get_status() propagates after cleanup.""" + from winml.modelkit.session import ep_registry + + ns = _install_fake_windowsml(monkeypatch) + fake_bar = _install_fake_tqdm(monkeypatch) + + op = MagicMock() + op.get_status.side_effect = OSError("native error") + + def fake_ensure_async(on_complete=None, on_progress=None): + on_complete() + return op + + provider = MagicMock() + provider.name = "FakeEP" + provider.ready_state = ns.EpReadyState.NotPresent + provider.ensure_ready_async.side_effect = fake_ensure_async + + with pytest.raises(OSError, match="native error"): + ep_registry._ensure_provider_ready(provider) + + fake_bar.close.assert_called_once_with() + op.close.assert_called_once_with() + # Native error must NOT force-fill the bar — it should reflect where + # the download failed (here: 0 because no progress callbacks fired). + assert fake_bar.n == 0 + + +def test_ensure_provider_ready_prints_success_with_metadata( + monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str] +) -> None: + """After a successful install, print ' EP installed successfully.' + followed by Version and Package Family Name lines.""" + from winml.modelkit.session import ep_registry + + ns = _install_fake_windowsml(monkeypatch) + _install_fake_tqdm(monkeypatch) + + op = MagicMock() + + def fake_ensure_async(on_complete=None, on_progress=None): + on_complete() + return op + + provider = MagicMock() + provider.name = "OpenVINOExecutionProvider" + provider.version = "1.2.0" + provider.package_family_name = "Microsoft.OpenVINOExecutionProvider_8wekyb3d8bbwe" + provider.ready_state = ns.EpReadyState.NotPresent + provider.ensure_ready_async.side_effect = fake_ensure_async + + ep_registry._ensure_provider_ready(provider) + + err = capsys.readouterr().err + assert "OpenVINOExecutionProvider EP installed successfully." in err + assert "- Version: 1.2.0" in err + assert "- Package Family Name: Microsoft.OpenVINOExecutionProvider_8wekyb3d8bbwe" in err + + +def test_ensure_provider_ready_falls_back_to_path_metadata_when_native_empty( + monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str] +) -> None: + """When the native handle reports empty version/PFN, recover both from the + MSIX install path.""" + from winml.modelkit.session import ep_registry + + ns = _install_fake_windowsml(monkeypatch) + _install_fake_tqdm(monkeypatch) + + op = MagicMock() + + def fake_ensure_async(on_complete=None, on_progress=None): + on_complete() + return op + + provider = MagicMock() + provider.name = "OpenVINOExecutionProvider" + provider.version = "" + provider.package_family_name = "" + provider.library_path = ( + r"C:\Program Files\WindowsApps" + r"\MicrosoftCorporationII.WinML.Intel.OpenVINO.EP.1.8_1.8.79.0_x64__8wekyb3d8bbwe" + r"\ExecutionProvider\onnxruntime_providers_openvino_plugin.dll" + ) + provider.ready_state = ns.EpReadyState.NotPresent + provider.ensure_ready_async.side_effect = fake_ensure_async + + ep_registry._ensure_provider_ready(provider) + + err = capsys.readouterr().err + assert "- Version: 1.8.79.0" in err + assert ( + "- Package Family Name: MicrosoftCorporationII.WinML.Intel.OpenVINO.EP.1.8_8wekyb3d8bbwe" + in err + ) + + +def test_ensure_provider_ready_skips_metadata_lines_when_unavailable( + monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str] +) -> None: + """If neither the native handle nor the path yields metadata, omit the + Version / Package Family Name lines entirely (no blank fields).""" + from winml.modelkit.session import ep_registry + + ns = _install_fake_windowsml(monkeypatch) + _install_fake_tqdm(monkeypatch) + + op = MagicMock() + + def fake_ensure_async(on_complete=None, on_progress=None): + on_complete() + return op + + provider = MagicMock() + provider.name = "MysteryEP" + provider.version = "" + provider.package_family_name = "" + provider.library_path = r"C:\some\local\path\provider.dll" + provider.ready_state = ns.EpReadyState.NotPresent + provider.ensure_ready_async.side_effect = fake_ensure_async + + ep_registry._ensure_provider_ready(provider) + + err = capsys.readouterr().err + assert "MysteryEP EP installed successfully." in err + assert "- Version:" not in err + assert "- Package Family Name:" not in err + + +def test_ensure_provider_ready_prints_failure_message_on_timeout( + monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str] +) -> None: + """A timed-out download prints the ❌ failure notice with retry hints, + and does NOT emit the 'installed successfully.' line.""" + from winml.modelkit.session import ep_registry + + ns = _install_fake_windowsml(monkeypatch) + _install_fake_tqdm(monkeypatch) + monkeypatch.setattr(ep_registry, "EP_DOWNLOAD_TIMEOUT_SECONDS", 0.05) + + provider = MagicMock() + provider.name = "SlowEP" + provider.ready_state = ns.EpReadyState.NotPresent + provider.ensure_ready_async.return_value = MagicMock() + + with pytest.raises(TimeoutError): + ep_registry._ensure_provider_ready(provider) + + err = capsys.readouterr().err + assert "installed successfully" not in err + assert "Failed to download SlowEP EP" in err + assert "Check your internet connection" in err + assert "Troubleshoot:" in err + assert "https://aka.ms/winmlcli/ep-errors" in err + + +def test_ensure_provider_ready_prints_failure_message_on_async_launch_error( + monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str] +) -> None: + """A failure at the ensure_ready_async() launch also prints the ❌ block.""" + from winml.modelkit.session import ep_registry + + ns = _install_fake_windowsml(monkeypatch) + _install_fake_tqdm(monkeypatch) + + provider = MagicMock() + provider.name = "BadEP" + provider.ready_state = ns.EpReadyState.NotPresent + provider.ensure_ready_async.side_effect = OSError("native launch failed") + + with pytest.raises(OSError, match="native launch failed"): + ep_registry._ensure_provider_ready(provider) + + err = capsys.readouterr().err + assert "Failed to download BadEP EP" in err + assert "installed successfully" not in err + + +def test_ensure_provider_ready_closes_bar_when_ensure_ready_async_raises( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """If ensure_ready_async itself raises, the bar must still be closed + (op was never assigned, so op.close() is skipped via the None sentinel).""" + from winml.modelkit.session import ep_registry + + ns = _install_fake_windowsml(monkeypatch) + fake_bar = _install_fake_tqdm(monkeypatch) + + provider = MagicMock() + provider.name = "FakeEP" + provider.ready_state = ns.EpReadyState.NotPresent + provider.ensure_ready_async.side_effect = RuntimeError("native init failed") + + with pytest.raises(RuntimeError, match="native init failed"): + ep_registry._ensure_provider_ready(provider) + + fake_bar.close.assert_called_once_with() + + +def test_ensure_provider_ready_works_without_tqdm( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """When tqdm (a dev-only optional dep) is missing, the download still + completes via the _NoopBar fallback — no ImportError, no progress UI.""" + from winml.modelkit.session import ep_registry + + ns = _install_fake_windowsml(monkeypatch) + # Simulate tqdm being uninstalled: make `from tqdm import tqdm` raise. + monkeypatch.setitem(sys.modules, "tqdm", None) + + op = MagicMock() + + def fake_ensure_async(on_complete=None, on_progress=None): + # Drive a progress update too — the no-op bar must tolerate bar.n = ... + on_progress(0.5) + on_complete() + return op + + provider = MagicMock() + provider.name = "FakeEP" + provider.ready_state = ns.EpReadyState.NotPresent + provider.ensure_ready_async.side_effect = fake_ensure_async + + ep_registry._ensure_provider_ready(provider) + + op.get_status.assert_called_once_with() + op.close.assert_called_once_with() + + +class TestParseEpMetadataFromPath: + """`_parse_ep_metadata_from_path` recovers (version, PFN) from install paths.""" + + def test_parses_openvino_windowsapps_path(self) -> None: + from winml.modelkit.session import ep_registry + + path = ( + r"C:\Program Files\WindowsApps" + r"\MicrosoftCorporationII.WinML.Intel.OpenVINO.EP.1.8_1.8.79.0_x64__8wekyb3d8bbwe" + r"\ExecutionProvider\onnxruntime_providers_openvino_plugin.dll" + ) + version, pfn = ep_registry._parse_ep_metadata_from_path(path) + assert version == "1.8.79.0" + assert pfn == "MicrosoftCorporationII.WinML.Intel.OpenVINO.EP.1.8_8wekyb3d8bbwe" + + def test_empty_path_returns_empty(self) -> None: + from winml.modelkit.session import ep_registry + + assert ep_registry._parse_ep_metadata_from_path("") == ("", "") + + def test_non_windowsapps_path_returns_empty(self) -> None: + from winml.modelkit.session import ep_registry + + assert ep_registry._parse_ep_metadata_from_path(r"C:\local\ep\provider.dll") == ("", "") + + def test_non_numeric_version_segment_dropped_but_pfn_kept(self) -> None: + """A folder that doesn't carry a dotted-numeric version still yields a + PFN, but the version is left empty rather than guessed.""" + from winml.modelkit.session import ep_registry + + path = r"C:\Program Files\WindowsApps\Some.Package_notaversion_x64__pubhash\ep.dll" + version, pfn = ep_registry._parse_ep_metadata_from_path(path) + assert version == "" + assert pfn == "Some.Package_pubhash" + + +class TestEpDownloadTimeoutDefault: + """`_ep_download_timeout_default` reads ``WINMLCLI_EP_DOWNLOAD_TIMEOUT``.""" + + def test_default_when_unset(self, monkeypatch: pytest.MonkeyPatch) -> None: + from winml.modelkit.session import ep_registry + + monkeypatch.delenv("WINMLCLI_EP_DOWNLOAD_TIMEOUT", raising=False) + assert ep_registry._ep_download_timeout_default() == 5 * 60 + + def test_override_via_env(self, monkeypatch: pytest.MonkeyPatch) -> None: + from winml.modelkit.session import ep_registry + + monkeypatch.setenv("WINMLCLI_EP_DOWNLOAD_TIMEOUT", "1800") + assert ep_registry._ep_download_timeout_default() == 1800 + + def test_falls_back_to_default_on_invalid_value( + self, monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture + ) -> None: + from winml.modelkit.session import ep_registry + + monkeypatch.setenv("WINMLCLI_EP_DOWNLOAD_TIMEOUT", "not-a-number") + with caplog.at_level(logging.WARNING, logger=ep_registry.logger.name): + assert ep_registry._ep_download_timeout_default() == 5 * 60 + assert any("WINMLCLI_EP_DOWNLOAD_TIMEOUT" in r.getMessage() for r in caplog.records) + + def test_empty_string_uses_default(self, monkeypatch: pytest.MonkeyPatch) -> None: + from winml.modelkit.session import ep_registry + + monkeypatch.setenv("WINMLCLI_EP_DOWNLOAD_TIMEOUT", "") + assert ep_registry._ep_download_timeout_default() == 5 * 60