test(kompress): close patch-coverage gaps from #2716 (#2721)

## Description

Codecov flagged 9 uncovered lines on #2716 after it merged:
`hf_entry_known_absent`'s own body
in `headroom/onnx_runtime.py` was only ever exercised indirectly (every
existing test in
`tests/test_transforms/test_kompress_compressor.py` monkeypatched it
away rather than calling the
real implementation), and `_load_pytorch_weights` /
`_load_kompress_pytorch` in
`headroom/transforms/kompress_compressor.py` had three untested
branches: the double cache-miss
under `allow_download=False` (merged.pt confirmed absent AND the plain
fallback also not cached),
a genuine non-404 download failure propagating instead of silently
falling back, and the
already-cached fast path in `_load_kompress_pytorch`.

## Type of Change

- [ ] Bug fix
- [ ] New feature
- [x] Test coverage improvement, no production code change

## Changes Made

- `tests/test_onnx_runtime.py`: added `_write_fake_hf_cache` (builds a
minimal on-disk HF hub
cache layout, including the `.no_exist/<hash>/<filename>` marker
huggingface_hub writes after a
real 404) and three direct tests of `hf_entry_known_absent` against the
real
  `huggingface_hub.try_to_load_from_cache`, not a mock of it.
- `tests/test_transforms/test_kompress_compressor.py`: added
  `test_cache_only_raises_when_confirmed_absent_but_plain_also_missing`,
`test_genuine_download_failure_propagates_instead_of_falling_back`, and
a new
`TestLoadKompressPytorchCaching` class covering the already-cached fast
path.

## Testing

```text
$ .venv/bin/python3 -m pytest tests/test_onnx_runtime.py tests/test_transforms/test_kompress_compressor.py -q
51 passed

$ .venv/bin/python3 -m pytest tests/ -k "kompress or onnx_runtime" -q --cov=headroom.transforms.kompress_compressor --cov=headroom.onnx_runtime --cov-report=term-missing
# before: onnx_runtime.py Missing includes 132-136 (hf_entry_known_absent's entire body);
#         kompress_compressor.py Missing includes 805-806, 818, 836
# after:  none of those lines appear in Missing anymore
191 passed, 7 skipped

$ .venv/bin/python3 -m ruff format --check tests/test_onnx_runtime.py tests/test_transforms/test_kompress_compressor.py
2 files already formatted
$ .venv/bin/python3 -m ruff check tests/test_onnx_runtime.py tests/test_transforms/test_kompress_compressor.py
All checks passed!
```

## Review Readiness

- Test-only, additive diff (113 insertions, 0 deletions, 0 lines touched
outside the two test
  files). No behavior change possible.

## Checklist

- [x] My code follows the project's style guidelines
- [x] I have performed a self-review of my code
- [x] New and existing unit tests pass locally with my changes
- [x] I did not edit `CHANGELOG.md`

## Additional Notes

Not closing: the remaining branch-partial on the `device == "auto"`
cuda/mps/cpu selection in
`_load_kompress_pytorch` (would need mocking `torch.cuda.is_available()`
/
`torch.backends.mps.is_available()` for marginal benefit); left as-is.
This commit is contained in:
Raúl 2026-08-02 19:45:44 +02:00 committed by GitHub
parent 2797099bec
commit 1a2688b57f
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
2 changed files with 113 additions and 0 deletions

View file

