schrodinger.seam.testing.stu_tests.periodic_checkpointing module¶

Test periodic checkpointing and pipeline resumption from checkpoints.

This test verifies that: 1. Periodic checkpoints are created during pipeline execution 2. The pipeline can be interrupted (via nodefail) without losing work 3. A tar.zst checkpoint file is saved by the job server 4. The pipeline can be resumed from the checkpoint using seamcli 5. The resumption completes successfully

Usage:

$SCHRODINGER/run python3 -m schrodinger.seam.testing.stu_tests.periodic_checkpointing

class schrodinger.seam.testing.stu_tests.periodic_checkpointing.SlowTransform(sleep_time: float = 2.0)¶

Bases: PTransform

A transform that takes time to allow checkpoint creation.

__init__(sleep_time: float = 2.0)¶

Parameters:: sleep_time – Time to sleep per element (in seconds).

expand(pcoll)¶

schrodinger.seam.testing.stu_tests.periodic_checkpointing.get_job_spec_from_args(argv)¶

schrodinger.seam.testing.stu_tests.periodic_checkpointing.run_pipeline() → None¶: Run a pipeline long enough to create periodic checkpoints.

schrodinger.seam.testing.stu_tests.periodic_checkpointing.launch_test_job() → Job¶: Launch the test pipeline and return its job.

schrodinger.seam.testing.stu_tests.periodic_checkpointing.verify_checkpoint_exists(job: Job) → Path¶

Verify that a checkpoint tar.zst file exists for the job.

Parameters:: job – The job to check.
Returns:: Path to the checkpoint tarfile.
Raises:: RuntimeError – If no checkpoint file is found.

schrodinger.seam.testing.stu_tests.periodic_checkpointing.resume_pipeline_with_seamcli(cpt_path: Path) → None¶: Resume a checkpoint archive and wait for the new job to complete.

schrodinger.seam.testing.stu_tests.periodic_checkpointing.run_test_driver() → None¶: Interrupt the test pipeline and resume it from its checkpoint.

schrodinger.seam.testing.stu_tests.periodic_checkpointing.main() → None¶