diff --git a/src/hpc/cli.py b/src/hpc/cli.py index 6987aa0..54f8e8d 100644 --- a/src/hpc/cli.py +++ b/src/hpc/cli.py @@ -14,6 +14,7 @@ from .sync import SyncManager from .job import JobManager, JobStatus from .run import RunManager +from .scheduler import SchedulerError # Type alias for config option ConfigOption = Annotated[ @@ -345,7 +346,16 @@ def status(id: str = typer.Argument(None), config: ConfigOption = None): if detail is None or not detail.state: # Scheduler does not support detail (PJM), or sacct has not yet # recorded this job. Fall back to the existing single-line display. - job_status = job_manager.get_job_status(job_id) + # SchedulerError (parse-side absence) becomes a friendly message; + # other SSHError (real transport/command failure) propagates so + # legitimate failures stay visible to the user. + try: + job_status = job_manager.get_job_status(job_id) + except SchedulerError: + print( + f"Job {job_id}: status unavailable yet (scheduler accounting not ready)" + ) + return print(f"Job {job_id}: {job_status.value}") return diff --git a/src/hpc/scheduler.py b/src/hpc/scheduler.py index b234f80..b4590cb 100644 --- a/src/hpc/scheduler.py +++ b/src/hpc/scheduler.py @@ -5,6 +5,31 @@ from dataclasses import dataclass from enum import Enum +from .ssh import SSHError + + +class SchedulerError(SSHError): + """Scheduler returned no usable response. + + Raised by ``parse_status`` when the scheduler's status command exits + cleanly but its output contains no data row from which a ``JobStatus`` + can be derived — typically a freshly-submitted job that has not yet + been indexed by the accounting database (Slurm's ``sacct``) or a job + that aged out of the active view (PJM's ``pjstat`` post-``EXT`` + window). Distinguishing this from a real ``JobStatus.FAILED`` lets + callers retry / fall through instead of treating the absence of + information as a terminal failure. + + Inherits from ``SSHError`` so the three existing transient-catch + sites in ``JobManager`` cover it without modification: + ``wait_for_job``'s polling loop (continues), ``tail_job_output``'s + pre-tail status probe (falls through to ``tail -F``), and the inner + status probe inside ``get_job_output`` (swallows so the original + "No such file" diagnostic re-raises). CLI surfaces that want to + discriminate "no data yet" from a real SSH / command failure catch + ``SchedulerError`` specifically before the generic ``SSHError``. + """ + class JobStatus(Enum): PENDING = "PENDING" @@ -149,7 +174,12 @@ def parse_status(self, output: str) -> JobStatus: if tokens: lines.append(tokens[0].rstrip("+")) if not lines: - return JobStatus.FAILED + # No parseable row — sacct has not yet indexed the job (fresh + # submission, accounting lag). Surface as a transient absence + # rather than synthesizing a terminal FAILED. + raise SchedulerError( + "sacct returned no data row; job may not yet be indexed" + ) statuses = [_STATUS_MAP.get(s, JobStatus.FAILED) for s in lines] for status in ( JobStatus.RUNNING, @@ -298,9 +328,9 @@ def parse_status(self, output: str) -> JobStatus: # 48969021 EXT 0 0 # The first parseable row wins; multi-row aggregation for array # / step jobs is tracked separately (Issue #12) and out of scope - # here. An empty / header-only response falls back to ``FAILED``, - # the same fallback the pre-EC/SN parser used (Issue #8 owns - # the principled fix). + # here. An empty / header-only / short-row-only response raises + # ``SchedulerError`` so callers can distinguish "no data yet" + # from a real terminal failure. for raw in output.splitlines(): tokens = raw.split() if len(tokens) < 4 or tokens[0] == "JOB_ID": @@ -311,7 +341,10 @@ def parse_status(self, output: str) -> JobStatus: JobStatus.COMPLETED if ec == "0" and sn == "0" else JobStatus.FAILED ) return self._STATUS_MAP.get(st, JobStatus.FAILED) - return JobStatus.FAILED + raise SchedulerError( + "pjstat returned no data row; job may not yet be indexed" + " or has aged out of the active view" + ) # Slurm job state strings as reported by `sacct`. Unknown states fall back diff --git a/tests/test_cli.py b/tests/test_cli.py index 2164e2f..c9b7b01 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -139,6 +139,29 @@ def test_status_falls_back_when_detail_unavailable(cli_runner, temp_dir, monkeyp assert "ExitCode" not in result.stdout +def test_status_prints_friendly_message_when_status_unavailable( + cli_runner, temp_dir, monkeypatch +): + """SchedulerError on the fallback ``get_job_status`` becomes a friendly + "status unavailable yet" line instead of an SSHError stack.""" + from hpc.scheduler import SchedulerError + + monkeypatch.chdir(temp_dir) + cli_runner.invoke(app, ["init"]) + + with patch("hpc.cli.JobManager") as MockJobManager: + instance = MockJobManager.return_value + instance.get_job_detail.return_value = None + instance.get_job_status.side_effect = SchedulerError( + "sacct returned no data row" + ) + result = cli_runner.invoke(app, ["status", "12345678"]) + + assert result.exit_code == 0 + assert "Job 12345678: status unavailable yet" in result.stdout + assert "scheduler accounting not ready" in result.stdout + + def test_status_normalizes_decorated_cancelled_state(cli_runner, temp_dir, monkeypatch): """``CANCELLED+`` and ``CANCELLED by 12345`` should still trigger detail rendering.""" from hpc.scheduler import JobDetail diff --git a/tests/test_job.py b/tests/test_job.py index 899d3f8..f83315a 100644 --- a/tests/test_job.py +++ b/tests/test_job.py @@ -6,7 +6,7 @@ import pytest from hpc.job import JobManager, JobStatus, _extract_prologue_directives -from hpc.scheduler import JobDetail +from hpc.scheduler import JobDetail, SchedulerError from hpc.ssh import SSHManager, SSHError from hpc.config import HpcConfig, ClusterConfig, EnvConfig, SlurmConfig, PjmConfig @@ -290,6 +290,19 @@ def test_get_job_status_uses_sacct(self, mock_ssh_manager, sample_config): assert "12345678" in call_args.args[1] assert "--noheader" in call_args.args[1] + def test_get_job_status_raises_scheduler_error_on_empty_output( + self, mock_ssh_manager, sample_config + ): + # Boundary pin: the SchedulerError raised by parse_status reaches + # JobManager's public API unchanged (no conversion to a different + # type). Since SchedulerError is an SSHError subclass, the three + # transient-handling callers still catch it via `except SSHError`. + manager = JobManager(ssh_manager=mock_ssh_manager, config=sample_config) + mock_ssh_manager.run_command.return_value = MagicMock(stdout="") + + with pytest.raises(SchedulerError): + manager.get_job_status("12345678") + class TestJobManagerDetail: def test_get_job_detail_slurm_invokes_sacct_and_returns_detail( @@ -428,6 +441,26 @@ def test_pjm_error_flag_uses_err_extension(self, mock_ssh_manager): call_args = mock_ssh_manager.run_command.call_args assert call_args.args[1] == ["/scratch/user/proj/.hpc/runs/test_run/job.err"] + def test_inner_scheduler_error_does_not_mask_original_no_such_file( + self, mock_ssh_manager, sample_config + ): + # Regression guard: when ``cat`` raises "No such file" and the + # inner status probe raises SchedulerError (job not yet indexed), + # the existing ``except SSHError: pass`` swallows the inner + # SchedulerError (subclass of SSHError) and re-raises the original + # cat error. The user must see the original missing-file + # diagnostic, not the inner accounting-not-ready one. + manager = JobManager(ssh_manager=mock_ssh_manager, config=sample_config) + mock_ssh_manager.run_command.side_effect = [ + SSHError("SSH command failed: cat: No such file or directory"), + MagicMock(stdout=""), # status probe; parse_status raises SchedulerError + ] + + with pytest.raises(SSHError) as exc_info: + manager.get_job_output("test_run", "12345678") + assert "No such file" in str(exc_info.value) + assert not isinstance(exc_info.value, SchedulerError) + class TestJobManagerTailJobOutput: def _running_status(self): diff --git a/tests/test_scheduler.py b/tests/test_scheduler.py index 4071d9e..3847d1d 100644 --- a/tests/test_scheduler.py +++ b/tests/test_scheduler.py @@ -1,6 +1,9 @@ """Scheduler unit tests.""" -from hpc.scheduler import PJM, JobDetail, JobStatus, Slurm +import pytest + +from hpc.scheduler import PJM, JobDetail, JobStatus, SchedulerError, Slurm +from hpc.ssh import SSHError class TestSlurmStatusCmd: @@ -97,15 +100,18 @@ def test_aggregation_strips_plus_on_every_row(self): output = "COMPLETED+\nCOMPLETED+\nCOMPLETED+\n" assert Slurm().parse_status(output) == JobStatus.COMPLETED - # Empty / unknown — empty output preserves the pre-existing FAILED - # behavior; unknown states fall back to FAILED so we stay + # Empty / unknown — empty input raises SchedulerError so callers can + # distinguish "no data yet" (accounting lag) from a real terminal + # FAILED. Unknown but parseable states still map to FAILED to stay # conservative with respect to wait termination. - def test_empty_output(self): - assert Slurm().parse_status("") == JobStatus.FAILED + def test_empty_output_raises_scheduler_error(self): + with pytest.raises(SchedulerError): + Slurm().parse_status("") - def test_whitespace_only_output(self): - assert Slurm().parse_status(" \n \n") == JobStatus.FAILED + def test_whitespace_only_output_raises_scheduler_error(self): + with pytest.raises(SchedulerError): + Slurm().parse_status(" \n \n") def test_unknown_status_falls_back_to_failed(self): assert Slurm().parse_status("BOGUS_STATE\n") == JobStatus.FAILED @@ -235,18 +241,18 @@ def test_ccl_maps_to_cancelled(self): def test_rjt_maps_to_failed(self): assert PJM().parse_status(self._row("RJT")) == JobStatus.FAILED - def test_empty_output_falls_back_to_failed(self): - # Regression guard for the deferred Issue #8: today, empty - # pjstat output is conflated with a true failure. The eventual - # fix should be a deliberate behavior change rather than an - # accidental one. - assert PJM().parse_status("") == JobStatus.FAILED + def test_empty_output_raises_scheduler_error(self): + # Empty pjstat output is a structural absence (accounting lag / + # aged-out job), not a real terminal failure; raise so callers + # treat it as transient. + with pytest.raises(SchedulerError): + PJM().parse_status("") - def test_header_only_output_falls_back_to_failed(self): - # Same deferred-#8 guard, header-only shape: ``pjstat`` exited - # cleanly but produced no data row (job aged out of the active - # view's post-EXT window). - assert PJM().parse_status(self._HEADER) == JobStatus.FAILED + def test_header_only_output_raises_scheduler_error(self): + # ``pjstat`` exited cleanly but produced no data row (job aged + # out of the active view's post-EXT window). + with pytest.raises(SchedulerError): + PJM().parse_status(self._HEADER) def test_multi_row_input_takes_first_parseable_row(self): # Documents today's single-row semantics so the eventual array @@ -261,6 +267,22 @@ def test_short_row_skipped_before_data_row(self): output = f"{self._HEADER}short\n48971221 EXT 0 0\n" assert PJM().parse_status(output) == JobStatus.COMPLETED + def test_unknown_st_falls_back_to_failed(self): + # An unrecognized but structurally-valid ST token (4 data + # tokens, non-``EXT``) stays on the conservative FAILED + # fallback rather than raising — only structural absence + # raises SchedulerError. + assert PJM().parse_status(self._row("ZZZ")) == JobStatus.FAILED + + +class TestSchedulerErrorInheritance: + def test_scheduler_error_is_ssh_error_subclass(self): + # The three callers in JobManager (wait_for_job, tail_job_output, + # get_job_output) catch SSHError to apply transient/unknown + # handling. SchedulerError must inherit so those existing + # handlers cover it without modification. + assert issubclass(SchedulerError, SSHError) + class TestPJMParseJobID: def test_parse_job_id_prefers_job_token(self): diff --git a/tests/test_wait.py b/tests/test_wait.py index cb9b73c..3fd7dc6 100644 --- a/tests/test_wait.py +++ b/tests/test_wait.py @@ -53,6 +53,25 @@ def test_wait_returns_on_failed(self, mock_ssh_manager, sample_config): status = manager.wait_for_job("12345678", interval=0.01) assert status == JobStatus.FAILED + def test_wait_keeps_polling_on_empty_status_output( + self, mock_ssh_manager, sample_config + ): + # Regression guard for https://github.com/ultimatile/hpc/issues/13: + # a freshly-submitted job whose sacct row hasn't been indexed yet + # returns empty stdout. The old parse_status mapped that to FAILED + # and wait_for_job exited prematurely; SchedulerError (subclass + # of SSHError) now routes through the retry path so polling + # continues. + manager = JobManager(ssh_manager=mock_ssh_manager, config=sample_config) + mock_ssh_manager.run_command.side_effect = [ + MagicMock(stdout=""), + MagicMock(stdout="COMPLETED\n"), + ] + + status = manager.wait_for_job("12345678", interval=0.01) + assert status == JobStatus.COMPLETED + assert mock_ssh_manager.run_command.call_count == 2 + def test_wait_adaptive_interval(self, mock_ssh_manager, sample_config): manager = JobManager(ssh_manager=mock_ssh_manager, config=sample_config) mock_ssh_manager.run_command.side_effect = [