"""Unit tier tests for Scooling Lab T0/T2 contracts.""" from __future__ import annotations from datetime import UTC, datetime, timedelta import unittest from scooling_lab_helpers import valid_payload from scooling_lab.contracts import TrainingJobRequest, TrainingJobStatus from scooling_lab.dataset_review import ( RejectionReasonCode, dataset_shape_from_registration, validate_synthetic_dataset_shape, ) from scooling_lab.errors import ApiError, ErrorCode, error_payload from scooling_lab.license_policy import BomEntry, LicensePolicyError, validate_entry from scooling_lab.provenance import ProvenanceRecord, validate_provenance_record from scooling_lab.retention import ( RetentionPolicyClass, expires_at, is_expired, retention_policy_from_mapping, ) from scooling_lab.service import TrainingApiService from scooling_lab.state_machine import cancel, transition from scooling_lab.store import TrainingJobStore class ScoolingLabUnitTests(unittest.TestCase): """Unit tests for state transitions, schema, license policy, and errors.""" def test_unit_state_transitions_are_explicit_and_idempotent(self) -> None: """queued -> running -> succeeded works and replay is a no-op.""" self.assertEqual( transition(TrainingJobStatus.QUEUED, TrainingJobStatus.RUNNING), TrainingJobStatus.RUNNING, ) self.assertEqual( transition(TrainingJobStatus.RUNNING, TrainingJobStatus.SUCCEEDED), TrainingJobStatus.SUCCEEDED, ) self.assertEqual( transition(TrainingJobStatus.SUCCEEDED, TrainingJobStatus.SUCCEEDED), TrainingJobStatus.SUCCEEDED, ) def test_unit_invalid_state_transition_raises_safe_error(self) -> None: """queued -> succeeded is not allowed by the transition table.""" with self.assertRaises(ApiError) as raised: transition(TrainingJobStatus.QUEUED, TrainingJobStatus.SUCCEEDED) self.assertEqual(raised.exception.code, ErrorCode.INVALID_TRANSITION) def test_unit_t4_cancel_transitions_and_terminal_noops(self) -> None: """Cancellation changes active states and leaves terminal states unchanged.""" self.assertEqual(cancel(TrainingJobStatus.QUEUED), TrainingJobStatus.CANCELLED) self.assertEqual(cancel(TrainingJobStatus.RUNNING), TrainingJobStatus.CANCELLED) for terminal in ( TrainingJobStatus.SUCCEEDED, TrainingJobStatus.FAILED, TrainingJobStatus.CANCELLED, TrainingJobStatus.DELETED, ): with self.subTest(terminal=terminal): self.assertEqual(cancel(terminal), terminal) def test_unit_t4_retry_state_transitions_refuse_invalid_originals(self) -> None: """Only failed or cancelled jobs can produce a fresh retry job.""" service = TrainingApiService(TrainingJobStore(), auto_run_worker=False) queued = service.create_training_job(valid_payload("retry-invalid-queued")) with self.assertRaises(ApiError) as queued_retry: service.retry_training_job(str(queued["id"])) self.assertEqual(queued_retry.exception.code, ErrorCode.INVALID_TRANSITION) succeeded_service = TrainingApiService(TrainingJobStore()) succeeded = succeeded_service.create_training_job(valid_payload("retry-invalid-done")) with self.assertRaises(ApiError) as succeeded_retry: succeeded_service.retry_training_job(str(succeeded["id"])) self.assertEqual(succeeded_retry.exception.code, ErrorCode.INVALID_TRANSITION) def test_unit_t4_dataset_validation_maps_shapes_to_reason_codes(self) -> None: """Synthetic dataset validation returns enum reasons for invalid shapes.""" _, valid_shape = dataset_shape_from_registration( { "datasetId": "unit-valid-shape", "rowCount": 3, "declaredSchema": { "exampleId": "string", "inputTokenCount": "integer", "outputTokenCount": "integer", "split": "string", }, } ) self.assertIsNone(validate_synthetic_dataset_shape(valid_shape)) _, too_large = dataset_shape_from_registration( { "datasetId": "unit-large-shape", "rowCount": 10_001, "declaredSchema": { "exampleId": "string", "inputTokenCount": "integer", "outputTokenCount": "integer", "split": "string", }, } ) self.assertEqual( validate_synthetic_dataset_shape(too_large), RejectionReasonCode.SYNTHETIC_LIMIT, ) _, schema_mismatch = dataset_shape_from_registration( { "datasetId": "unit-schema-shape", "rowCount": 3, "declaredSchema": { "exampleId": "string", "inputTokenCount": "integer", "split": "string", }, } ) self.assertEqual( validate_synthetic_dataset_shape(schema_mismatch), RejectionReasonCode.SCHEMA_MISMATCH, ) def test_unit_deleted_state_transition_is_terminal(self) -> None: """succeeded -> deleted works and deleted cannot transition elsewhere.""" self.assertEqual( transition(TrainingJobStatus.SUCCEEDED, TrainingJobStatus.DELETED), TrainingJobStatus.DELETED, ) self.assertEqual( transition(TrainingJobStatus.DELETED, TrainingJobStatus.DELETED), TrainingJobStatus.DELETED, ) with self.assertRaises(ApiError): transition(TrainingJobStatus.DELETED, TrainingJobStatus.RUNNING) def test_unit_schema_rejects_unsafe_fields_and_unapproved_models(self) -> None: """Dangerous keys and unknown model ids fail in schema validation.""" payload = valid_payload() payload["workerUrl"] = "https://attacker.invalid/worker" with self.assertRaises(ApiError): TrainingJobRequest.from_mapping(payload) model_payload = valid_payload() model_payload["modelId"] = "unapproved-model" with self.assertRaises(ApiError): TrainingJobRequest.from_mapping(model_payload) def test_unit_provenance_schema_rejects_free_text_paths_and_urls(self) -> None: """Provenance accepts only exact hashes, ids, timestamps, and schema version.""" record = ProvenanceRecord( artifact_hash="b" * 64, base_model_id="fixture-tiny-llm", created_at="2026-06-11T00:00:00Z", dataset_hash="a" * 64, job_id="job_" + "1" * 24, training_config_hash="c" * 64, ) validate_provenance_record(record) unsafe_values = ( "prompt body with words", "../private/path", "https://attacker.invalid/model", ) for unsafe in unsafe_values: payload = record.to_dict() payload["baseModelId"] = unsafe with self.subTest(unsafe=unsafe): with self.assertRaises(ApiError): validate_provenance_record(payload) def test_unit_retention_ttl_math_is_bounded_and_deterministic(self) -> None: """Retention policies validate TTL bounds and evaluate expiry exactly.""" policy = retention_policy_from_mapping( {"policyClass": "ephemeral", "ttlSeconds": 60} ) self.assertEqual(policy.policy_class, RetentionPolicyClass.EPHEMERAL) self.assertEqual(expires_at("2026-06-11T00:00:00Z", policy), "2026-06-11T00:01:00Z") self.assertFalse( is_expired( "2026-06-11T00:00:00Z", policy, datetime(2026, 6, 11, 0, 0, 59, tzinfo=UTC), ) ) self.assertTrue( is_expired( "2026-06-11T00:00:00Z", policy, datetime(2026, 6, 11, 0, 1, 0, tzinfo=UTC), ) ) with self.assertRaises(ApiError): retention_policy_from_mapping({"policyClass": "ephemeral", "ttlSeconds": 59}) def test_unit_deletion_state_transition_removes_content_fields(self) -> None: """deleteArtifact leaves a content-free deleted job tombstone.""" service = TrainingApiService(TrainingJobStore()) created = service.create_training_job(valid_payload("delete-unit")) job_id = str(created["id"]) artifact_id = str(service.list_artifacts(job_id)["artifacts"][0]["id"]) first_delete = service.delete_artifact(job_id, artifact_id) second_delete = service.delete_artifact(job_id, artifact_id) tombstone = service.get_training_job(job_id) self.assertTrue(first_delete["deleted"]) self.assertTrue(first_delete["verified"]) self.assertTrue(second_delete["alreadyDeleted"]) self.assertEqual(tombstone["status"], "deleted") self.assertNotIn("request", tombstone) self.assertNotIn("artifactHash", str(tombstone)) def test_unit_license_policy_blocks_agpl_and_blocked_paths(self) -> None: """The BOM policy rejects AGPL licenses and Unsloth Studio/CLI paths.""" validate_entry( BomEntry( name="safe", version="1.0.0", license="MIT", source_path="third_party/safe", evidence="fixture", ) ) with self.assertRaises(LicensePolicyError): validate_entry( BomEntry( name="blocked", version="1.0.0", license="AGPL-3.0", source_path="third_party/blocked", evidence="fixture", ) ) with self.assertRaises(LicensePolicyError): validate_entry( BomEntry( name="blocked-path", version="1.0.0", license="Apache-2.0", source_path="vendor/unsloth_cli/main.py", evidence="fixture", ) ) def test_unit_error_model_does_not_echo_payload_or_paths(self) -> None: """Safe errors expose only code and stable public message.""" payload = error_payload(ApiError(ErrorCode.VALIDATION_ERROR, 400)) text = str(payload) self.assertIn("VALIDATION_ERROR", text) self.assertNotIn("/Users/", text) self.assertNotIn("attacker.invalid", text) self.assertNotIn("payload", text.lower()) if __name__ == "__main__": unittest.main()