Buckets:
| import json | |
| import tempfile | |
| import unittest | |
| from pathlib import Path | |
| from unittest.mock import patch | |
| from shared.artifacts import ( | |
| claim_run_directory, | |
| file_sha256, | |
| save_adapter_atomic, | |
| source_fingerprint, | |
| timestamped_run_name, | |
| ) | |
| class _Saver: | |
| def save_pretrained(self, path, **_kwargs): | |
| (Path(path) / "weights").write_text("ok\n", encoding="utf-8") | |
| class SharedArtifactTest(unittest.TestCase): | |
| def test_source_fingerprint_requires_an_explicit_package_closure(self): | |
| with self.assertRaisesRegex(ValueError, "explicit package"): | |
| source_fingerprint() | |
| def test_submission_claim_requires_matching_runtime_attestation(self): | |
| with tempfile.TemporaryDirectory() as directory: | |
| run = Path(directory) / "run" | |
| submission = run / "_submission" | |
| submission.mkdir(parents=True) | |
| ready = submission / "READY" | |
| ready.write_text( | |
| json.dumps({"schema": 1, "metadata_sha256": {}}) + "\n", | |
| encoding="utf-8", | |
| ) | |
| (submission / "manifest.json").write_text( | |
| json.dumps({"copied_hash": "source"}) + "\n", | |
| encoding="utf-8", | |
| ) | |
| (submission / "runtime-verification-17.json").write_text( | |
| json.dumps( | |
| { | |
| "schema": "spatial_mmcot_runtime_verification_v1", | |
| "slurm_job_id": "17", | |
| "source_hash": "source", | |
| "ready_sha256": file_sha256(ready), | |
| } | |
| ) | |
| + "\n", | |
| encoding="utf-8", | |
| ) | |
| with patch.dict("os.environ", {"SLURM_JOB_ID": "17"}, clear=False): | |
| self.assertEqual(claim_run_directory(run), run) | |
| with self.assertRaisesRegex(RuntimeError, "already claimed"): | |
| claim_run_directory(run) | |
| def test_existing_unverified_directory_is_never_claimed(self): | |
| with tempfile.TemporaryDirectory() as directory: | |
| run = Path(directory) / "run" | |
| run.mkdir() | |
| with patch.dict("os.environ", {}, clear=True): | |
| with self.assertRaisesRegex(FileExistsError, "verified Slurm"): | |
| claim_run_directory(run) | |
| def test_slurm_cannot_create_an_unprepared_run(self): | |
| with tempfile.TemporaryDirectory() as directory: | |
| run = Path(directory) / "missing" | |
| with patch.dict("os.environ", {"SLURM_JOB_ID": "19"}, clear=False): | |
| with self.assertRaisesRegex(FileNotFoundError, "submission-prepared"): | |
| claim_run_directory(run) | |
| self.assertFalse(run.exists()) | |
| def test_final_adapter_publish_is_atomic_and_idempotent(self): | |
| with tempfile.TemporaryDirectory() as directory: | |
| target = Path(directory) / "adapter" | |
| save_adapter_atomic(_Saver(), _Saver(), target) | |
| self.assertTrue((target / "COMPLETE").is_file()) | |
| self.assertEqual(save_adapter_atomic(_Saver(), _Saver(), target), target) | |
| def test_timestamp_name_honors_the_canonical_timezone_override(self): | |
| with patch.dict( | |
| "os.environ", {"SPATIAL_MMCOT_TZ_OFFSET": "+8"}, clear=False | |
| ): | |
| name = timestamped_run_name("fixture") | |
| self.assertRegex(name, r"^\d{8}-\d{6}_fixture$") | |
| if __name__ == "__main__": | |
| unittest.main() | |
Xet Storage Details
- Size:
- 3.48 kB
- Xet hash:
- da4f0663566fbd44d381fb18f680bccd7a2cef8fdbd4687c5031ef7d8f4353ce
·
Xet efficiently stores files, intelligently splitting them into unique chunks and accelerating uploads and downloads. More info.