test_data_integrity.py
file-level
1
files
1
commits
0
hotspots
0
🧊 dead
0
💥 blast risk
| 1 | """Data-integrity tier tests for provenance and deletion invariants.""" |
| 2 | |
| 3 | from __future__ import annotations |
| 4 | |
| 5 | import json |
| 6 | import tempfile |
| 7 | import unittest |
| 8 | from pathlib import Path |
| 9 | |
| 10 | from scooling_lab_helpers import PROJECT_ROOT, valid_payload |
| 11 | |
| 12 | from scooling_lab.contracts import TrainingJobStatus |
| 13 | from scooling_lab.fake_worker import fixture_dataset_hash |
| 14 | from scooling_lab.provenance import validate_provenance_record |
| 15 | from scooling_lab.service import TrainingApiService |
| 16 | from scooling_lab.store import TrainingJobStore |
| 17 | |
| 18 | |
| 19 | EXPECTED_UNSLOTH_NOTICE = ( |
| 20 | "Files under unsloth/*, tests/*, scripts/* are Apache 2.0 licensed.\n" |
| 21 | "Files under studio/*, unsloth_cli/* which is optional to install are AGPLv3 licensed." |
| 22 | ) |
| 23 | |
| 24 | |
| 25 | class ScoolingLabDataIntegrityTests(unittest.TestCase): |
| 26 | """Data-integrity tests for stable hashes, reloads, and tombstones.""" |
| 27 | |
| 28 | def test_data_integrity_provenance_hashes_stable_across_restarts(self) -> None: |
| 29 | """Persisted jobs reload with the same provenance and artifact hashes.""" |
| 30 | |
| 31 | with tempfile.TemporaryDirectory() as directory: |
| 32 | store_path = Path(directory) / "store.json" |
| 33 | service = TrainingApiService(TrainingJobStore(persistence_path=store_path)) |
| 34 | created = service.create_training_job(valid_payload("restart-stability")) |
| 35 | job_id = str(created["id"]) |
| 36 | first_artifact = service.list_artifacts(job_id)["artifacts"][0] |
| 37 | first_provenance = service.get_provenance(job_id) |
| 38 | |
| 39 | reloaded = TrainingApiService(TrainingJobStore(persistence_path=store_path)) |
| 40 | replayed = reloaded.create_training_job(valid_payload("restart-stability")) |
| 41 | second_artifact = reloaded.list_artifacts(job_id)["artifacts"][0] |
| 42 | second_provenance = reloaded.get_provenance(job_id) |
| 43 | |
| 44 | self.assertEqual(replayed["id"], job_id) |
| 45 | self.assertEqual(first_artifact["artifactHash"], second_artifact["artifactHash"]) |
| 46 | self.assertEqual(first_artifact["datasetHash"], second_artifact["datasetHash"]) |
| 47 | self.assertEqual(first_provenance, second_provenance) |
| 48 | |
| 49 | def test_data_integrity_deletion_verification_finds_zero_residue(self) -> None: |
| 50 | """Deletion verification proves deleted hashes are absent from store output.""" |
| 51 | |
| 52 | service = TrainingApiService(TrainingJobStore()) |
| 53 | created = service.create_training_job(valid_payload("zero-residue")) |
| 54 | job_id = str(created["id"]) |
| 55 | artifact = service.list_artifacts(job_id)["artifacts"][0] |
| 56 | provenance = service.get_provenance(job_id) |
| 57 | deleted_hashes = ( |
| 58 | str(artifact["datasetHash"]), |
| 59 | str(artifact["artifactHash"]), |
| 60 | str(provenance["trainingConfigHash"]), |
| 61 | ) |
| 62 | |
| 63 | service.delete_artifact(job_id, str(artifact["id"])) |
| 64 | |
| 65 | self.assertTrue(service.verify_deleted_artifact_absence(deleted_hashes)) |
| 66 | |
| 67 | def test_data_integrity_tombstone_carries_no_content_fields(self) -> None: |
| 68 | """Deleted job tombstones do not carry request, model, dataset, or hash fields.""" |
| 69 | |
| 70 | service = TrainingApiService(TrainingJobStore()) |
| 71 | created = service.create_training_job(valid_payload("tombstone")) |
| 72 | job_id = str(created["id"]) |
| 73 | artifact_id = str(service.list_artifacts(job_id)["artifacts"][0]["id"]) |
| 74 | |
| 75 | service.delete_artifact(job_id, artifact_id) |
| 76 | tombstone_json = json.dumps(service.get_training_job(job_id), sort_keys=True) |
| 77 | |
| 78 | forbidden_terms = ( |
| 79 | "artifactHash", |
| 80 | "datasetHash", |
| 81 | "fixture-tiny-llm", |
| 82 | "fixture:synthetic-tiny-v1", |
| 83 | "request", |
| 84 | "trainingParameters", |
| 85 | ) |
| 86 | for forbidden in forbidden_terms: |
| 87 | with self.subTest(forbidden=forbidden): |
| 88 | self.assertNotIn(forbidden, tombstone_json) |
| 89 | |
| 90 | def test_data_integrity_t4_retry_lineage_and_original_immutability(self) -> None: |
| 91 | """Retry provenance is valid while the cancelled original remains unchanged.""" |
| 92 | |
| 93 | store = TrainingJobStore() |
| 94 | service = TrainingApiService(store, auto_run_worker=False) |
| 95 | original = service.create_training_job(valid_payload("di-t4-original")) |
| 96 | original_id = str(original["id"]) |
| 97 | cancelled = service.cancel_training_job(original_id) |
| 98 | original_snapshot = json.dumps( |
| 99 | service.get_training_job(original_id), |
| 100 | sort_keys=True, |
| 101 | ) |
| 102 | retry_service = TrainingApiService(store) |
| 103 | |
| 104 | retried = retry_service.retry_training_job(original_id) |
| 105 | provenance = retry_service.get_provenance(str(retried["id"])) |
| 106 | original_after_retry = json.dumps( |
| 107 | retry_service.get_training_job(original_id), |
| 108 | sort_keys=True, |
| 109 | ) |
| 110 | |
| 111 | self.assertEqual(cancelled["status"], "cancelled") |
| 112 | self.assertEqual(original_snapshot, original_after_retry) |
| 113 | self.assertEqual(retried["retryOfJobId"], original_id) |
| 114 | self.assertEqual(retried["status"], "succeeded") |
| 115 | validate_provenance_record(provenance) |
| 116 | self.assertEqual(provenance["jobId"], retried["id"]) |
| 117 | self.assertNotEqual(provenance["jobId"], original_id) |
| 118 | |
| 119 | def test_data_integrity_t4_cancelled_terminal_record_has_no_artifacts(self) -> None: |
| 120 | """Cancelled jobs are terminal active-records without artifact metadata.""" |
| 121 | |
| 122 | service = TrainingApiService(TrainingJobStore(), auto_run_worker=False) |
| 123 | created = service.create_training_job(valid_payload("di-t4-cancelled")) |
| 124 | job_id = str(created["id"]) |
| 125 | |
| 126 | cancelled = service.cancel_training_job(job_id) |
| 127 | |
| 128 | self.assertEqual(cancelled["status"], TrainingJobStatus.CANCELLED.value) |
| 129 | self.assertEqual(service.list_artifacts(job_id)["artifacts"], []) |
| 130 | |
| 131 | def test_data_integrity_job_and_artifact_hashes_survive_restart(self) -> None: |
| 132 | """Job id, dataset hash, and artifact hash are stable after reload.""" |
| 133 | |
| 134 | with tempfile.TemporaryDirectory() as directory: |
| 135 | persistence_path = Path(directory) / "jobs.json" |
| 136 | service = TrainingApiService(TrainingJobStore(persistence_path)) |
| 137 | created = service.create_training_job(valid_payload("restart")) |
| 138 | job_id = str(created["id"]) |
| 139 | artifact = service.list_artifacts(job_id)["artifacts"][0] |
| 140 | artifact_hash = str(artifact["artifactHash"]) |
| 141 | dataset_hash = str(artifact["datasetHash"]) |
| 142 | |
| 143 | reloaded = TrainingApiService(TrainingJobStore(persistence_path)) |
| 144 | recreated = reloaded.create_training_job(valid_payload("restart")) |
| 145 | reloaded_artifact = reloaded.list_artifacts(job_id)["artifacts"][0] |
| 146 | |
| 147 | self.assertEqual(recreated["id"], job_id) |
| 148 | self.assertEqual(reloaded_artifact["datasetHash"], dataset_hash) |
| 149 | self.assertEqual(reloaded_artifact["artifactHash"], artifact_hash) |
| 150 | self.assertEqual(dataset_hash, fixture_dataset_hash()) |
| 151 | |
| 152 | def test_data_integrity_unsloth_evidence_preserved_byte_for_byte(self) -> None: |
| 153 | """CI-visible license evidence preserves the pinned candidate and notice.""" |
| 154 | |
| 155 | evidence = (PROJECT_ROOT / "docs/UNSLOTH-LICENSE-EVIDENCE.md").read_text( |
| 156 | encoding="utf-8" |
| 157 | ) |
| 158 | self.assertIn("- Candidate version: `2026.6.1`.", evidence) |
| 159 | self.assertIn(EXPECTED_UNSLOTH_NOTICE, evidence) |
| 160 | |
| 161 | |
| 162 | if __name__ == "__main__": |
| 163 | unittest.main() |