`62c3366`

Cover base model edge branches

Authored by mfwolffe <wolffemf@dukes.jmu.edu> 2 weeks ago

SHA: 62c33663c932d9fc185d8450258add0067c8bf56
Parents: b73a078
Tree: 8b2d8c1

5 changed files

Status	File	+	-
M	`tests/unit/base_models/test_downloader.py`	22	0
M	`tests/unit/base_models/test_license.py`	25	0
M	`tests/unit/base_models/test_probes.py`	216	0
M	`tests/unit/base_models/test_registry_refresh.py`	119	1
M	`tests/unit/base_models/test_schema.py`	1	1

tests/unit/base_models/test_downloader.pymodified

  import pytest
  from dlm.base_models import BaseModelSpec, GatedModelError, download_spec, sha256_of_directory
 +from dlm.base_models.downloader import _resolve_revision
  def _spec() -> BaseModelSpec:
              pytest.raises(RuntimeError, match="offline"),
          ):
              download_spec(_spec(), local_files_only=True)
++
 +    def test_repository_not_found_raises_runtime_error(self) -> None:
 +        from unittest.mock import Mock
++
 +        from huggingface_hub.errors import RepositoryNotFoundError
++
 +        with (
 +            patch(
 +                "huggingface_hub.snapshot_download",
 +                side_effect=RepositoryNotFoundError("missing", response=Mock()),
 +            ),
 +            pytest.raises(RuntimeError, match="HF repository not found"),
 +        ):
 +            download_spec(_spec())
++
 +    def test_resolve_revision_falls_back_to_expected_outside_snapshot_layout(
 +        self, tmp_path: Path
 +    ) -> None:
 +        local_copy = tmp_path / "local-model"
 +        local_copy.mkdir()
 +        assert _resolve_revision(local_copy, "a" * 40) == "a" * 40

tests/unit/base_models/test_license.pymodified

  from dlm.base_models import BASE_MODELS, GatedModelError, LicenseAcceptance
  from dlm.base_models.license import is_gated, require_acceptance
 +from dlm.base_models.schema import BaseModelSpec
  def _non_gated_spec() -> object:
          assert acc is not None
          assert acc.via == via
 +    def test_gated_spec_without_license_url_fails_loudly(self) -> None:
 +        spec = BaseModelSpec.model_validate(
 +            {
 +                "key": "broken-gated",
 +                "hf_id": "org/broken",
 +                "revision": "0" * 40,
 +                "architecture": "DemoForCausalLM",
 +                "params": 1_000_000_000,
 +                "target_modules": ["q_proj"],
 +                "template": "chatml",
 +                "gguf_arch": "demo",
 +                "tokenizer_pre": "demo",
 +                "license_spdx": "Other",
 +                "license_url": None,
 +                "requires_acceptance": True,
 +                "redistributable": False,
 +                "size_gb_fp16": 2.0,
 +                "context_length": 4096,
 +                "recommended_seq_len": 2048,
 +            }
 +        )
 +        with pytest.raises(GatedModelError):
 +            require_acceptance(spec, accept_license=True, via="cli_flag")
++
  class TestAcceptanceModel:
      def test_frozen_rejects_mutation(self) -> None:

tests/unit/base_models/test_probes.pymodified

  from dlm.base_models.probes import (
      _LLAMA_CPP_CHKTXT,
      probe_architecture,
 +    probe_audio_token,
      probe_chat_template,
      probe_gguf_arch_supported,
      probe_pretokenizer_hash,
      probe_pretokenizer_label,
 +    probe_vl_image_token,
      run_all,
+ )
+     )
 +def _vl_spec() -> BaseModelSpec:
 +    return BaseModelSpec.model_validate(
 +        {
 +            **_spec().model_dump(),
 +            "key": "demo-vl",
 +            "modality": "vision-language",
 +            "vl_preprocessor_plan": {
 +                "target_size": [224, 224],
 +                "image_token": "<image>",
 +                "num_image_tokens": 256,
 +            },
 +        }
 +    )
++
++
 +def _audio_spec() -> BaseModelSpec:
 +    return BaseModelSpec.model_validate(
 +        {
 +            **_spec().model_dump(),
 +            "key": "demo-audio",
 +            "modality": "audio-language",
 +            "audio_preprocessor_plan": {
 +                "sample_rate": 16000,
 +                "audio_token": "<audio>",
 +                "num_audio_tokens": 64,
 +                "max_length_seconds": 30.0,
 +            },
 +        }
 +    )
++
++
  class TestProbeArchitecture:
      def test_matching_architectures_pass(self) -> None:
          fake_cfg = SimpleNamespace(architectures=["DemoForCausalLM"])
              result = probe_chat_template(_spec())
          assert result.passed is False
 +    def test_gated_repo_raises_gated_model_error(self) -> None:
 +        from unittest.mock import Mock
