File size: 2,072 Bytes
e0265b9 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 | from __future__ import annotations
import logging
from pathlib import Path
from adam.job_manager import JobManager
from adam.models import ExecutionPlan, Job, JobStatus
def _job(index: int, status: JobStatus = JobStatus.FINISHED) -> Job:
return Job(
id=f"JOB{index:04d}",
plan=ExecutionPlan(
request="test",
summary="Test job",
steps=[],
project_name=f"Project {index}",
),
status=status,
)
def test_job_history_is_not_silently_truncated(tmp_path: Path) -> None:
manager = JobManager(tmp_path, None, logging.getLogger("test.jobs")) # type: ignore[arg-type]
manager.jobs = [_job(index) for index in range(125)]
manager._save()
restored = JobManager(tmp_path, None, logging.getLogger("test.jobs")) # type: ignore[arg-type]
assert len(restored.jobs) == 125
def test_history_cleanup_only_removes_completed_and_failed(tmp_path: Path) -> None:
manager = JobManager(tmp_path, None, logging.getLogger("test.jobs")) # type: ignore[arg-type]
manager.jobs = [
_job(1, JobStatus.FINISHED),
_job(2, JobStatus.FAILED),
_job(3, JobStatus.CANCELLED),
_job(4, JobStatus.INTERRUPTED),
]
assert manager.remove_completed_or_failed() == 2
assert [job.status for job in manager.jobs] == [
JobStatus.CANCELLED,
JobStatus.INTERRUPTED,
]
def test_end_task_acknowledges_an_interrupted_job(tmp_path: Path) -> None:
manager = JobManager(tmp_path, None, logging.getLogger("test.jobs")) # type: ignore[arg-type]
job = _job(1, JobStatus.INTERRUPTED)
job.logs.append("[startup] Previous session ended before this job. Review it before retrying.")
manager.jobs = [job]
assert manager.end_task(job.id) is True
assert job.status == JobStatus.CANCELLED
assert not any("Previous session ended" in line for line in job.logs)
restored = JobManager(tmp_path, None, logging.getLogger("test.jobs")) # type: ignore[arg-type]
assert restored.jobs[0].status == JobStatus.CANCELLED
|