@ -1,9 +1,11 @@
import os
import sys
from headroom.onnx_runtime import (
ONNX_CPU_ARENA_ENV,
cpu_arena_enabled,
create_cpu_session_options,
hf_entry_known_absent,
)
@ -104,3 +106,55 @@ def test_create_cpu_session_options_handles_older_session_options(monkeypatch):
assert options.intra_op_num_threads is None
assert options.inter_op_num_threads is None
def _write_fake_hf_cache(
root: str, repo_id: str, revision: str, *, no_exist_files: list[str]
) -> None:
"""Build a minimal on-disk HF hub cache layout for a single repo/revision.
Mirrors the real cache structure closely enough for
``huggingface_hub.try_to_load_from_cache`` to read it: a ``refs/<name>``
pointer file, a ``snapshots/<hash>`` directory, and a
``.no_exist/<hash>/<filename>`` marker per file whose absence is cached.
"""
from huggingface_hub.file_download import repo_folder_name
repo_folder = os.path.join(root, repo_folder_name(repo_id=repo_id, repo_type="model"))
os.makedirs(os.path.join(repo_folder, "refs"), exist_ok=True)
with open(os.path.join(repo_folder, "refs", revision), "w") as f:
f.write("abc123")
os.makedirs(os.path.join(repo_folder, "snapshots", "abc123"), exist_ok=True)
no_exist_dir = os.path.join(repo_folder, ".no_exist", "abc123")
os.makedirs(no_exist_dir, exist_ok=True)
for filename in no_exist_files:
open(os.path.join(no_exist_dir, filename), "w").close()
def test_hf_entry_known_absent_true_when_404_was_cached(tmp_path, monkeypatch):
from huggingface_hub import constants
_write_fake_hf_cache(str(tmp_path), "acme/widget", "main", no_exist_files=["merged.pt"])
monkeypatch.setattr(constants, "HF_HUB_CACHE", str(tmp_path))
monkeypatch.delenv("HEADROOM_HF_PIN", raising=False)
assert hf_entry_known_absent("acme/widget", "merged.pt") is True
def test_hf_entry_known_absent_false_when_never_checked(tmp_path, monkeypatch):
from huggingface_hub import constants
_write_fake_hf_cache(str(tmp_path), "acme/widget", "main", no_exist_files=[])
monkeypatch.setattr(constants, "HF_HUB_CACHE", str(tmp_path))
monkeypatch.delenv("HEADROOM_HF_PIN", raising=False)
assert hf_entry_known_absent("acme/widget", "merged.pt") is False
def test_hf_entry_known_absent_false_when_repo_not_cached_at_all(tmp_path, monkeypatch):
from huggingface_hub import constants
monkeypatch.setattr(constants, "HF_HUB_CACHE", str(tmp_path))
monkeypatch.delenv("HEADROOM_HF_PIN", raising=False)
assert hf_entry_known_absent("nobody/nothing", "merged.pt") is False

View file

@ -753,3 +753,62 @@ class TestPytorchWeightLoading:
for name, param in model.state_dict().items():
assert torch.equal(param, fresh_model.state_dict()[name])
def test_cache_only_raises_when_confirmed_absent_but_plain_also_missing(
self, monkeypatch
) -> None:
"""merged.pt confirmed absent, but the plain fallback isn't cached either:
still nothing to load from, so this must defer rather than raise a
confusing lower-level error.
"""
import pytest
pytest.importorskip("torch")
import headroom.transforms.kompress_compressor as kmod
def fake_download(model_id, filename, *, allow_network=True, **kwargs): # noqa: ANN001
raise kmod.LocalEntryNotFoundError(f"{filename} not cached")
monkeypatch.setattr(kmod, "hf_hub_download_local_first", fake_download)
monkeypatch.setattr(kmod, "hf_entry_known_absent", lambda *a, **k: True)
with pytest.raises(kmod.KompressModelNotCached):
kmod._load_pytorch_weights(
SimpleNamespace(), "chopratejas/kompress-base", allow_download=False
)
def test_genuine_download_failure_propagates_instead_of_falling_back(self, monkeypatch) -> None:
"""A real network/download failure (not a 404, not a cache miss under
allow_download=False) must propagate as-is, not be swallowed into a
silent fallback to the plain format.
"""
import pytest
pytest.importorskip("torch")
import headroom.transforms.kompress_compressor as kmod
def fake_download(model_id, filename, *, allow_network=True, **kwargs): # noqa: ANN001
raise OSError("connection reset")
monkeypatch.setattr(kmod, "hf_hub_download_local_first", fake_download)
with pytest.raises(OSError, match="connection reset"):
kmod._load_pytorch_weights(SimpleNamespace(), "some/repo", allow_download=True)
class TestLoadKompressPytorchCaching:
def test_returns_cached_entry_without_reloading(self, monkeypatch) -> None:
import pytest
pytest.importorskip("torch")
import headroom.transforms.kompress_compressor as kmod
sentinel = ("cached-model", "cached-tokenizer", "pytorch")
monkeypatch.setattr(kmod, "_kompress_cache", {"some/repo": sentinel})
def boom(*a, **k): # noqa: ANN001, ANN202
raise AssertionError("should not attempt to reload an already-cached model")
monkeypatch.setattr(kmod, "_load_pytorch_weights", boom)
assert kmod._load_kompress_pytorch("some/repo") == sentinel