++
 +        from huggingface_hub.errors import GatedRepoError
++
 +        with (
 +            patch(
 +                "transformers.AutoTokenizer.from_pretrained",
 +                side_effect=GatedRepoError("gated", response=Mock()),
 +            ),
 +            pytest.raises(GatedModelError),
 +        ):
 +            probe_chat_template(_spec())
++
  class TestProbeGgufArch:
      def test_skips_when_vendor_missing(self, tmp_path: Path) -> None:
          result = probe_gguf_arch_supported(_spec(), vendor_path=vendor)
          assert result.passed is True
 +    def test_read_error_fails(self, tmp_path: Path) -> None:
 +        vendor = tmp_path / "llama.cpp"
 +        vendor.mkdir()
 +        script = vendor / "convert_hf_to_gguf.py"
 +        script.write_text('@Model.register("DemoForCausalLM")\n', encoding="utf-8")
 +        with patch.object(Path, "read_text", side_effect=OSError("boom")):
 +            result = probe_gguf_arch_supported(_spec(), vendor_path=vendor)
 +        assert result.passed is False
 +        assert "read failed" in result.detail
++
  class TestProbePretokenizerLabel:
      def test_skips_when_table_missing(self, tmp_path: Path) -> None:
          result = probe_pretokenizer_label(_spec(), hashes_path=hashes)
          assert result.passed is False
 +    def test_wrong_shape_table_fails(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
 +        hashes = tmp_path / "h.json"
 +        hashes.write_text("[]", encoding="utf-8")
 +        monkeypatch.setattr("dlm.base_models.probes.json.loads", lambda _text: [["nested"]])
 +        result = probe_pretokenizer_label(_spec(), hashes_path=hashes)
 +        assert result.passed is False
 +        assert "wrong shape" in result.detail
++
  class TestProbePretokenizerHash:
      """Audit-04 B8: real sha256-of-token-ids fingerprint check."""
          assert result.passed is False
          assert "wrong shape" in result.detail
 +    def test_tokenizer_encode_failure_fails(self, tmp_path: Path) -> None:
 +        table = tmp_path / "fp.json"
 +        table.write_text(json.dumps({"demo": "a" * 64}), encoding="utf-8")
++
 +        with patch(
 +            "transformers.AutoTokenizer.from_pretrained",
 +            return_value=SimpleNamespace(
 +                encode=lambda _text: (_ for _ in ()).throw(ValueError("boom"))
 +            ),
 +        ):
 +            result = probe_pretokenizer_hash(_spec(), fingerprints_path=table)
++
 +        assert result.passed is False
 +        assert "tokenizer.encode failed" in result.detail
++
++
 +class TestProbeVlImageToken:
 +    def test_non_vl_spec_skips(self) -> None:
 +        result = probe_vl_image_token(_spec())
 +        assert result.skipped is True
++
 +    def test_gated_processor_raises(self) -> None:
 +        from unittest.mock import Mock
++
 +        from huggingface_hub.errors import GatedRepoError
++
 +        with (
 +            patch(
 +                "dlm.base_models._typed_shims.load_auto_processor",
 +                side_effect=GatedRepoError("gated", response=Mock()),
 +            ),
 +            pytest.raises(GatedModelError),
 +        ):
 +            probe_vl_image_token(_vl_spec())
++
 +    def test_missing_tokenizer_fails(self) -> None:
 +        with patch(
 +            "dlm.base_models._typed_shims.load_auto_processor",
 +            return_value=SimpleNamespace(),
 +        ):
 +            result = probe_vl_image_token(_vl_spec())
 +        assert result.passed is False
 +        assert "no `.tokenizer`" in result.detail
++
 +    def test_tokenizer_encode_error_fails(self) -> None:
 +        tokenizer = SimpleNamespace(
 +            encode=lambda _placeholder, add_special_tokens=False: (_ for _ in ()).throw(
 +                ValueError("boom")
 +            )
 +        )
 +        with patch(
 +            "dlm.base_models._typed_shims.load_auto_processor",
 +            return_value=SimpleNamespace(tokenizer=tokenizer),
 +        ):
 +            result = probe_vl_image_token(_vl_spec())
 +        assert result.passed is False
 +        assert "tokenizer rejected placeholder" in result.detail
++
 +    def test_multi_token_placeholder_fails(self) -> None:
 +        tokenizer = SimpleNamespace(encode=lambda _placeholder, add_special_tokens=False: [1, 2])
 +        with patch(
 +            "dlm.base_models._typed_shims.load_auto_processor",
 +            return_value=SimpleNamespace(tokenizer=tokenizer),
 +        ):
 +            result = probe_vl_image_token(_vl_spec())
 +        assert result.passed is False
 +        assert "expected 1" in result.detail
++
++
 +class TestProbeAudioToken:
 +    def test_non_audio_spec_skips(self) -> None:
 +        result = probe_audio_token(_spec())
 +        assert result.skipped is True
++
 +    def test_gated_processor_raises(self) -> None:
 +        from unittest.mock import Mock
++
 +        from huggingface_hub.errors import GatedRepoError
++
 +        with (
 +            patch(
 +                "dlm.base_models._typed_shims.load_auto_processor",
 +                side_effect=GatedRepoError("gated", response=Mock()),
 +            ),
 +            pytest.raises(GatedModelError),
 +        ):
 +            probe_audio_token(_audio_spec())
++
 +    def test_missing_tokenizer_fails(self) -> None:
 +        with patch(
 +            "dlm.base_models._typed_shims.load_auto_processor",
 +            return_value=SimpleNamespace(),
 +        ):
 +            result = probe_audio_token(_audio_spec())
 +        assert result.passed is False
 +        assert "no `.tokenizer`" in result.detail
++
 +    def test_tokenizer_encode_error_fails(self) -> None:
 +        tokenizer = SimpleNamespace(
 +            encode=lambda _placeholder, add_special_tokens=False: (_ for _ in ()).throw(
 +                ValueError("boom")
 +            )
 +        )
 +        with patch(
 +            "dlm.base_models._typed_shims.load_auto_processor",
 +            return_value=SimpleNamespace(tokenizer=tokenizer),
 +        ):
 +            result = probe_audio_token(_audio_spec())
 +        assert result.passed is False
 +        assert "tokenizer rejected placeholder" in result.detail
++
 +    def test_multi_token_placeholder_fails(self) -> None:
 +        tokenizer = SimpleNamespace(encode=lambda _placeholder, add_special_tokens=False: [1, 2])
 +        with patch(
 +            "dlm.base_models._typed_shims.load_auto_processor",
 +            return_value=SimpleNamespace(tokenizer=tokenizer),
 +        ):
 +            result = probe_audio_token(_audio_spec())
 +        assert result.passed is False
 +        assert "expected 1" in result.detail
++
  class TestRunAll:
      def test_aggregates_all_five_probes(self) -> None:
              "pretokenizer_label",
              "pretokenizer_hash",
+         }
++
 +    def test_vl_run_all_uses_vl_probe_and_skips_export_checks(self) -> None:
 +        spec = _vl_spec()
 +        fake_cfg = SimpleNamespace(architectures=["DemoForCausalLM"])
 +        tokenizer = SimpleNamespace(encode=lambda _placeholder, add_special_tokens=False: [7])
 +        with (
 +            patch("transformers.AutoConfig.from_pretrained", return_value=fake_cfg),
 +            patch(
 +                "dlm.base_models._typed_shims.load_auto_processor",
 +                return_value=SimpleNamespace(tokenizer=tokenizer),
 +            ),
 +        ):
 +            report = run_all(spec)
 +        names = {r.name for r in report.results}
 +        assert names == {"architecture", "vl_image_token"}
