test_performance.py
file-level
1
files
1
commits
0
hotspots
0
🧊 dead
0
💥 blast risk
| 1 | """Performance tier tests for Scooling Lab bounded local operations.""" |
| 2 | |
| 3 | from __future__ import annotations |
| 4 | |
| 5 | from datetime import UTC, datetime, timedelta |
| 6 | import time |
| 7 | import unittest |
| 8 | |
| 9 | from scooling_lab_helpers import PROJECT_ROOT, valid_payload |
| 10 | |
| 11 | from scooling_lab.bom import main as bom_main |
| 12 | from scooling_lab.service import TrainingApiService |
| 13 | from scooling_lab.store import TrainingJobStore |
| 14 | |
| 15 | |
| 16 | class ScoolingLabPerformanceTests(unittest.TestCase): |
| 17 | """Performance tests for status/list endpoints and BOM scan budget.""" |
| 18 | |
| 19 | def test_performance_status_and_list_artifacts_are_bounded(self) -> None: |
| 20 | """Repeated status and artifact reads remain inside a local budget.""" |
| 21 | |
| 22 | service = TrainingApiService(TrainingJobStore()) |
| 23 | created = service.create_training_job(valid_payload("performance")) |
| 24 | job_id = str(created["id"]) |
| 25 | |
| 26 | start = time.perf_counter() |
| 27 | for _ in range(200): |
| 28 | self.assertEqual(service.get_training_job(job_id)["status"], "succeeded") |
| 29 | self.assertEqual(len(service.list_artifacts(job_id)["artifacts"]), 1) |
| 30 | elapsed = time.perf_counter() - start |
| 31 | |
| 32 | self.assertLess(elapsed, 1.0) |
| 33 | |
| 34 | def test_performance_bom_scan_runs_inside_ci_budget(self) -> None: |
| 35 | """The real BOM check completes within the configured CI budget.""" |
| 36 | |
| 37 | start = time.perf_counter() |
| 38 | exit_code = bom_main( |
| 39 | [ |
| 40 | "--root", |
| 41 | str(PROJECT_ROOT), |
| 42 | "--output", |
| 43 | "DEPENDENCIES.md", |
| 44 | "--check", |
| 45 | "--budget-seconds", |
| 46 | "30", |
| 47 | ] |
| 48 | ) |
| 49 | elapsed = time.perf_counter() - start |
| 50 | |
| 51 | self.assertEqual(exit_code, 0) |
| 52 | self.assertLess(elapsed, 30) |
| 53 | |
| 54 | def test_performance_provenance_emission_is_constant_time_bounded(self) -> None: |
| 55 | """Repeated provenance emission remains inside a fixed local budget.""" |
| 56 | |
| 57 | service = TrainingApiService(TrainingJobStore(queue_limit=1_200)) |
| 58 | |
| 59 | start = time.perf_counter() |
| 60 | for index in range(1_000): |
| 61 | created = service.create_training_job(valid_payload(f"provenance-{index}")) |
| 62 | provenance = service.get_provenance(str(created["id"])) |
| 63 | self.assertTrue(str(provenance["artifactHash"])) |
| 64 | elapsed = time.perf_counter() - start |
| 65 | |
| 66 | self.assertLess(elapsed, 3.0) |
| 67 | |
| 68 | def test_performance_sweep_is_bounded_over_ten_thousand_fixture_jobs(self) -> None: |
| 69 | """Retention sweep over N=10k fixture jobs completes inside CI budget.""" |
| 70 | |
| 71 | service = TrainingApiService(TrainingJobStore(queue_limit=10_500)) |
| 72 | payload_policy = {"policyClass": "ephemeral", "ttlSeconds": 60} |
| 73 | for index in range(10_000): |
| 74 | service.create_training_job(valid_payload(f"sweep-{index}", payload_policy)) |
| 75 | |
| 76 | start = time.perf_counter() |
| 77 | summary = service.sweep_expired_artifacts( |
| 78 | datetime.now(UTC) + timedelta(seconds=120) |
| 79 | ) |
| 80 | elapsed = time.perf_counter() - start |
| 81 | |
| 82 | self.assertEqual(summary["deletedCount"], 10_000) |
| 83 | self.assertLess(elapsed, 5.0) |
| 84 | |
| 85 | def test_performance_t4_cancel_retry_and_validation_are_bounded(self) -> None: |
| 86 | """Cancel, retry, and dataset shape validation stay inside a local budget.""" |
| 87 | |
| 88 | store = TrainingJobStore(queue_limit=1_600, max_concurrent_running=2) |
| 89 | service = TrainingApiService(store, auto_run_worker=False) |
| 90 | created = [ |
| 91 | service.create_training_job(valid_payload(f"perf-t4-{index}")) |
| 92 | for index in range(1_000) |
| 93 | ] |
| 94 | |
| 95 | start = time.perf_counter() |
| 96 | for job in created[:300]: |
| 97 | service.cancel_training_job(str(job["id"])) |
| 98 | for job in created[:300]: |
| 99 | retry = service.retry_training_job(str(job["id"])) |
| 100 | self.assertEqual(retry["retryOfJobId"], job["id"]) |
| 101 | for index in range(300): |
| 102 | dataset_id = f"perf-dataset-{index}" |
| 103 | service.register_dataset( |
| 104 | { |
| 105 | "datasetId": dataset_id, |
| 106 | "rowCount": 8, |
| 107 | "declaredSchema": { |
| 108 | "exampleId": "string", |
| 109 | "inputTokenCount": "integer", |
| 110 | "outputTokenCount": "integer", |
| 111 | "split": "string", |
| 112 | }, |
| 113 | } |
| 114 | ) |
| 115 | decision = service.submit_dataset_for_review(dataset_id) |
| 116 | self.assertEqual(decision["status"], "approved") |
| 117 | elapsed = time.perf_counter() - start |
| 118 | |
| 119 | self.assertLessEqual(service.get_queue_state()["runningCount"], 2) |
| 120 | self.assertLess(elapsed, 2.0) |
| 121 | |
| 122 | |
| 123 | if __name__ == "__main__": |
| 124 | unittest.main() |