test_job_registry.py

#!/usr/bin/env python3
from job_manager import JobManager
import time
import json


def test_job_lifecycle():
    """Test complete job lifecycle."""
    manager = JobManager()

    # Create a new job named "data_processing_job"
    job_id = manager.create_job("data_processing_job")

    # Print job_id and initial status
    initial_status = manager.get_job_status(job_id)
    print(f"Created job with ID: {job_id}")
    print("Initial job status:")
    print(json.dumps(initial_status, default=str, indent=2))

    # Start the job
    started = manager.start_job(job_id)
    print(f"Job started: {started}")

    # Simulate work with time.sleep(2)
    time.sleep(2)

    # Complete job with sample results
    # Sample result: {"records_processed": 1500, "errors": 0}
    result_data = {"records_processed": 1500, "errors": 0}
    completed = manager.complete_job(job_id, result_data)
    print(f"Job completed: {completed}")

    # Retrieve and print final job status
    final_status = manager.get_job_status(job_id)
    print("Final job status:")
    print(json.dumps(final_status, default=str, indent=2))

    # Calculate and print job duration
    duration = manager.get_job_duration(job_id)
    print(f"Job duration: {duration} seconds")

    manager.close()


def test_failed_job():
    """Test job failure scenario."""
    manager = JobManager()

    # Create a new job named "failing_job"
    job_id = manager.create_job("failing_job")
    print(f"Created failing job with ID: {job_id}")

    # Start the job
    started = manager.start_job(job_id)
    print(f"Failing job started: {started}")

    # Fail the job with error message
    failed = manager.fail_job(job_id, "Simulated processing error: unable to connect to upstream service")
    print(f"Job failed: {failed}")

    # Retrieve and print job status
    status = manager.get_job_status(job_id)
    print("Failed job status:")
    print(json.dumps(status, default=str, indent=2))

    manager.close()


def test_query_by_status():
    """Test querying jobs by status."""
    manager = JobManager()

    # Query all completed jobs
    completed_jobs = manager.get_jobs_by_status("completed")

    # Print count and details
    print(f"Completed jobs count: {len(completed_jobs)}")
    for job in completed_jobs:
        print(json.dumps(job, default=str, indent=2))

    # Query all failed jobs
    failed_jobs = manager.get_jobs_by_status("failed")

    # Print count and details
    print(f"Failed jobs count: {len(failed_jobs)}")
    for job in failed_jobs:
        print(json.dumps(job, default=str, indent=2))

    manager.close()


if __name__ == "__main__":
    print("Testing Job Lifecycle...")
    test_job_lifecycle()

    print("\nTesting Failed Job...")
    test_failed_job()

    print("\nTesting Query by Status...")
    test_query_by_status()