++
 +    def test_audio_run_all_uses_audio_probe_and_skips_export_checks(self) -> None:
 +        spec = _audio_spec()
 +        fake_cfg = SimpleNamespace(architectures=["DemoForCausalLM"])
 +        tokenizer = SimpleNamespace(encode=lambda _placeholder, add_special_tokens=False: [9])
 +        with (
 +            patch("transformers.AutoConfig.from_pretrained", return_value=fake_cfg),
 +            patch(
 +                "dlm.base_models._typed_shims.load_auto_processor",
 +                return_value=SimpleNamespace(tokenizer=tokenizer),
 +            ),
 +        ):
 +            report = run_all(spec)
 +        names = {r.name for r in report.results}
 +        assert names == {"architecture", "audio_token"}

tests/unit/base_models/test_registry_refresh.pymodified

  from __future__ import annotations
  from types import SimpleNamespace
 +from urllib.error import HTTPError
 -from dlm.base_models.registry_refresh import Drift, check_entry
 +import pytest
++
 +from dlm.base_models.registry_refresh import Drift, check_entry, check_registry, fetch_text
  from dlm.base_models.schema import BaseModelSpec
          return self._info
 +class _RaisingApi:
 +    def __init__(self, exc: Exception) -> None:
 +        self._exc = exc
++
 +    def model_info(self, _hf_id: str) -> SimpleNamespace:
 +        raise self._exc
++
++
 +class _FakeResponse:
 +    def __init__(self, body: bytes, charset: str = "utf-8") -> None:
 +        self._body = body
 +        self.headers = SimpleNamespace(get_content_charset=lambda: charset)
++
 +    def __enter__(self) -> _FakeResponse:
 +        return self
