From 1f0e6616ab08547bc9a380308d73528e79cd02c9 Mon Sep 17 00:00:00 2001 From: Hualiang Xie Date: Wed, 15 Jul 2026 14:06:35 +0800 Subject: [PATCH 1/4] feat(loader): add pipeline_tag fallback to task detection When architecture-based task detection (Stage 1c) fails because config.architectures contains a generic name like 'Model', fall back to querying the HuggingFace Hub pipeline_tag (Stage 1e) before the last-resort feature-extraction default (Stage 1d). - Add TaskSource.PIPELINE_TAG enum value for provenance tracking - Add Stage 1e in resolve_task() between 1c and 1d: - Checks model_id is not a local path via _is_local_path - Queries HfApi().model_info(model_id).pipeline_tag - Validates against KNOWN_TASKS after normalize_task() - Wrapped in try/except for graceful fallthrough on errors - Add 4 tests covering valid tag, invalid task, API failure, local path Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- src/winml/modelkit/loader/resolution.py | 19 ++++++ tests/unit/loader/test_detect_task.py | 78 +++++++++++++++++++++++++ 2 files changed, 97 insertions(+) diff --git a/src/winml/modelkit/loader/resolution.py b/src/winml/modelkit/loader/resolution.py index 871c7c817..6ff74896a 100644 --- a/src/winml/modelkit/loader/resolution.py +++ b/src/winml/modelkit/loader/resolution.py @@ -24,6 +24,7 @@ from .task import ( HF_TASK_DEFAULTS, + KNOWN_TASKS, get_default_task_for_model_id, get_supported_tasks, normalize_task, @@ -253,6 +254,7 @@ class TaskSource(str, Enum): SENTINEL_DEFAULT = "sentinel-default" # (model_type, None) sentinel TASKS_MANAGER = "tasks-manager" # Optimum inference (incl. fill-mask upgrade) WRAPPED_LIBRARY = "wrapped-library" # no architectures -> first supported task + PIPELINE_TAG = "pipeline-tag" # Hub pipeline_tag fallback HF_TASK_DEFAULT = "hf-task-default" # last-resort default @@ -572,6 +574,23 @@ def resolve_task( except ValueError: opt_task = None + # 1e. Hub pipeline_tag fallback + if opt_task is None and model_id: + try: + from ..utils.hub_utils import _is_local_path + + if not _is_local_path(model_id): + from huggingface_hub import HfApi + + tag = HfApi().model_info(model_id).pipeline_tag + if tag: + normalized_tag = normalize_task(tag) + if normalized_tag in KNOWN_TASKS: + opt_task = normalized_tag + source = TaskSource.PIPELINE_TAG + except Exception: + pass + # 1d. last-resort default if opt_task is None: opt_task = next(iter(HF_TASK_DEFAULTS)) diff --git a/tests/unit/loader/test_detect_task.py b/tests/unit/loader/test_detect_task.py index e2a39aea6..6eaaf45aa 100644 --- a/tests/unit/loader/test_detect_task.py +++ b/tests/unit/loader/test_detect_task.py @@ -228,3 +228,81 @@ def test_resolve_task_case1_surfaces_modality_aware_task() -> None: r = resolve_task(cfg) assert r.task == "image-feature-extraction" assert r.optimum_task == "feature-extraction" + + +# ============================================================================= +# Stage 1e — Hub pipeline_tag fallback +# ============================================================================= + +_IS_LOCAL = "winml.modelkit.utils.hub_utils._is_local_path" +_HF_API = "huggingface_hub.HfApi" + + +def _fake_model_info(pipeline_tag: str | None): + """Return a mock model_info object with a ``pipeline_tag`` attribute.""" + return type("FakeModelInfo", (), {"pipeline_tag": pipeline_tag})() + + +def test_resolve_task_uses_pipeline_tag_when_architecture_fails() -> None: + """When config.architectures contains a generic name (e.g. 'Model') that + TasksManager cannot resolve, the Hub pipeline_tag is used as fallback.""" + cfg = _FakeConfig("faketype", name_or_path="audeering/wav2vec2-large-robust-24-ft-age-gender") + mock_api = type( + "MockApi", + (), + {"model_info": lambda self, _: _fake_model_info("audio-classification")}, + )() + with ( + patch(_INFER, side_effect=ValueError("unknown arch")), + patch(_IS_LOCAL, return_value=False), + patch(_HF_API, return_value=mock_api), + ): + r = resolve_task(cfg) + assert r.task == "audio-classification" + assert r.source == TaskSource.PIPELINE_TAG + + +def test_resolve_task_pipeline_tag_skips_invalid_task() -> None: + """When pipeline_tag is not a recognized task, falls through to last-resort default.""" + cfg = _FakeConfig("faketype", name_or_path="someone/some-model") + mock_api = type( + "MockApi", + (), + {"model_info": lambda self, _: _fake_model_info("not-a-real-task")}, + )() + with ( + patch(_INFER, side_effect=ValueError("unknown arch")), + patch(_IS_LOCAL, return_value=False), + patch(_HF_API, return_value=mock_api), + ): + r = resolve_task(cfg) + assert r.source == TaskSource.HF_TASK_DEFAULT + + +def test_resolve_task_pipeline_tag_handles_api_failure() -> None: + """When the Hub API call fails (network error, etc.), falls through gracefully.""" + cfg = _FakeConfig("faketype", name_or_path="someone/some-model") + mock_api = type( + "MockApi", + (), + {"model_info": lambda self, _: (_ for _ in ()).throw(ConnectionError("offline"))}, + )() + with ( + patch(_INFER, side_effect=ValueError("unknown arch")), + patch(_IS_LOCAL, return_value=False), + patch(_HF_API, return_value=mock_api), + ): + r = resolve_task(cfg) + assert r.source == TaskSource.HF_TASK_DEFAULT + + +def test_resolve_task_pipeline_tag_skips_local_path() -> None: + """When model_id is a local path, the Hub API is not called.""" + cfg = _FakeConfig("faketype", name_or_path="./local-model") + with ( + patch(_INFER, side_effect=ValueError("unknown arch")), + patch(_IS_LOCAL, return_value=True), + patch(_HF_API, side_effect=AssertionError("must not call HfApi for local paths")), + ): + r = resolve_task(cfg) + assert r.source == TaskSource.HF_TASK_DEFAULT From 03e19dfd64a4a0f8b0fd2df42c88eaa1d58c92fd Mon Sep 17 00:00:00 2001 From: Hualiang Xie Date: Wed, 15 Jul 2026 14:42:06 +0800 Subject: [PATCH 2/4] fix: add explanatory comment to bare except clause MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Addresses CodeQL 'Empty except' finding by documenting that the pass is intentional — network errors, invalid model IDs, and missing pipeline_tag all gracefully fall through to Stage 1d. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- src/winml/modelkit/loader/resolution.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/winml/modelkit/loader/resolution.py b/src/winml/modelkit/loader/resolution.py index 6ff74896a..a822b8a82 100644 --- a/src/winml/modelkit/loader/resolution.py +++ b/src/winml/modelkit/loader/resolution.py @@ -588,7 +588,7 @@ def resolve_task( if normalized_tag in KNOWN_TASKS: opt_task = normalized_tag source = TaskSource.PIPELINE_TAG - except Exception: + except Exception: # graceful fallthrough: network errors, invalid model_id, etc. pass # 1d. last-resort default From 591e02ee437697313eeb5a1a83e80a152d069a5c Mon Sep 17 00:00:00 2001 From: Hualiang Xie Date: Wed, 15 Jul 2026 17:26:57 +0800 Subject: [PATCH 3/4] =?UTF-8?q?refactor:=20address=20review=20=E2=80=94=20?= =?UTF-8?q?centralize=20Hub=20access,=20add=20timeout=20and=20logging?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Extract get_pipeline_tag() helper in hub_utils.py: - Encapsulates _is_local_path check, HfApi call with 10s timeout, and logger.debug on failure (matching existing hub_utils conventions) - Simplify Stage 1e in resolution.py to call get_pipeline_tag() directly - Update resolve_task docstring to mention pipeline-tag stage - Rewrite tests to mock get_pipeline_tag at source, verify call args Addresses review feedback: - Timeout on HfApi().model_info() (was unbounded) - logger.debug breadcrumb instead of bare pass - Centralized Hub access (no more reaching into private _is_local_path) - Stronger test assertion (mock_tag.assert_called_once_with) Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- src/winml/modelkit/loader/resolution.py | 26 ++++++---------- src/winml/modelkit/utils/hub_utils.py | 22 +++++++++++++ tests/unit/loader/test_detect_task.py | 41 ++++++------------------- 3 files changed, 41 insertions(+), 48 deletions(-) diff --git a/src/winml/modelkit/loader/resolution.py b/src/winml/modelkit/loader/resolution.py index a822b8a82..8f9425d80 100644 --- a/src/winml/modelkit/loader/resolution.py +++ b/src/winml/modelkit/loader/resolution.py @@ -442,8 +442,8 @@ def resolve_task( """Resolve a single model's task + class from an HF config. Stages: 0 user override -> 1 detect (override / no-architectures / - TasksManager / default) -> 2 model class -> 3 modality upgrade - (detection path only) -> 4 composite tag. + TasksManager / pipeline-tag / default) -> 2 model class -> 3 modality + upgrade (detection path only) -> 4 composite tag. ``model_type_override`` lets a caller drive resolution with a build variant (e.g. ``qwen3_transformer_only``) without mutating the loaded HF config; when @@ -576,20 +576,14 @@ def resolve_task( # 1e. Hub pipeline_tag fallback if opt_task is None and model_id: - try: - from ..utils.hub_utils import _is_local_path - - if not _is_local_path(model_id): - from huggingface_hub import HfApi - - tag = HfApi().model_info(model_id).pipeline_tag - if tag: - normalized_tag = normalize_task(tag) - if normalized_tag in KNOWN_TASKS: - opt_task = normalized_tag - source = TaskSource.PIPELINE_TAG - except Exception: # graceful fallthrough: network errors, invalid model_id, etc. - pass + from ..utils.hub_utils import get_pipeline_tag + + tag = get_pipeline_tag(model_id) + if tag: + normalized_tag = normalize_task(tag) + if normalized_tag in KNOWN_TASKS: + opt_task = normalized_tag + source = TaskSource.PIPELINE_TAG # 1d. last-resort default if opt_task is None: diff --git a/src/winml/modelkit/utils/hub_utils.py b/src/winml/modelkit/utils/hub_utils.py index d45adf87f..a62876d98 100644 --- a/src/winml/modelkit/utils/hub_utils.py +++ b/src/winml/modelkit/utils/hub_utils.py @@ -144,6 +144,28 @@ def is_hub_model(model_name_or_path: str) -> tuple[bool, dict]: return False, {"type": "local", "path": model_name_or_path} +_PIPELINE_TAG_TIMEOUT = 10 # seconds + + +def get_pipeline_tag(model_id: str) -> str | None: + """Return the Hub ``pipeline_tag`` for *model_id*, or ``None``. + + Lightweight helper that skips the full metadata extraction of + ``is_hub_model``. Returns ``None`` (never raises) when *model_id* is a + local path, the Hub is unreachable, or the model has no tag. + """ + if _is_local_path(model_id): + return None + try: + from huggingface_hub import HfApi + + info = HfApi().model_info(model_id, timeout=_PIPELINE_TAG_TIMEOUT) + return getattr(info, "pipeline_tag", None) + except Exception: + logger.debug("pipeline_tag lookup failed for '%s'", model_id, exc_info=True) + return None + + def inject_hub_metadata(onnx_model: Any, model_name_or_path: str, metadata: dict) -> None: """Inject HuggingFace Hub metadata into ONNX model. diff --git a/tests/unit/loader/test_detect_task.py b/tests/unit/loader/test_detect_task.py index 6eaaf45aa..8b8745488 100644 --- a/tests/unit/loader/test_detect_task.py +++ b/tests/unit/loader/test_detect_task.py @@ -234,28 +234,16 @@ def test_resolve_task_case1_surfaces_modality_aware_task() -> None: # Stage 1e — Hub pipeline_tag fallback # ============================================================================= -_IS_LOCAL = "winml.modelkit.utils.hub_utils._is_local_path" -_HF_API = "huggingface_hub.HfApi" - - -def _fake_model_info(pipeline_tag: str | None): - """Return a mock model_info object with a ``pipeline_tag`` attribute.""" - return type("FakeModelInfo", (), {"pipeline_tag": pipeline_tag})() +_GET_PIPELINE_TAG = "winml.modelkit.utils.hub_utils.get_pipeline_tag" def test_resolve_task_uses_pipeline_tag_when_architecture_fails() -> None: """When config.architectures contains a generic name (e.g. 'Model') that TasksManager cannot resolve, the Hub pipeline_tag is used as fallback.""" cfg = _FakeConfig("faketype", name_or_path="audeering/wav2vec2-large-robust-24-ft-age-gender") - mock_api = type( - "MockApi", - (), - {"model_info": lambda self, _: _fake_model_info("audio-classification")}, - )() with ( patch(_INFER, side_effect=ValueError("unknown arch")), - patch(_IS_LOCAL, return_value=False), - patch(_HF_API, return_value=mock_api), + patch(_GET_PIPELINE_TAG, return_value="audio-classification"), ): r = resolve_task(cfg) assert r.task == "audio-classification" @@ -265,44 +253,33 @@ def test_resolve_task_uses_pipeline_tag_when_architecture_fails() -> None: def test_resolve_task_pipeline_tag_skips_invalid_task() -> None: """When pipeline_tag is not a recognized task, falls through to last-resort default.""" cfg = _FakeConfig("faketype", name_or_path="someone/some-model") - mock_api = type( - "MockApi", - (), - {"model_info": lambda self, _: _fake_model_info("not-a-real-task")}, - )() with ( patch(_INFER, side_effect=ValueError("unknown arch")), - patch(_IS_LOCAL, return_value=False), - patch(_HF_API, return_value=mock_api), + patch(_GET_PIPELINE_TAG, return_value="not-a-real-task"), ): r = resolve_task(cfg) assert r.source == TaskSource.HF_TASK_DEFAULT def test_resolve_task_pipeline_tag_handles_api_failure() -> None: - """When the Hub API call fails (network error, etc.), falls through gracefully.""" + """When the Hub API call fails (network error, etc.), get_pipeline_tag returns None.""" cfg = _FakeConfig("faketype", name_or_path="someone/some-model") - mock_api = type( - "MockApi", - (), - {"model_info": lambda self, _: (_ for _ in ()).throw(ConnectionError("offline"))}, - )() with ( patch(_INFER, side_effect=ValueError("unknown arch")), - patch(_IS_LOCAL, return_value=False), - patch(_HF_API, return_value=mock_api), + patch(_GET_PIPELINE_TAG, return_value=None), ): r = resolve_task(cfg) assert r.source == TaskSource.HF_TASK_DEFAULT def test_resolve_task_pipeline_tag_skips_local_path() -> None: - """When model_id is a local path, the Hub API is not called.""" + """When model_id is a local path, get_pipeline_tag is still called but returns None + (local-path rejection is internal to get_pipeline_tag).""" cfg = _FakeConfig("faketype", name_or_path="./local-model") with ( patch(_INFER, side_effect=ValueError("unknown arch")), - patch(_IS_LOCAL, return_value=True), - patch(_HF_API, side_effect=AssertionError("must not call HfApi for local paths")), + patch(_GET_PIPELINE_TAG, return_value=None) as mock_tag, ): r = resolve_task(cfg) + mock_tag.assert_called_once_with("./local-model") assert r.source == TaskSource.HF_TASK_DEFAULT From bb2b39446b1300e2ac268801779cb1248257ca58 Mon Sep 17 00:00:00 2001 From: Hualiang Xie Date: Fri, 17 Jul 2026 11:01:08 +0800 Subject: [PATCH 4/4] =?UTF-8?q?refactor:=20address=20review=20=E2=80=94=20?= =?UTF-8?q?exportable-task=20gate,=20lower=20timeout,=20label=20fix?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Gate Stage 1d pipeline_tag on the model-type's ONNX-exportable set (get_supported_tasks) instead of the full KNOWN_TASKS display taxonomy, so a non-exportable pipeline label (text-to-image, reinforcement-learning, ...) degrades to the last-resort default instead of failing at Stage 2/3. - Lower _PIPELINE_TAG_TIMEOUT from 10s to 5s (best-effort fallback shouldn't block long on unreachable/slow networks). - Rename inline stage labels so they match execution order: pipeline_tag is now 1d and last-resort default is 1e (was mislabeled 1e/1d). - Export get_pipeline_tag from utils package (public API). - Add focused unit tests for get_pipeline_tag itself (local-path skip without network, tag extraction, missing tag, API-error fallthrough). - Update Stage 1d resolver tests for the exportable-set gate. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- src/winml/modelkit/loader/resolution.py | 22 +++++++--- src/winml/modelkit/utils/__init__.py | 2 + src/winml/modelkit/utils/hub_utils.py | 2 +- tests/unit/loader/test_detect_task.py | 16 ++++--- tests/unit/utils/test_hub_utils.py | 56 +++++++++++++++++++++++++ 5 files changed, 86 insertions(+), 12 deletions(-) create mode 100644 tests/unit/utils/test_hub_utils.py diff --git a/src/winml/modelkit/loader/resolution.py b/src/winml/modelkit/loader/resolution.py index 8f9425d80..7759e78d2 100644 --- a/src/winml/modelkit/loader/resolution.py +++ b/src/winml/modelkit/loader/resolution.py @@ -24,7 +24,6 @@ from .task import ( HF_TASK_DEFAULTS, - KNOWN_TASKS, get_default_task_for_model_id, get_supported_tasks, normalize_task, @@ -556,7 +555,7 @@ def resolve_task( if opt_task is None and not getattr(config, "architectures", None) and model_type: # Populate Optimum's ONNX export-config registry before querying it; # get_supported_tasks returns [] if this hasn't been imported. - import optimum.exporters.onnx.model_configs # noqa: F401 + import optimum.exporters.onnx.model_configs supported = get_supported_tasks(model_type, resolve_optimum_library(model_type)) if supported: @@ -574,18 +573,29 @@ def resolve_task( except ValueError: opt_task = None - # 1e. Hub pipeline_tag fallback - if opt_task is None and model_id: + # 1d. Hub pipeline_tag fallback + if opt_task is None and model_id and model_type: from ..utils.hub_utils import get_pipeline_tag tag = get_pipeline_tag(model_id) if tag: normalized_tag = normalize_task(tag) - if normalized_tag in KNOWN_TASKS: + # Gate on the model-type's ONNX-exportable set, NOT the full KNOWN_TASKS + # display taxonomy. A Hub pipeline_tag is a HuggingFace *pipeline* label and + # may name a task with no export path (e.g. text-to-image, + # reinforcement-learning, time-series-forecasting). Admitting one would flow a + # non-exportable task into Stage 2 (model-class) / Stage 3 instead of degrading + # to the last-resort default. Populate Optimum's ONNX export-config registry + # first (as Stage 1b does) so get_supported_tasks doesn't return []. + import optimum.exporters.onnx.model_configs # noqa: F401 + + if normalized_tag in get_supported_tasks( + model_type, resolve_optimum_library(model_type) + ): opt_task = normalized_tag source = TaskSource.PIPELINE_TAG - # 1d. last-resort default + # 1e. last-resort default if opt_task is None: opt_task = next(iter(HF_TASK_DEFAULTS)) source = TaskSource.HF_TASK_DEFAULT diff --git a/src/winml/modelkit/utils/__init__.py b/src/winml/modelkit/utils/__init__.py index 10a2c420a..9aef1b73a 100644 --- a/src/winml/modelkit/utils/__init__.py +++ b/src/winml/modelkit/utils/__init__.py @@ -7,6 +7,7 @@ from .config_utils import merge_config from .constants import normalize_ep_name from .hub_utils import ( + get_pipeline_tag, inject_hub_metadata, is_hub_model, load_hf_components_from_onnx, @@ -28,6 +29,7 @@ "ModelInputKind", "WinMLManifest", "classify_model_input", + "get_pipeline_tag", "inject_hub_metadata", "is_hub_model", "load_hf_components_from_onnx", diff --git a/src/winml/modelkit/utils/hub_utils.py b/src/winml/modelkit/utils/hub_utils.py index a62876d98..9ade503e6 100644 --- a/src/winml/modelkit/utils/hub_utils.py +++ b/src/winml/modelkit/utils/hub_utils.py @@ -144,7 +144,7 @@ def is_hub_model(model_name_or_path: str) -> tuple[bool, dict]: return False, {"type": "local", "path": model_name_or_path} -_PIPELINE_TAG_TIMEOUT = 10 # seconds +_PIPELINE_TAG_TIMEOUT = 5 # seconds def get_pipeline_tag(model_id: str) -> str | None: diff --git a/tests/unit/loader/test_detect_task.py b/tests/unit/loader/test_detect_task.py index 8b8745488..bdb0046a9 100644 --- a/tests/unit/loader/test_detect_task.py +++ b/tests/unit/loader/test_detect_task.py @@ -231,31 +231,37 @@ def test_resolve_task_case1_surfaces_modality_aware_task() -> None: # ============================================================================= -# Stage 1e — Hub pipeline_tag fallback +# Stage 1d — Hub pipeline_tag fallback # ============================================================================= _GET_PIPELINE_TAG = "winml.modelkit.utils.hub_utils.get_pipeline_tag" +_GET_SUPPORTED = "winml.modelkit.loader.resolution.get_supported_tasks" def test_resolve_task_uses_pipeline_tag_when_architecture_fails() -> None: """When config.architectures contains a generic name (e.g. 'Model') that - TasksManager cannot resolve, the Hub pipeline_tag is used as fallback.""" + TasksManager cannot resolve, an exportable Hub pipeline_tag is used as fallback.""" cfg = _FakeConfig("faketype", name_or_path="audeering/wav2vec2-large-robust-24-ft-age-gender") with ( patch(_INFER, side_effect=ValueError("unknown arch")), patch(_GET_PIPELINE_TAG, return_value="audio-classification"), + patch(_GET_SUPPORTED, return_value=["feature-extraction", "audio-classification"]), ): r = resolve_task(cfg) assert r.task == "audio-classification" assert r.source == TaskSource.PIPELINE_TAG -def test_resolve_task_pipeline_tag_skips_invalid_task() -> None: - """When pipeline_tag is not a recognized task, falls through to last-resort default.""" +def test_resolve_task_pipeline_tag_skips_non_exportable_task() -> None: + """A pipeline_tag that is not in the model-type's ONNX-exportable set (e.g. a + HuggingFace pipeline label like text-to-image with no export path) is rejected and + resolution falls through to the last-resort default rather than surfacing a task + Stage 2 can't build.""" cfg = _FakeConfig("faketype", name_or_path="someone/some-model") with ( patch(_INFER, side_effect=ValueError("unknown arch")), - patch(_GET_PIPELINE_TAG, return_value="not-a-real-task"), + patch(_GET_PIPELINE_TAG, return_value="text-to-image"), + patch(_GET_SUPPORTED, return_value=["feature-extraction", "audio-classification"]), ): r = resolve_task(cfg) assert r.source == TaskSource.HF_TASK_DEFAULT diff --git a/tests/unit/utils/test_hub_utils.py b/tests/unit/utils/test_hub_utils.py new file mode 100644 index 000000000..80db7bb70 --- /dev/null +++ b/tests/unit/utils/test_hub_utils.py @@ -0,0 +1,56 @@ +# ------------------------------------------------------------------------- +# Copyright (c) Microsoft Corporation. All rights reserved. +# Licensed under the MIT License. +# -------------------------------------------------------------------------- +"""Tests for winml.modelkit.utils.hub_utils. + +Focused coverage of :func:`get_pipeline_tag` — the lightweight Hub helper used +by the Stage 1d ``pipeline_tag`` fallback in loader task resolution. These tests +exercise the helper directly (local-path short-circuit, API failure, tag +extraction) rather than through a mocked stand-in, so the real ``_is_local_path`` +guard and ``except`` fallthrough are covered. +""" + +from __future__ import annotations + +from unittest.mock import MagicMock, patch + +from winml.modelkit.utils import get_pipeline_tag + + +_HF_API = "huggingface_hub.HfApi" + + +def test_get_pipeline_tag_local_path_returns_none_without_network() -> None: + """A local path is rejected up front — the Hub API is never constructed.""" + with patch(_HF_API) as mock_api: + assert get_pipeline_tag("./local-model") is None + mock_api.assert_not_called() + + +def test_get_pipeline_tag_returns_tag_for_hub_model() -> None: + """A reachable Hub model returns its pipeline_tag.""" + info = MagicMock(pipeline_tag="audio-classification") + api = MagicMock() + api.model_info.return_value = info + with patch(_HF_API, return_value=api): + assert get_pipeline_tag("audeering/wav2vec2-large-robust-24-ft-age-gender") == ( + "audio-classification" + ) + api.model_info.assert_called_once() + + +def test_get_pipeline_tag_none_when_model_has_no_tag() -> None: + """A model without a pipeline_tag yields None.""" + api = MagicMock() + api.model_info.return_value = MagicMock(pipeline_tag=None) + with patch(_HF_API, return_value=api): + assert get_pipeline_tag("someone/some-model") is None + + +def test_get_pipeline_tag_swallows_api_error_and_returns_none() -> None: + """A network/Hub error is caught — the helper returns None instead of raising.""" + api = MagicMock() + api.model_info.side_effect = ConnectionError("offline") + with patch(_HF_API, return_value=api): + assert get_pipeline_tag("someone/some-model") is None