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