aaronrene / scooling-lab public
test_unit.py python
192 lines 7.3 KB
Raw
sha256:91e875d4a97bb1e35f37992f803988d5713931f1782d870c371c50054574af22 Add fixture provenance retention deletion Human minor ⚠ breaking 43 days ago
1 """Unit tier tests for Scooling Lab T0/T2 contracts."""
2
3 from __future__ import annotations
4
5 from datetime import UTC, datetime, timedelta
6 import unittest
7
8 from scooling_lab_helpers import valid_payload
9
10 from scooling_lab.contracts import TrainingJobRequest, TrainingJobStatus
11 from scooling_lab.errors import ApiError, ErrorCode, error_payload
12 from scooling_lab.license_policy import BomEntry, LicensePolicyError, validate_entry
13 from scooling_lab.provenance import ProvenanceRecord, validate_provenance_record
14 from scooling_lab.retention import (
15 RetentionPolicyClass,
16 expires_at,
17 is_expired,
18 retention_policy_from_mapping,
19 )
20 from scooling_lab.service import TrainingApiService
21 from scooling_lab.state_machine import transition
22 from scooling_lab.store import TrainingJobStore
23
24
25 class ScoolingLabUnitTests(unittest.TestCase):
26 """Unit tests for state transitions, schema, license policy, and errors."""
27
28 def test_unit_state_transitions_are_explicit_and_idempotent(self) -> None:
29 """queued -> running -> succeeded works and replay is a no-op."""
30
31 self.assertEqual(
32 transition(TrainingJobStatus.QUEUED, TrainingJobStatus.RUNNING),
33 TrainingJobStatus.RUNNING,
34 )
35 self.assertEqual(
36 transition(TrainingJobStatus.RUNNING, TrainingJobStatus.SUCCEEDED),
37 TrainingJobStatus.SUCCEEDED,
38 )
39 self.assertEqual(
40 transition(TrainingJobStatus.SUCCEEDED, TrainingJobStatus.SUCCEEDED),
41 TrainingJobStatus.SUCCEEDED,
42 )
43
44 def test_unit_invalid_state_transition_raises_safe_error(self) -> None:
45 """queued -> succeeded is not allowed by the transition table."""
46
47 with self.assertRaises(ApiError) as raised:
48 transition(TrainingJobStatus.QUEUED, TrainingJobStatus.SUCCEEDED)
49 self.assertEqual(raised.exception.code, ErrorCode.INVALID_TRANSITION)
50
51 def test_unit_deleted_state_transition_is_terminal(self) -> None:
52 """succeeded -> deleted works and deleted cannot transition elsewhere."""
53
54 self.assertEqual(
55 transition(TrainingJobStatus.SUCCEEDED, TrainingJobStatus.DELETED),
56 TrainingJobStatus.DELETED,
57 )
58 self.assertEqual(
59 transition(TrainingJobStatus.DELETED, TrainingJobStatus.DELETED),
60 TrainingJobStatus.DELETED,
61 )
62 with self.assertRaises(ApiError):
63 transition(TrainingJobStatus.DELETED, TrainingJobStatus.RUNNING)
64
65 def test_unit_schema_rejects_unsafe_fields_and_unapproved_models(self) -> None:
66 """Dangerous keys and unknown model ids fail in schema validation."""
67
68 payload = valid_payload()
69 payload["workerUrl"] = "https://attacker.invalid/worker"
70 with self.assertRaises(ApiError):
71 TrainingJobRequest.from_mapping(payload)
72
73 model_payload = valid_payload()
74 model_payload["modelId"] = "unapproved-model"
75 with self.assertRaises(ApiError):
76 TrainingJobRequest.from_mapping(model_payload)
77
78 def test_unit_provenance_schema_rejects_free_text_paths_and_urls(self) -> None:
79 """Provenance accepts only exact hashes, ids, timestamps, and schema version."""
80
81 record = ProvenanceRecord(
82 artifact_hash="b" * 64,
83 base_model_id="fixture-tiny-llm",
84 created_at="2026-06-11T00:00:00Z",
85 dataset_hash="a" * 64,
86 job_id="job_" + "1" * 24,
87 training_config_hash="c" * 64,
88 )
89 validate_provenance_record(record)
90
91 unsafe_values = (
92 "prompt body with words",
93 "../private/path",
94 "https://attacker.invalid/model",
95 )
96 for unsafe in unsafe_values:
97 payload = record.to_dict()
98 payload["baseModelId"] = unsafe
99 with self.subTest(unsafe=unsafe):
100 with self.assertRaises(ApiError):
101 validate_provenance_record(payload)
102
103 def test_unit_retention_ttl_math_is_bounded_and_deterministic(self) -> None:
104 """Retention policies validate TTL bounds and evaluate expiry exactly."""
105
106 policy = retention_policy_from_mapping(
107 {"policyClass": "ephemeral", "ttlSeconds": 60}
108 )
109 self.assertEqual(policy.policy_class, RetentionPolicyClass.EPHEMERAL)
110 self.assertEqual(expires_at("2026-06-11T00:00:00Z", policy), "2026-06-11T00:01:00Z")
111 self.assertFalse(
112 is_expired(
113 "2026-06-11T00:00:00Z",
114 policy,
115 datetime(2026, 6, 11, 0, 0, 59, tzinfo=UTC),
116 )
117 )
118 self.assertTrue(
119 is_expired(
120 "2026-06-11T00:00:00Z",
121 policy,
122 datetime(2026, 6, 11, 0, 1, 0, tzinfo=UTC),
123 )
124 )
125 with self.assertRaises(ApiError):
126 retention_policy_from_mapping({"policyClass": "ephemeral", "ttlSeconds": 59})
127
128 def test_unit_deletion_state_transition_removes_content_fields(self) -> None:
129 """deleteArtifact leaves a content-free deleted job tombstone."""
130
131 service = TrainingApiService(TrainingJobStore())
132 created = service.create_training_job(valid_payload("delete-unit"))
133 job_id = str(created["id"])
134 artifact_id = str(service.list_artifacts(job_id)["artifacts"][0]["id"])
135
136 first_delete = service.delete_artifact(job_id, artifact_id)
137 second_delete = service.delete_artifact(job_id, artifact_id)
138 tombstone = service.get_training_job(job_id)
139
140 self.assertTrue(first_delete["deleted"])
141 self.assertTrue(first_delete["verified"])
142 self.assertTrue(second_delete["alreadyDeleted"])
143 self.assertEqual(tombstone["status"], "deleted")
144 self.assertNotIn("request", tombstone)
145 self.assertNotIn("artifactHash", str(tombstone))
146
147 def test_unit_license_policy_blocks_agpl_and_blocked_paths(self) -> None:
148 """The BOM policy rejects AGPL licenses and Unsloth Studio/CLI paths."""
149
150 validate_entry(
151 BomEntry(
152 name="safe",
153 version="1.0.0",
154 license="MIT",
155 source_path="third_party/safe",
156 evidence="fixture",
157 )
158 )
159 with self.assertRaises(LicensePolicyError):
160 validate_entry(
161 BomEntry(
162 name="blocked",
163 version="1.0.0",
164 license="AGPL-3.0",
165 source_path="third_party/blocked",
166 evidence="fixture",
167 )
168 )
169 with self.assertRaises(LicensePolicyError):
170 validate_entry(
171 BomEntry(
172 name="blocked-path",
173 version="1.0.0",
174 license="Apache-2.0",
175 source_path="vendor/unsloth_cli/main.py",
176 evidence="fixture",
177 )
178 )
179
180 def test_unit_error_model_does_not_echo_payload_or_paths(self) -> None:
181 """Safe errors expose only code and stable public message."""
182
183 payload = error_payload(ApiError(ErrorCode.VALIDATION_ERROR, 400))
184 text = str(payload)
185 self.assertIn("VALIDATION_ERROR", text)
186 self.assertNotIn("/Users/", text)
187 self.assertNotIn("attacker.invalid", text)
188 self.assertNotIn("payload", text.lower())
189
190
191 if __name__ == "__main__":
192 unittest.main()
File History 1 commit
sha256:91e875d4a97bb1e35f37992f803988d5713931f1782d870c371c50054574af22 Add fixture provenance retention deletion Human minor ⚠ 43 days ago