| """The entry file must hydrate its own pinned snapshot, including for forks.""" |
|
|
| from pathlib import Path |
| import runpy |
| import tempfile |
| from types import SimpleNamespace |
| import unittest |
| from unittest.mock import Mock, patch |
|
|
|
|
| class HubLoadingTests(unittest.TestCase): |
| def test_fork_revision_and_cache_are_preserved(self): |
| source = Path(__file__).resolve().parents[1] / "env.py" |
| with tempfile.TemporaryDirectory() as temporary: |
| cache = Path(temporary) |
| revision = "a" * 40 |
| snapshot = cache / "models--owner--g1-custom" / "snapshots" / revision |
| snapshot.mkdir(parents=True) |
| entry = snapshot / "env.py" |
| entry.write_bytes(source.read_bytes()) |
| factory = SimpleNamespace(SimulatorFactory=Mock(), init_channel=Mock()) |
| with ( |
| patch("huggingface_hub.snapshot_download") as download, |
| patch.dict("sys.modules", {"sim.simulator_factory": factory}), |
| patch("sys.path", list(__import__("sys").path)), |
| ): |
| runpy.run_path(str(entry)) |
| download.assert_called_once_with("owner/g1-custom", revision=revision, cache_dir=str(cache)) |
|
|
| def test_incomplete_local_checkout_fails_without_remote_fallback(self): |
| source = Path(__file__).resolve().parents[1] / "env.py" |
| with tempfile.TemporaryDirectory() as temporary: |
| entry = Path(temporary) / "env.py" |
| entry.write_bytes(source.read_bytes()) |
| with patch("huggingface_hub.snapshot_download") as download: |
| with self.assertRaisesRegex(FileNotFoundError, "Incomplete simulator checkout"): |
| runpy.run_path(str(entry)) |
| download.assert_not_called() |
|
|
|
|
| if __name__ == "__main__": |
| unittest.main() |
|
|
