aaronrene / scooling-lab public
test_t3_data_integrity.py python
145 lines 5.9 KB
Raw
sha256:ca1d0e687bff8686b37126eb8e4fb6d38b40e189352a4920d66ae89c7274e340 Add T4 job cancellation retry and validation Human minor ⚠ breaking 41 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 (
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()
File History 1 commit
sha256:ca1d0e687bff8686b37126eb8e4fb6d38b40e189352a4920d66ae89c7274e340 Add T4 job cancellation retry and validation Human minor ⚠ 41 days ago