test_stress.py
file-level
1
files
1
commits
0
hotspots
0
🧊 dead
0
💥 blast risk
| 1 | """Stress tier tests for duplicate creation and fixture queue limits.""" |
| 2 | |
| 3 | from __future__ import annotations |
| 4 | |
| 5 | import concurrent.futures |
| 6 | from datetime import UTC, datetime, timedelta |
| 7 | import unittest |
| 8 | |
| 9 | from scooling_lab_helpers import valid_payload |
| 10 | |
| 11 | from scooling_lab.contracts import TrainingJobStatus |
| 12 | from scooling_lab.errors import ApiError, ErrorCode |
| 13 | from scooling_lab.service import TrainingApiService |
| 14 | from scooling_lab.store import TrainingJobStore |
| 15 | |
| 16 | |
| 17 | class ScoolingLabStressTests(unittest.TestCase): |
| 18 | """Stress tests for deterministic idempotency and bounded queues.""" |
| 19 | |
| 20 | def test_stress_concurrent_duplicate_create_is_deduplicated(self) -> None: |
| 21 | """Many concurrent duplicate creates return one deterministic job id.""" |
| 22 | |
| 23 | service = TrainingApiService(TrainingJobStore(), auto_run_worker=False) |
| 24 | payload = valid_payload("duplicate") |
| 25 | |
| 26 | with concurrent.futures.ThreadPoolExecutor(max_workers=8) as executor: |
| 27 | results = list( |
| 28 | executor.map(lambda _: service.create_training_job(payload), range(24)) |
| 29 | ) |
| 30 | |
| 31 | job_ids = {result["id"] for result in results} |
| 32 | self.assertEqual(len(job_ids), 1) |
| 33 | self.assertEqual(results[0]["status"], "queued") |
| 34 | |
| 35 | def test_stress_many_fixture_jobs_respect_queue_limit(self) -> None: |
| 36 | """The fixture queue rejects excess queued/running jobs deterministically.""" |
| 37 | |
| 38 | service = TrainingApiService( |
| 39 | TrainingJobStore(queue_limit=2), auto_run_worker=False |
| 40 | ) |
| 41 | service.create_training_job(valid_payload("queue-a")) |
| 42 | service.create_training_job(valid_payload("queue-b")) |
| 43 | |
| 44 | with self.assertRaises(ApiError) as raised: |
| 45 | service.create_training_job(valid_payload("queue-c")) |
| 46 | self.assertEqual(raised.exception.code, ErrorCode.QUEUE_LIMIT_EXCEEDED) |
| 47 | |
| 48 | def test_stress_many_jobs_interleave_deletes_and_sweeps(self) -> None: |
| 49 | """Many completed jobs can be deleted and swept without resurrecting data.""" |
| 50 | |
| 51 | service = TrainingApiService(TrainingJobStore(queue_limit=256)) |
| 52 | job_ids: list[str] = [] |
| 53 | artifact_ids: list[str] = [] |
| 54 | for index in range(120): |
| 55 | policy = {"policyClass": "ephemeral", "ttlSeconds": 60} |
| 56 | created = service.create_training_job(valid_payload(f"stress-{index}", policy)) |
| 57 | job_id = str(created["id"]) |
| 58 | artifact = service.list_artifacts(job_id)["artifacts"][0] |
| 59 | job_ids.append(job_id) |
| 60 | artifact_ids.append(str(artifact["id"])) |
| 61 | |
| 62 | for job_id, artifact_id in zip(job_ids[::3], artifact_ids[::3], strict=True): |
| 63 | service.delete_artifact(job_id, artifact_id) |
| 64 | summary = service.sweep_expired_artifacts(datetime.now(UTC) + timedelta(seconds=120)) |
| 65 | |
| 66 | self.assertGreater(summary["deletedCount"], 0) |
| 67 | for job_id in job_ids: |
| 68 | self.assertEqual(service.get_training_job(job_id)["status"], "deleted") |
| 69 | self.assertEqual(service.list_artifacts(job_id)["artifacts"], []) |
| 70 | |
| 71 | def test_stress_concurrent_delete_and_read_never_resurrects_data(self) -> None: |
| 72 | """Concurrent delete and read operations settle on a deleted tombstone.""" |
| 73 | |
| 74 | service = TrainingApiService(TrainingJobStore()) |
| 75 | created = service.create_training_job(valid_payload("stress-race")) |
| 76 | job_id = str(created["id"]) |
| 77 | artifact_id = str(service.list_artifacts(job_id)["artifacts"][0]["id"]) |
| 78 | |
| 79 | def delete_once() -> str: |
| 80 | result = service.delete_artifact(job_id, artifact_id) |
| 81 | return str(result["deleted"]) |
| 82 | |
| 83 | def read_once() -> str: |
| 84 | try: |
| 85 | return str(service.get_training_job(job_id)["status"]) |
| 86 | except ApiError as error: |
| 87 | return error.code.value |
| 88 | |
| 89 | with concurrent.futures.ThreadPoolExecutor(max_workers=8) as executor: |
| 90 | futures = [ |
| 91 | executor.submit(delete_once if index % 2 == 0 else read_once) |
| 92 | for index in range(40) |
| 93 | ] |
| 94 | results = [future.result() for future in futures] |
| 95 | |
| 96 | self.assertIn("True", results) |
| 97 | self.assertEqual(service.get_training_job(job_id)["status"], "deleted") |
| 98 | self.assertEqual(service.list_artifacts(job_id)["artifacts"], []) |
| 99 | |
| 100 | def test_stress_t4_many_cancels_and_retries_preserve_slot_accounting(self) -> None: |
| 101 | """Concurrent cancel/retry waves never exceed the running concurrency bound.""" |
| 102 | |
| 103 | store = TrainingJobStore(queue_limit=160, max_concurrent_running=3) |
| 104 | service = TrainingApiService(store, auto_run_worker=False) |
| 105 | created = [ |
| 106 | service.create_training_job(valid_payload(f"stress-t4-{index}")) |
| 107 | for index in range(24) |
| 108 | ] |
| 109 | running_ids = [str(job["id"]) for job in created[:3]] |
| 110 | for job_id in running_ids: |
| 111 | store.update_status(job_id, TrainingJobStatus.RUNNING) |
| 112 | |
| 113 | with concurrent.futures.ThreadPoolExecutor(max_workers=12) as executor: |
| 114 | cancelled = list(executor.map(service.cancel_training_job, running_ids)) |
| 115 | retried = list( |
| 116 | executor.map( |
| 117 | service.retry_training_job, |
| 118 | [str(job["id"]) for job in cancelled], |
| 119 | ) |
| 120 | ) |
| 121 | |
| 122 | queue_state = service.get_queue_state() |
| 123 | self.assertEqual([job["status"] for job in cancelled], ["cancelled"] * 3) |
| 124 | self.assertEqual([job["status"] for job in retried], ["queued"] * 3) |
| 125 | self.assertLessEqual(queue_state["runningCount"], 3) |
| 126 | self.assertEqual(queue_state["runningCount"], 3) |
| 127 | self.assertEqual( |
| 128 | queue_state["activeCount"], |
| 129 | queue_state["queuedCount"] + queue_state["runningCount"], |
| 130 | ) |
| 131 | |
| 132 | |
| 133 | if __name__ == "__main__": |
| 134 | unittest.main() |