test_stress.py
python
sha256:91e875d4a97bb1e35f37992f803988d5713931f1782d870c371c50054574af22
Add fixture provenance retention deletion
Human
minor
⚠ breaking
42 days ago
| 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.errors import ApiError, ErrorCode |
| 12 | from scooling_lab.service import TrainingApiService |
| 13 | from scooling_lab.store import TrainingJobStore |
| 14 | |
| 15 | |
| 16 | class ScoolingLabStressTests(unittest.TestCase): |
| 17 | """Stress tests for deterministic idempotency and bounded queues.""" |
| 18 | |
| 19 | def test_stress_concurrent_duplicate_create_is_deduplicated(self) -> None: |
| 20 | """Many concurrent duplicate creates return one deterministic job id.""" |
| 21 | |
| 22 | service = TrainingApiService(TrainingJobStore(), auto_run_worker=False) |
| 23 | payload = valid_payload("duplicate") |
| 24 | |
| 25 | with concurrent.futures.ThreadPoolExecutor(max_workers=8) as executor: |
| 26 | results = list( |
| 27 | executor.map(lambda _: service.create_training_job(payload), range(24)) |
| 28 | ) |
| 29 | |
| 30 | job_ids = {result["id"] for result in results} |
| 31 | self.assertEqual(len(job_ids), 1) |
| 32 | self.assertEqual(results[0]["status"], "queued") |
| 33 | |
| 34 | def test_stress_many_fixture_jobs_respect_queue_limit(self) -> None: |
| 35 | """The fixture queue rejects excess queued/running jobs deterministically.""" |
| 36 | |
| 37 | service = TrainingApiService( |
| 38 | TrainingJobStore(queue_limit=2), auto_run_worker=False |
| 39 | ) |
| 40 | service.create_training_job(valid_payload("queue-a")) |
| 41 | service.create_training_job(valid_payload("queue-b")) |
| 42 | |
| 43 | with self.assertRaises(ApiError) as raised: |
| 44 | service.create_training_job(valid_payload("queue-c")) |
| 45 | self.assertEqual(raised.exception.code, ErrorCode.QUEUE_LIMIT_EXCEEDED) |
| 46 | |
| 47 | def test_stress_many_jobs_interleave_deletes_and_sweeps(self) -> None: |
| 48 | """Many completed jobs can be deleted and swept without resurrecting data.""" |
| 49 | |
| 50 | service = TrainingApiService(TrainingJobStore(queue_limit=256)) |
| 51 | job_ids: list[str] = [] |
| 52 | artifact_ids: list[str] = [] |
| 53 | for index in range(120): |
| 54 | policy = {"policyClass": "ephemeral", "ttlSeconds": 60} |
| 55 | created = service.create_training_job(valid_payload(f"stress-{index}", policy)) |
| 56 | job_id = str(created["id"]) |
| 57 | artifact = service.list_artifacts(job_id)["artifacts"][0] |
| 58 | job_ids.append(job_id) |
| 59 | artifact_ids.append(str(artifact["id"])) |
| 60 | |
| 61 | for job_id, artifact_id in zip(job_ids[::3], artifact_ids[::3], strict=True): |
| 62 | service.delete_artifact(job_id, artifact_id) |
| 63 | summary = service.sweep_expired_artifacts(datetime.now(UTC) + timedelta(seconds=120)) |
| 64 | |
| 65 | self.assertGreater(summary["deletedCount"], 0) |
| 66 | for job_id in job_ids: |
| 67 | self.assertEqual(service.get_training_job(job_id)["status"], "deleted") |
| 68 | self.assertEqual(service.list_artifacts(job_id)["artifacts"], []) |
| 69 | |
| 70 | def test_stress_concurrent_delete_and_read_never_resurrects_data(self) -> None: |
| 71 | """Concurrent delete and read operations settle on a deleted tombstone.""" |
| 72 | |
| 73 | service = TrainingApiService(TrainingJobStore()) |
| 74 | created = service.create_training_job(valid_payload("stress-race")) |
| 75 | job_id = str(created["id"]) |
| 76 | artifact_id = str(service.list_artifacts(job_id)["artifacts"][0]["id"]) |
| 77 | |
| 78 | def delete_once() -> str: |
| 79 | result = service.delete_artifact(job_id, artifact_id) |
| 80 | return str(result["deleted"]) |
| 81 | |
| 82 | def read_once() -> str: |
| 83 | try: |
| 84 | return str(service.get_training_job(job_id)["status"]) |
| 85 | except ApiError as error: |
| 86 | return error.code.value |
| 87 | |
| 88 | with concurrent.futures.ThreadPoolExecutor(max_workers=8) as executor: |
| 89 | futures = [ |
| 90 | executor.submit(delete_once if index % 2 == 0 else read_once) |
| 91 | for index in range(40) |
| 92 | ] |
| 93 | results = [future.result() for future in futures] |
| 94 | |
| 95 | self.assertIn("True", results) |
| 96 | self.assertEqual(service.get_training_job(job_id)["status"], "deleted") |
| 97 | self.assertEqual(service.list_artifacts(job_id)["artifacts"], []) |
| 98 | |
| 99 | |
| 100 | if __name__ == "__main__": |
| 101 | unittest.main() |
File History
1 commit
sha256:91e875d4a97bb1e35f37992f803988d5713931f1782d870c371c50054574af22
Add fixture provenance retention deletion
Human
minor
⚠
42 days ago