test_t3_data_integrity.py
file-level
1
files
1
commits
0
hotspots
0
🧊 dead
0
💥 blast risk
| 1 | """Data-integrity tier tests — T3 provenance/tombstone invariants.""" |
| 2 | |
| 3 | from __future__ import annotations |
| 4 | |
| 5 | import json |
| 6 | import unittest |
| 7 | from datetime import UTC, datetime, timedelta |
| 8 | |
| 9 | from scooling_lab_helpers import valid_payload |
| 10 | |
| 11 | from scooling_lab.dataset_review import ( |
| 12 | DatasetStore, |
| 13 | RejectionReasonCode, |
| 14 | default_dataset_shape, |
| 15 | ) |
| 16 | from scooling_lab.errors import ApiError |
| 17 | from scooling_lab.provenance import validate_provenance_record |
| 18 | from scooling_lab.service import TrainingApiService |
| 19 | from scooling_lab.store import TrainingJobStore |
| 20 | |
| 21 | |
| 22 | class T3DataIntegrityTests(unittest.TestCase): |
| 23 | """Data-integrity tests for T3 provenance retention and tombstone content.""" |
| 24 | |
| 25 | def test_data_integrity_t3_expired_tombstone_provenance_matches_original( |
| 26 | self, |
| 27 | ) -> None: |
| 28 | """Provenance retained after expiry is byte-for-byte the original record.""" |
| 29 | |
| 30 | service = TrainingApiService(TrainingJobStore()) |
| 31 | policy = {"policyClass": "ephemeral", "ttlSeconds": 60} |
| 32 | created = service.create_training_job(valid_payload("di-expiry-match", policy)) |
| 33 | job_id = str(created["id"]) |
| 34 | prov_before = service.get_provenance(job_id) |
| 35 | |
| 36 | service.sweep_expired_artifacts(datetime.now(UTC) + timedelta(seconds=120)) |
| 37 | |
| 38 | prov_after = service.get_provenance(job_id) |
| 39 | self.assertEqual( |
| 40 | json.dumps(prov_before, sort_keys=True), |
| 41 | json.dumps(prov_after, sort_keys=True), |
| 42 | ) |
| 43 | validate_provenance_record(prov_after) |
| 44 | |
| 45 | def test_data_integrity_t3_expired_tombstone_carries_no_content_bytes( |
| 46 | self, |
| 47 | ) -> None: |
| 48 | """Expired tombstone does not include artifact content, request, or model fields.""" |
| 49 | |
| 50 | service = TrainingApiService(TrainingJobStore()) |
| 51 | policy = {"policyClass": "ephemeral", "ttlSeconds": 60} |
| 52 | created = service.create_training_job(valid_payload("di-expiry-clean", policy)) |
| 53 | job_id = str(created["id"]) |
| 54 | |
| 55 | service.sweep_expired_artifacts(datetime.now(UTC) + timedelta(seconds=120)) |
| 56 | tombstone_json = json.dumps(service.get_training_job(job_id), sort_keys=True) |
| 57 | |
| 58 | forbidden = ( |
| 59 | "fixture-tiny-llm", |
| 60 | "fixture:synthetic-tiny-v1", |
| 61 | "trainingParameters", |
| 62 | "request", |
| 63 | ) |
| 64 | for term in forbidden: |
| 65 | with self.subTest(term=term): |
| 66 | self.assertNotIn(term, tombstone_json) |
| 67 | |
| 68 | def test_data_integrity_t3_expired_tombstone_has_no_artifact_list(self) -> None: |
| 69 | """Artifact list is empty after TTL expiry.""" |
| 70 | |
| 71 | service = TrainingApiService(TrainingJobStore()) |
| 72 | policy = {"policyClass": "ephemeral", "ttlSeconds": 60} |
| 73 | created = service.create_training_job(valid_payload("di-expiry-arts", policy)) |
| 74 | job_id = str(created["id"]) |
| 75 | |
| 76 | service.sweep_expired_artifacts(datetime.now(UTC) + timedelta(seconds=120)) |
| 77 | |
| 78 | arts = service.list_artifacts(job_id) |
| 79 | self.assertEqual(arts["artifacts"], []) |
| 80 | |
| 81 | def test_data_integrity_t3_explicit_delete_hash_absence_verified(self) -> None: |
| 82 | """Explicit delete still removes hashes from every store serialization.""" |
| 83 | |
| 84 | service = TrainingApiService(TrainingJobStore()) |
| 85 | created = service.create_training_job(valid_payload("di-explicit-delete")) |
| 86 | job_id = str(created["id"]) |
| 87 | artifact = service.list_artifacts(job_id)["artifacts"][0] |
| 88 | hashes = ( |
| 89 | str(artifact["datasetHash"]), |
| 90 | str(artifact["artifactHash"]), |
| 91 | str(service.get_provenance(job_id)["trainingConfigHash"]), |
| 92 | ) |
| 93 | service.delete_artifact(job_id, str(artifact["id"])) |
| 94 | self.assertTrue(service.verify_deleted_artifact_absence(hashes)) |
| 95 | |
| 96 | def test_data_integrity_t3_rejection_reason_is_enum_not_free_text(self) -> None: |
| 97 | """Rejection records expose only enum codes, never caller-supplied text.""" |
| 98 | |
| 99 | ds_store = DatasetStore() |
| 100 | ds_store.register_shape( |
| 101 | "di-rejection-check", |
| 102 | default_dataset_shape(RejectionReasonCode.SYNTHETIC_LIMIT), |
| 103 | ) |
| 104 | ds_store.submit_for_review("di-rejection-check") |
| 105 | public = ds_store.get("di-rejection-check").to_public_dict() |
| 106 | reason_value = str(public.get("rejectionReasonCode", "")) |
| 107 | self.assertIn(reason_value, {c.value for c in RejectionReasonCode}) |
| 108 | self.assertNotIn("caller supplied reason text", str(public)) |
| 109 | |
| 110 | def test_data_integrity_t3_provenance_schema_version_preserved_through_expiry( |
| 111 | self, |
| 112 | ) -> None: |
| 113 | """schemaVersion on retained provenance equals the canonical value.""" |
| 114 | |
| 115 | from scooling_lab.provenance import PROVENANCE_SCHEMA_VERSION |
| 116 | |
| 117 | service = TrainingApiService(TrainingJobStore()) |
| 118 | policy = {"policyClass": "ephemeral", "ttlSeconds": 60} |
| 119 | created = service.create_training_job(valid_payload("di-schema-ver", policy)) |
| 120 | job_id = str(created["id"]) |
| 121 | service.sweep_expired_artifacts(datetime.now(UTC) + timedelta(seconds=120)) |
| 122 | |
| 123 | prov = service.get_provenance(job_id) |
| 124 | self.assertEqual(prov["schemaVersion"], PROVENANCE_SCHEMA_VERSION) |
| 125 | |
| 126 | def test_data_integrity_t3_expiry_does_not_affect_other_jobs(self) -> None: |
| 127 | """Sweeping one expired job does not touch a job with a long retention TTL.""" |
| 128 | |
| 129 | service = TrainingApiService(TrainingJobStore()) |
| 130 | short_policy = {"policyClass": "ephemeral", "ttlSeconds": 60} |
| 131 | long_policy = {"policyClass": "extended", "ttlSeconds": 86_400} |
| 132 | |
| 133 | short = service.create_training_job(valid_payload("di-short", short_policy)) |
| 134 | long_ = service.create_training_job(valid_payload("di-long", long_policy)) |
| 135 | |
| 136 | service.sweep_expired_artifacts(datetime.now(UTC) + timedelta(seconds=120)) |
| 137 | |
| 138 | self.assertEqual(service.get_training_job(str(short["id"]))["status"], "deleted") |
| 139 | self.assertEqual(service.get_training_job(str(long_["id"]))["status"], "succeeded") |
| 140 | arts_long = service.list_artifacts(str(long_["id"])) |
| 141 | self.assertEqual(len(arts_long["artifacts"]), 1) |
| 142 | |
| 143 | |
| 144 | if __name__ == "__main__": |
| 145 | unittest.main() |