++
 +    def __exit__(self, exc_type: object, exc: object, tb: object) -> bool:
 +        return False
++
 +    def read(self) -> bytes:
 +        return self._body
++
++
 +class TestDrift:
 +    def test_render_formats_each_field_on_its_own_line(self) -> None:
 +        drift = Drift(
 +            key="demo-1b",
 +            hf_id="org/demo-1b",
 +            fields=(("revision", "old", "new"), ("gating", "False", "True")),
 +        )
 +        assert drift.render() == (
 +            "  demo-1b (org/demo-1b)\n"
 +            "    revision               'old' → 'new'\n"
 +            "    gating                 'False' → 'True'"
 +        )
++
++
 +class TestFetchText:
 +    def test_fetch_text_decodes_response_body(self, monkeypatch: pytest.MonkeyPatch) -> None:
 +        monkeypatch.setattr(
 +            "dlm.base_models.registry_refresh.urlopen",
 +            lambda req, timeout: _FakeResponse("olá".encode()),
 +        )
 +        assert fetch_text("https://example.com") == "olá"
++
++
  class TestCheckEntry:
      def test_no_drift_when_revision_and_gating_match(self) -> None:
          spec = _spec()
              "official marker",
              "missing from https://example.com/provenance",
          ) in drift.fields
++
 +    def test_gated_repo_is_reported_as_drift(self) -> None:
 +        from unittest.mock import Mock
++
 +        from huggingface_hub.errors import GatedRepoError
++
 +        drift = check_entry(
 +            _RaisingApi(GatedRepoError("gated", response=Mock())),
 +            _spec(),
 +        )
 +        assert isinstance(drift, Drift)
 +        assert ("gating", "readable", "now fully gated") in drift.fields
++
 +    def test_missing_repository_is_reported_as_drift(self) -> None:
 +        from unittest.mock import Mock
++
 +        from huggingface_hub.errors import RepositoryNotFoundError
++
 +        drift = check_entry(
 +            _RaisingApi(RepositoryNotFoundError("missing", response=Mock())),
 +            _spec(),
 +        )
 +        assert isinstance(drift, Drift)
 +        assert ("repository", "present", "missing (renamed or deleted)") in drift.fields
++
 +    def test_gating_mismatch_is_reported_when_enabled(self) -> None:
 +        spec = _spec(requires_acceptance=False)
 +        drift = check_entry(_Api(sha=spec.revision, gated=True), spec)
 +        assert isinstance(drift, Drift)
 +        assert ("requires_acceptance", "False", "True") in drift.fields
++
 +    def test_unreachable_provenance_url_is_reported(self) -> None:
 +        spec = _spec(
 +            refresh_check_hf_gating=False,
 +            provenance_url="https://example.com/provenance",
 +            provenance_match_text="official marker",
 +        )
 +        drift = check_entry(
 +            _Api(sha=spec.revision),
 +            spec,
 +            fetch_url_text=lambda _url: (_ for _ in ()).throw(
 +                HTTPError(_url, 404, "missing", hdrs=None, fp=None)
 +            ),
 +        )
 +        assert isinstance(drift, Drift)
 +        assert drift.fields[0][0] == "provenance_url"
 +        assert "unreachable" in drift.fields[0][2]
++
++
 +class TestCheckRegistry:
 +    def test_check_registry_collects_non_null_drifts(self, monkeypatch: pytest.MonkeyPatch) -> None:
 +        entries = {
 +            "one": _spec(key="one", hf_id="org/one"),
 +            "two": _spec(key="two", hf_id="org/two"),
 +        }
 +        monkeypatch.setattr("dlm.base_models.registry_refresh.BASE_MODELS", entries)
 +        monkeypatch.setattr("dlm.base_models.registry_refresh.HfApi", lambda: object())
++
 +        def _fake_check_entry(
 +            api: object, entry: BaseModelSpec, *, fetch_url_text: object
 +        ) -> Drift | None:
 +            if entry.key == "one":
 +                return Drift(key="one", hf_id=entry.hf_id, fields=(("revision", "old", "new"),))
 +            return None
++
 +        monkeypatch.setattr("dlm.base_models.registry_refresh.check_entry", _fake_check_entry)
 +        drifts = check_registry()
 +        assert len(drifts) == 1
 +        assert drifts[0].key == "one"

tests/unit/base_models/test_schema.pymodified

  class TestHfIdValidator:
 -    @pytest.mark.parametrize("bad_id", ["", "no-slash", "trailing/"])
 +    @pytest.mark.parametrize("bad_id", ["", "no-slash", "trailing/", "org/name/extra"])
      def test_invalid_hf_id_rejected(self, bad_id: str) -> None:
          # `trailing/` passes the `/ in value` gate but leading slash doesn't.
          # Pydantic's min_length catches empty. Bad ones without `/` raise.