diff --git a/python/lightning_sdk/cli/job/__init__.py b/python/lightning_sdk/cli/job/__init__.py index d527bb08..24e4cfe7 100644 --- a/python/lightning_sdk/cli/job/__init__.py +++ b/python/lightning_sdk/cli/job/__init__.py @@ -8,11 +8,13 @@ def register_commands(group: click.Group) -> None: from lightning_sdk.cli.job.delete import delete_job from lightning_sdk.cli.job.inspect import inspect_job from lightning_sdk.cli.job.list import list_jobs + from lightning_sdk.cli.job.logs import logs_job from lightning_sdk.cli.job.run import run_job from lightning_sdk.cli.job.stop import stop_job group.add_command(run_job, name="run") group.add_command(list_jobs, name="list") group.add_command(inspect_job, name="inspect") + group.add_command(logs_job, name="logs") group.add_command(stop_job, name="stop") group.add_command(delete_job, name="delete") diff --git a/python/lightning_sdk/cli/job/logs.py b/python/lightning_sdk/cli/job/logs.py new file mode 100644 index 00000000..07590f0d --- /dev/null +++ b/python/lightning_sdk/cli/job/logs.py @@ -0,0 +1,38 @@ +"""Job logs command.""" + +from typing import Optional + +import rich_click as click + +from lightning_sdk.cli.legacy.job_and_mmt_action import _JobAndMMTAction +from lightning_sdk.cli.utils.logging import LightningCommand +from lightning_sdk.status import Status + +_TERMINAL = (Status.Completed, Status.Failed, Status.Stopped) + + +@click.command("logs", cls=LightningCommand) +@click.argument("name", required=False) +@click.option( + "--teamspace", + default=None, + help=( + "the name of the teamspace the job lives in. " + "Should be specified as {teamspace_owner}/{teamspace_name} (e.g my-org/my-teamspace). " + "If not specified can be selected interactively." + ), +) +def logs_job(name: Optional[str] = None, teamspace: Optional[str] = None) -> None: + """Print the logs for a job. + + Logs are available once the job reaches a terminal state (Completed, Failed or + Stopped). While the job is still pending or running this prints its current status + instead of erroring — re-run once it has finished. + """ + job = _JobAndMMTAction().job(name=name, teamspace=teamspace) + if job.status not in _TERMINAL: + raise click.ClickException( + f"Job '{job.name}' is {job.status}; logs are only available once it reaches a " + "terminal state (Completed/Failed/Stopped). Re-run this command after it finishes." + ) + click.echo(job.logs) diff --git a/python/tests/cli/job/test_logs.py b/python/tests/cli/job/test_logs.py new file mode 100644 index 00000000..f94e2512 --- /dev/null +++ b/python/tests/cli/job/test_logs.py @@ -0,0 +1,76 @@ +from unittest.mock import MagicMock + +from click.testing import CliRunner + +from lightning_sdk.status import Status +from tests.cli.help import assert_help_contains, mock_command_logging + + +@mock_command_logging +def test_job_logs_help() -> None: + assert_help_contains( + "lightning job logs --help", + "Usage: lightning job logs", + "Print the logs for a job.", + ) + + +@mock_command_logging +def test_jobs_logs_help() -> None: + assert_help_contains( + "lightning jobs logs --help", + "Usage: lightning jobs logs", + "Print the logs for a job.", + ) + + +@mock_command_logging +def test_job_logs_help_shows_positional_name() -> None: + assert_help_contains("lightning job logs --help", "Usage: lightning job logs [OPTIONS] [NAME]") + + +def _patch_action(monkeypatch, job: MagicMock, captured: dict) -> None: + class _FakeJobAndMMTAction: + def job(self, name=None, teamspace=None): + captured["name"] = name + captured["teamspace"] = teamspace + return job + + monkeypatch.setattr("lightning_sdk.cli.job.logs._JobAndMMTAction", _FakeJobAndMMTAction) + + +@mock_command_logging +def test_job_logs_prints_logs_when_terminal(monkeypatch) -> None: + from lightning_sdk.cli.job.logs import logs_job + + captured: dict = {} + job = MagicMock() + job.name = "my-job" + job.status = Status.Completed + job.logs = "hello from the job\n42" + _patch_action(monkeypatch, job, captured) + + result = CliRunner().invoke(logs_job, ["my-job", "--teamspace", "org/teamspace"]) + + assert result.exit_code == 0 + assert captured == {"name": "my-job", "teamspace": "org/teamspace"} + assert "hello from the job" in result.output + assert "42" in result.output + + +@mock_command_logging +def test_job_logs_errors_while_not_terminal(monkeypatch) -> None: + from lightning_sdk.cli.job.logs import logs_job + + captured: dict = {} + job = MagicMock() + job.name = "my-job" + job.status = Status.Pending + _patch_action(monkeypatch, job, captured) + + result = CliRunner().invoke(logs_job, ["my-job", "--teamspace", "org/teamspace"]) + + # errors cleanly (mentioning the status) rather than raising the raw SDK RuntimeError + assert result.exit_code != 0 + assert "Pending" in result.output + assert not isinstance(result.exception, RuntimeError)