Spaces:
Running
Running
Download tests/test_huggingface_surface.py from noqt/eggcracker: direct link, hf CLI and curl.
- Browser
- Download file 17.1 kB
-
https://huggingface.co/spaces/noqt/eggcracker/resolve/main/tests/test_huggingface_surface.py
- Command line
-
hf download hf://spaces/noqt/eggcracker/tests/test_huggingface_surface.py
-
curl -L -o test_huggingface_surface.py https://huggingface.co/spaces/noqt/eggcracker/resolve/main/tests/test_huggingface_surface.py
17.1 kB
| from __future__ import annotations | |
| import hashlib | |
| import json | |
| import os | |
| import subprocess | |
| import sys | |
| import tempfile | |
| import unittest | |
| from pathlib import Path, PurePosixPath | |
| from types import SimpleNamespace | |
| from unittest.mock import Mock, patch | |
| ROOT = Path(__file__).resolve().parents[1] | |
| sys.path.insert(0, str(ROOT / "scripts")) | |
| from build_huggingface_surface import ( | |
| BOUND_RELEASE, | |
| SurfaceBuildError, | |
| _policy_root_file, | |
| _require_unique_paths, | |
| _tree_path, | |
| _validated_release_reference, | |
| build_surface, | |
| ) | |
| from upload_huggingface_surface import ( | |
| LEGACY_BOOTSTRAP_PARENT, | |
| LEGACY_BOOTSTRAP_SOURCE, | |
| UploadError, | |
| upload_surface, | |
| ) | |
| from verify_huggingface_surface import VerificationError, verify_surface | |
| class HuggingFaceSurfaceTest(unittest.TestCase): | |
| def _minimal_surface(self, root: Path, revision: str) -> tuple[Path, Path, Path]: | |
| staging = root / "surface" | |
| staging.mkdir() | |
| payload = b"fixture\n" | |
| (staging / "README.md").write_bytes(payload) | |
| manifest = { | |
| "schema": "noqt.huggingface_manifest.v1", | |
| "source_repository": "https://github.com/noqt/Lumi-Eggcracker", | |
| "source_revision": revision, | |
| "source_url": f"https://github.com/noqt/Lumi-Eggcracker/commit/{revision}", | |
| "release_reference": {}, | |
| "files": {"README.md": {"sha256": hashlib.sha256(payload).hexdigest(), "size": len(payload)}}, | |
| } | |
| manifest_bytes = (json.dumps(manifest, indent=2, sort_keys=True) + "\n").encode() | |
| (staging / "HUGGINGFACE_MANIFEST.json").write_bytes(manifest_bytes) | |
| marker = { | |
| "schema": "noqt.huggingface_sync.v1", | |
| "mode": "DERIVED_READ_ONLY_MIRROR", | |
| "source_repository": manifest["source_repository"], | |
| "source_revision": revision, | |
| "source_url": manifest["source_url"], | |
| "release_reference": {}, | |
| "target": "https://huggingface.co/spaces/noqt/eggcracker", | |
| "manifest": "HUGGINGFACE_MANIFEST.json", | |
| "manifest_sha256": hashlib.sha256(manifest_bytes).hexdigest(), | |
| "mirrored_files": 1, | |
| } | |
| (staging / "HUGGINGFACE_SYNC.json").write_text(json.dumps(marker), encoding="utf-8") | |
| upload_record = root / "upload.json" | |
| upload_record.write_text( | |
| json.dumps( | |
| { | |
| "schema": "noqt.huggingface_upload.v1", | |
| "repository": "noqt/eggcracker", | |
| "repo_type": "space", | |
| "branch": "main", | |
| "source_revision": revision, | |
| "parent_commit": "1" * 40, | |
| "upload_commit": "2" * 40, | |
| "manifest_sha256": marker["manifest_sha256"], | |
| } | |
| ), | |
| encoding="utf-8", | |
| ) | |
| return staging, upload_record, root / "report.json" | |
| def test_rejects_nonportable_or_noncanonical_git_paths(self) -> None: | |
| for raw_path in ( | |
| b"../escape", | |
| b"folder\\escape", | |
| b"folder//escape", | |
| b"./escape", | |
| b"C:escape", | |
| b"C:/escape", | |
| b"file:stream", | |
| b"CON", | |
| b"folder/aux.txt", | |
| b"folder/LPT1.log", | |
| b"trailing.", | |
| b"trailing ", | |
| ): | |
| with ( | |
| self.subTest(raw_path=raw_path), | |
| self.assertRaisesRegex(SurfaceBuildError, "unsafe or non-portable"), | |
| ): | |
| _tree_path(raw_path) | |
| with self.assertRaisesRegex(SurfaceBuildError, "portable path collision"): | |
| _require_unique_paths( | |
| [PurePosixPath("README.md"), PurePosixPath("readme.md")], | |
| "test paths", | |
| ) | |
| def test_policy_outputs_are_distinct_portable_root_files(self) -> None: | |
| for value in ("../marker.json", "nested/marker.json", "C:marker.json", "marker.json:stream", "NUL"): | |
| with self.subTest(value=value), self.assertRaises(SurfaceBuildError): | |
| _policy_root_file(value, "source_marker") | |
| marker = _policy_root_file("HUGGINGFACE_SYNC.json", "source_marker") | |
| manifest = _policy_root_file("HUGGINGFACE_MANIFEST.json", "manifest") | |
| _require_unique_paths([marker, manifest], "sync policy outputs") | |
| def test_policy_binds_release_boundary_and_only_authorized_triggers(self) -> None: | |
| policy = json.loads((ROOT / "huggingface-sync-policy.json").read_text(encoding="utf-8")) | |
| self.assertEqual( | |
| {"PUSH_TO_MAIN", "RELEASE_PUBLISHED", "WORKFLOW_DISPATCH"}, | |
| set(policy["sync_triggers"]), | |
| ) | |
| self.assertNotIn("DAILY_DRIFT_REPAIR", policy["sync_triggers"]) | |
| self.assertEqual(BOUND_RELEASE["source_commit"], policy["release_reference"]["source_commit"]) | |
| self.assertEqual( | |
| BOUND_RELEASE["release_manifest_sha256"], | |
| policy["release_reference"]["release_manifest"]["sha256"], | |
| ) | |
| self.assertEqual(policy["release_reference"], _validated_release_reference(policy)) | |
| workflow = (ROOT / ".github" / "workflows" / "sync-huggingface.yml").read_text(encoding="utf-8") | |
| self.assertNotIn("schedule:", workflow) | |
| self.assertIn("release:", workflow) | |
| self.assertIn("types: [published]", workflow) | |
| self.assertIn("ref: ${{ github.event_name == 'release' && 'main' || github.sha }}", workflow) | |
| self.assertIn("RELEASE_TAG: ${{ github.event.release.tag_name }}", workflow) | |
| self.assertIn("refs/tags/$RELEASE_TAG^{tag}", workflow) | |
| self.assertNotIn("source_ref='${{ github.event.release.tag_name }}'", workflow) | |
| self.assertIn("upload_huggingface_surface.py", workflow) | |
| self.assertIn("verify_huggingface_surface.py", workflow) | |
| self.assertIn("--source-root .", workflow) | |
| self.assertIn("parent_commit", (ROOT / "scripts" / "upload_huggingface_surface.py").read_text(encoding="utf-8")) | |
| def test_upload_and_readback_fail_closed_without_hub_auth(self) -> None: | |
| revision = "a" * 40 | |
| with tempfile.TemporaryDirectory() as temp_dir: | |
| staging, upload_record, report = self._minimal_surface(Path(temp_dir), revision) | |
| with patch.dict(os.environ, {"HF_TOKEN": ""}): | |
| with self.assertRaisesRegex(UploadError, "HF_TOKEN is missing"): | |
| upload_surface( | |
| repo_id="noqt/eggcracker", | |
| repo_type="space", | |
| folder=staging, | |
| revision="main", | |
| source_revision=revision, | |
| output=Path(temp_dir) / "upload-result.json", | |
| ) | |
| with self.assertRaisesRegex(VerificationError, "HF_TOKEN is missing"): | |
| verify_surface( | |
| repo_id="noqt/eggcracker", | |
| repo_type="space", | |
| staging=staging, | |
| upload_record=upload_record, | |
| source_revision=revision, | |
| readback=Path(temp_dir) / "readback", | |
| report=report, | |
| ) | |
| def test_upload_rejects_stale_candidate_after_newer_remote_source(self) -> None: | |
| candidate = subprocess.run( | |
| ["git", "rev-parse", "HEAD^"], | |
| check=True, | |
| capture_output=True, | |
| text=True, | |
| ).stdout.strip() | |
| remote_source = subprocess.run( | |
| ["git", "rev-parse", "HEAD"], | |
| check=True, | |
| capture_output=True, | |
| text=True, | |
| ).stdout.strip() | |
| with tempfile.TemporaryDirectory() as temp_dir: | |
| staging, _, _ = self._minimal_surface(Path(temp_dir), candidate) | |
| remote_marker = Path(temp_dir) / "remote-marker.json" | |
| remote_marker.write_text( | |
| json.dumps({"schema": "noqt.huggingface_sync.v1", "source_revision": remote_source}), | |
| encoding="utf-8", | |
| ) | |
| fake_api = SimpleNamespace( | |
| repo_info=lambda **_: SimpleNamespace(sha="2" * 40), | |
| hf_hub_download=lambda **_: str(remote_marker), | |
| upload_folder=Mock(), | |
| ) | |
| fake_hub = SimpleNamespace(HfApi=lambda **_: fake_api) | |
| with ( | |
| patch.dict(os.environ, {"HF_TOKEN": "test-token"}), | |
| patch.dict(sys.modules, {"huggingface_hub": fake_hub}), | |
| self.assertRaisesRegex(UploadError, "not an ancestor"), | |
| ): | |
| upload_surface( | |
| repo_id="noqt/eggcracker", | |
| repo_type="space", | |
| folder=staging, | |
| revision="main", | |
| source_revision=candidate, | |
| output=Path(temp_dir) / "upload-result.json", | |
| source_root=ROOT, | |
| ) | |
| fake_api.upload_folder.assert_not_called() | |
| def test_upload_allows_only_the_pinned_first_migration_parent_without_marker(self) -> None: | |
| candidate = subprocess.run( | |
| ["git", "rev-parse", "HEAD"], | |
| check=True, | |
| capture_output=True, | |
| text=True, | |
| ).stdout.strip() | |
| self.assertEqual( | |
| subprocess.run( | |
| ["git", "-C", str(ROOT), "merge-base", "--is-ancestor", LEGACY_BOOTSTRAP_SOURCE, candidate], | |
| check=False, | |
| ).returncode, | |
| 0, | |
| ) | |
| class EntryNotFoundError(Exception): | |
| pass | |
| with tempfile.TemporaryDirectory() as temp_dir: | |
| staging, _, _ = self._minimal_surface(Path(temp_dir), candidate) | |
| fake_commit = SimpleNamespace( | |
| oid="3" * 40, | |
| commit_url="https://huggingface.co/spaces/noqt/eggcracker/commit/" + "3" * 40, | |
| ) | |
| fake_api = SimpleNamespace( | |
| repo_info=lambda **_: SimpleNamespace(sha=LEGACY_BOOTSTRAP_PARENT), | |
| hf_hub_download=Mock(side_effect=EntryNotFoundError("marker is absent")), | |
| upload_folder=Mock(return_value=fake_commit), | |
| ) | |
| fake_hub = SimpleNamespace( | |
| HfApi=lambda **_: fake_api, | |
| errors=SimpleNamespace(EntryNotFoundError=EntryNotFoundError), | |
| ) | |
| output = Path(temp_dir) / "upload-result.json" | |
| with ( | |
| patch.dict(os.environ, {"HF_TOKEN": "test-token"}), | |
| patch.dict(sys.modules, {"huggingface_hub": fake_hub}), | |
| ): | |
| result = upload_surface( | |
| repo_id="noqt/eggcracker", | |
| repo_type="space", | |
| folder=staging, | |
| revision="main", | |
| source_revision=candidate, | |
| output=output, | |
| source_root=ROOT, | |
| ) | |
| self.assertEqual(candidate, result["source_revision"]) | |
| self.assertEqual(LEGACY_BOOTSTRAP_PARENT, result["parent_commit"]) | |
| self.assertEqual(LEGACY_BOOTSTRAP_PARENT, fake_api.hf_hub_download.call_args.kwargs["revision"]) | |
| fake_api.upload_folder.assert_called_once() | |
| self.assertEqual(LEGACY_BOOTSTRAP_PARENT, fake_api.upload_folder.call_args.kwargs["parent_commit"]) | |
| self.assertEqual(json.loads(output.read_text(encoding="utf-8"))["upload_commit"], "3" * 40) | |
| def test_upload_rejects_missing_marker_for_any_other_parent(self) -> None: | |
| candidate = subprocess.run( | |
| ["git", "rev-parse", "HEAD"], | |
| check=True, | |
| capture_output=True, | |
| text=True, | |
| ).stdout.strip() | |
| class EntryNotFoundError(Exception): | |
| pass | |
| with tempfile.TemporaryDirectory() as temp_dir: | |
| staging, _, _ = self._minimal_surface(Path(temp_dir), candidate) | |
| fake_api = SimpleNamespace( | |
| repo_info=lambda **_: SimpleNamespace(sha="d" * 40), | |
| hf_hub_download=Mock(side_effect=EntryNotFoundError("marker is absent")), | |
| upload_folder=Mock(), | |
| ) | |
| fake_hub = SimpleNamespace( | |
| HfApi=lambda **_: fake_api, | |
| errors=SimpleNamespace(EntryNotFoundError=EntryNotFoundError), | |
| ) | |
| with ( | |
| patch.dict(os.environ, {"HF_TOKEN": "test-token"}), | |
| patch.dict(sys.modules, {"huggingface_hub": fake_hub}), | |
| self.assertRaisesRegex(UploadError, "remote source marker is unreadable"), | |
| ): | |
| upload_surface( | |
| repo_id="noqt/eggcracker", | |
| repo_type="space", | |
| folder=staging, | |
| revision="main", | |
| source_revision=candidate, | |
| output=Path(temp_dir) / "upload-result.json", | |
| source_root=ROOT, | |
| ) | |
| fake_api.upload_folder.assert_not_called() | |
| def test_builds_exact_tracked_surface_with_reviewed_overlays(self) -> None: | |
| revision = subprocess.run( | |
| ["git", "-C", str(ROOT), "rev-parse", "HEAD"], | |
| check=True, | |
| capture_output=True, | |
| text=True, | |
| ).stdout.strip() | |
| with tempfile.TemporaryDirectory() as temp_dir: | |
| output = Path(temp_dir) / "surface" | |
| summary = build_surface(ROOT, output, revision) | |
| marker = json.loads((output / "HUGGINGFACE_SYNC.json").read_text(encoding="utf-8")) | |
| manifest_path = output / "HUGGINGFACE_MANIFEST.json" | |
| manifest_bytes = manifest_path.read_bytes() | |
| manifest = json.loads(manifest_bytes) | |
| readme = (output / "README.md").read_text(encoding="utf-8") | |
| index = (output / "index.html").read_text(encoding="utf-8") | |
| self.assertEqual(revision, marker["source_revision"]) | |
| self.assertEqual(hashlib.sha256(manifest_bytes).hexdigest(), marker["manifest_sha256"]) | |
| self.assertEqual(revision, manifest["source_revision"]) | |
| self.assertEqual(revision, marker["source_url"].rsplit("/", 1)[-1]) | |
| self.assertEqual(policy_release := marker["release_reference"], manifest["release_reference"]) | |
| self.assertEqual(BOUND_RELEASE["source_commit"], policy_release["source_commit"]) | |
| self.assertEqual(len(manifest["files"]), marker["mirrored_files"]) | |
| self.assertEqual(len(manifest["files"]) + 2, summary["mirrored_files"]) | |
| self.assertTrue(readme.startswith("---\ntitle: Lumi Eggcracker\n")) | |
| self.assertIn(revision, readme) | |
| self.assertIn("git clone https://huggingface.co/spaces/noqt/eggcracker", readme) | |
| self.assertNotIn("git clone https://github.com/noqt/Lumi-Eggcracker.git", readme) | |
| retired_reference = "scadastrangelove/" + "awesome-ai-security-tools" | |
| self.assertNotIn(retired_reference, readme) | |
| self.assertIn(revision, index) | |
| self.assertIn(f"SOURCE COMMIT · {revision[:12]}", index) | |
| stale_revision = "bf6910dbd83d30a50a486f84" + "ac0fa96a0244e23e" | |
| self.assertNotIn(stale_revision, index) | |
| self.assertNotIn("Official NOQT", readme) | |
| self.assertIn("Official noqt", readme) | |
| self.assertNotIn("NOQT / EGGCRACKER", index) | |
| self.assertIn("noqt / Lumi Eggcracker", index) | |
| current_vllm = ROOT / "scripts" / "check_vllm_compatibility.py" | |
| mirrored_vllm = output / "scripts" / "check_vllm_compatibility.py" | |
| self.assertEqual(current_vllm.read_bytes(), mirrored_vllm.read_bytes()) | |
| self.assertFalse((output / ".github").exists()) | |
| self.assertFalse((output / "docs" / "huggingface").exists()) | |
| for path in output.rglob("*"): | |
| if path.is_file(): | |
| self.assertNotIn(retired_reference.encode(), path.read_bytes()) | |
| workflow = (ROOT / ".github" / "workflows" / "sync-huggingface.yml").read_text(encoding="utf-8") | |
| self.assertIn("github.event_name == 'release' || github.ref == 'refs/heads/main'", workflow) | |
| self.assertIn("persist-credentials: false", workflow) | |
| def test_rejects_non_exact_revision_and_existing_output(self) -> None: | |
| with tempfile.TemporaryDirectory() as temp_dir: | |
| root = Path(temp_dir) | |
| with self.assertRaisesRegex(SurfaceBuildError, "40-character Git SHA"): | |
| build_surface(ROOT, root / "bad-revision", "main") | |
| with self.assertRaisesRegex(SurfaceBuildError, "requested Git source commit"): | |
| build_surface(ROOT, root / "missing-commit", "a" * 40) | |
| existing = root / "existing" | |
| existing.mkdir() | |
| with self.assertRaisesRegex(SurfaceBuildError, "already exists"): | |
| build_surface(ROOT, existing, "b" * 40) | |
| if __name__ == "__main__": | |
| unittest.main() | |