aaronrene / scooling-lab public
test_t3_integration.py python
216 lines 9.1 KB
Raw
sha256:ca1d0e687bff8686b37126eb8e4fb6d38b40e189352a4920d66ae89c7274e340 Add T4 job cancellation retry and validation Human minor ⚠ breaking 41 days ago
1 """Integration tier tests — T3 dataset review + job lifecycle API contract."""
2
3 from __future__ import annotations
4
5 import unittest
6 from datetime import UTC, datetime, timedelta
7
8 from scooling_lab_helpers import valid_payload
9
10 from scooling_lab.dataset_review import (
11 DatasetStatus,
12 DatasetStore,
13 RejectionReasonCode,
14 default_dataset_shape,
15 )
16 from scooling_lab.errors import ApiError, ErrorCode
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 def _make_service(
23 *,
24 queue_limit: int = 10,
25 max_concurrent_running: int = 1,
26 dataset_store: DatasetStore | None = None,
27 auto_run: bool = True,
28 ) -> TrainingApiService:
29 """Convenience factory for integration tests."""
30
31 store = TrainingJobStore(
32 queue_limit=queue_limit, max_concurrent_running=max_concurrent_running
33 )
34 return TrainingApiService(
35 store, auto_run_worker=auto_run, dataset_store=dataset_store
36 )
37
38
39 class T3IntegrationDatasetReviewLifecycleTests(unittest.TestCase):
40 """Integration tests for the full dataset registration → review flow."""
41
42 def test_integration_t3_register_review_approve_then_submit_job(self) -> None:
43 """A dataset that goes through the full approval flow allows job creation."""
44
45 ds_store = DatasetStore()
46 ds_store.register("custom-ds-v1")
47 ds_store.submit_for_review("custom-ds-v1")
48 ds_store.approve("custom-ds-v1")
49 # Use a service that also approves the fixture dataset (default).
50 service = TrainingApiService(TrainingJobStore(), dataset_store=ds_store)
51
52 # Standard fixture job still works because fixture is pre-approved.
53 created = service.create_training_job(valid_payload("integration-approved"))
54 self.assertEqual(created["status"], "succeeded")
55
56 def test_integration_t3_unapproved_dataset_blocks_job_at_api_boundary(self) -> None:
57 """A dataset that is only registered (not approved) blocks job submission."""
58
59 ds_store = DatasetStore()
60 ds_store.register("custom-ds-v2")
61 service = TrainingApiService(TrainingJobStore(), dataset_store=ds_store)
62 with self.assertRaises(ApiError) as raised:
63 service.create_training_job(
64 {
65 "idempotencyKey": "integration-blocked",
66 "datasetId": "custom-ds-v2",
67 "modelId": "fixture-tiny-llm",
68 "requestedBy": "integration-test",
69 }
70 )
71 self.assertEqual(raised.exception.code, ErrorCode.DATASET_NOT_APPROVED)
72
73 def test_integration_t3_rejected_dataset_blocks_job(self) -> None:
74 """A rejected dataset is refused at job creation with DATASET_NOT_APPROVED."""
75
76 ds_store = DatasetStore()
77 ds_store.register_shape(
78 "custom-ds-v3",
79 default_dataset_shape(RejectionReasonCode.POLICY_VIOLATION),
80 )
81 ds_store.submit_for_review("custom-ds-v3")
82 ds_store.reject("custom-ds-v3", RejectionReasonCode.POLICY_VIOLATION)
83 service = TrainingApiService(TrainingJobStore(), dataset_store=ds_store)
84 with self.assertRaises(ApiError) as raised:
85 service.create_training_job(
86 {
87 "idempotencyKey": "integration-rejected",
88 "datasetId": "custom-ds-v3",
89 "modelId": "fixture-tiny-llm",
90 "requestedBy": "integration-test",
91 }
92 )
93 self.assertEqual(raised.exception.code, ErrorCode.DATASET_NOT_APPROVED)
94
95 def test_integration_t3_rejection_reason_in_dataset_record(self) -> None:
96 """Rejection reason code is visible in the dataset record's public dict."""
97
98 ds_store = DatasetStore()
99 ds_store.register_shape(
100 "reason-ds-v1",
101 default_dataset_shape(RejectionReasonCode.SCHEMA_MISMATCH),
102 )
103 ds_store.submit_for_review("reason-ds-v1")
104 ds_store.reject("reason-ds-v1", RejectionReasonCode.SCHEMA_MISMATCH)
105 record = ds_store.get("reason-ds-v1")
106 public = record.to_public_dict()
107 self.assertEqual(public["rejectionReasonCode"], "SCHEMA_MISMATCH")
108 self.assertEqual(public["status"], DatasetStatus.REJECTED.value)
109
110 def test_integration_t3_dataset_status_does_not_expose_free_text(self) -> None:
111 """Public dataset record contains no free-text fields from request payload."""
112
113 ds_store = DatasetStore()
114 ds_store.register_shape(
115 "clean-ds-v1",
116 default_dataset_shape(RejectionReasonCode.FORMAT_INVALID),
117 )
118 ds_store.submit_for_review("clean-ds-v1")
119 ds_store.reject("clean-ds-v1", RejectionReasonCode.FORMAT_INVALID)
120 public_str = str(ds_store.get("clean-ds-v1").to_public_dict())
121 self.assertNotIn("caller message", public_str)
122 self.assertIn("FORMAT_INVALID", public_str)
123
124
125 class T3IntegrationQueueStateTests(unittest.TestCase):
126 """Integration tests for queue state inspection."""
127
128 def test_integration_t3_queue_state_reflects_active_jobs(self) -> None:
129 """Queue state accurately counts queued/running/active jobs."""
130
131 service = _make_service(auto_run=False, max_concurrent_running=2, queue_limit=20)
132 initial = service.get_queue_state()
133 self.assertEqual(initial["queuedCount"], 0)
134 self.assertEqual(initial["runningCount"], 0)
135 self.assertEqual(initial["activeCount"], 0)
136
137 service.create_training_job(valid_payload("queue-state-a"))
138 service.create_training_job(valid_payload("queue-state-b"))
139 after = service.get_queue_state()
140 self.assertEqual(after["queuedCount"], 2)
141 self.assertEqual(after["maxConcurrentRunning"], 2)
142
143 def test_integration_t3_queue_state_after_completion(self) -> None:
144 """Completed jobs do not count in the active queue."""
145
146 service = _make_service(auto_run=True, max_concurrent_running=5, queue_limit=20)
147 service.create_training_job(valid_payload("complete-queue"))
148 after = service.get_queue_state()
149 self.assertEqual(after["queuedCount"], 0)
150 self.assertEqual(after["runningCount"], 0)
151 self.assertEqual(after["activeCount"], 0)
152
153
154 class T3IntegrationProvenanceOnCompletionTests(unittest.TestCase):
155 """Integration tests confirming provenance is always emitted via the Slice-5 validator."""
156
157 def test_integration_t3_every_succeeded_job_has_valid_provenance(self) -> None:
158 """Succeeded jobs always expose a schema-valid provenance record."""
159
160 service = _make_service()
161 for suffix in ("prov-a", "prov-b", "prov-c"):
162 created = service.create_training_job(valid_payload(suffix))
163 job_id = str(created["id"])
164 self.assertEqual(created["status"], "succeeded")
165 prov = service.get_provenance(job_id)
166 validate_provenance_record(prov)
167 self.assertEqual(prov["jobId"], job_id)
168
169 def test_integration_t3_provenance_links_to_artifact(self) -> None:
170 """Provenance artifact hash matches the artifact in the job's artifact list."""
171
172 service = _make_service()
173 created = service.create_training_job(valid_payload("prov-link"))
174 job_id = str(created["id"])
175 artifact = service.list_artifacts(job_id)["artifacts"][0]
176 prov = service.get_provenance(job_id)
177 self.assertEqual(prov["artifactHash"], artifact["artifactHash"])
178
179
180 class T3IntegrationRetentionIntegrationTests(unittest.TestCase):
181 """Integration tests for retention integration (Slice 5 + T3)."""
182
183 def test_integration_t3_expired_artifact_tombstone_has_provenance(self) -> None:
184 """After TTL expiry the tombstone still exposes its provenance record."""
185
186 service = _make_service()
187 policy = {"policyClass": "ephemeral", "ttlSeconds": 60}
188 created = service.create_training_job(valid_payload("retention-prov", policy))
189 job_id = str(created["id"])
190 prov_before = service.get_provenance(job_id)
191
192 service.sweep_expired_artifacts(datetime.now(UTC) + timedelta(seconds=120))
193
194 tombstone = service.get_training_job(job_id)
195 self.assertEqual(tombstone["status"], "deleted")
196 self.assertEqual(service.list_artifacts(job_id)["artifacts"], [])
197 prov_after = service.get_provenance(job_id)
198 self.assertEqual(prov_before["jobId"], prov_after["jobId"])
199 self.assertEqual(prov_before["artifactHash"], prov_after["artifactHash"])
200 validate_provenance_record(prov_after)
201
202 def test_integration_t3_explicit_delete_wipes_provenance(self) -> None:
203 """Explicit deleteArtifact wipes provenance (Slice-5 contract preserved)."""
204
205 service = _make_service()
206 created = service.create_training_job(valid_payload("explicit-delete"))
207 job_id = str(created["id"])
208 artifact_id = str(service.list_artifacts(job_id)["artifacts"][0]["id"])
209 service.delete_artifact(job_id, artifact_id)
210 with self.assertRaises(ApiError) as raised:
211 service.get_provenance(job_id)
212 self.assertEqual(raised.exception.code, ErrorCode.NOT_FOUND)
213
214
215 if __name__ == "__main__":
216 unittest.main()
File History 1 commit
sha256:ca1d0e687bff8686b37126eb8e4fb6d38b40e189352a4920d66ae89c7274e340 Add T4 job cancellation retry and validation Human minor ⚠ 41 days ago