Public Access
emil
ef3b882e5b
Post-merge / detect-type (push) Successful in 13s
Post-merge / validate-commit-msg (push) Successful in 10s
Post-merge / configure-repo (push) Successful in 28s
Post-merge / vikunja (push) Successful in 44s
Post-merge / sync-wiki (push) Successful in 58s
Post-merge / release (push) Successful in 1m7s
Post-merge / publish (push) Successful in 44s
Post-merge / badges (push) Successful in 1m6s
125 lines
4.0 KiB
Python
125 lines
4.0 KiB
Python
"""Unit tests for devx.utils.step_tracker."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import click
|
|
import pytest
|
|
from click.testing import CliRunner
|
|
|
|
from devx.utils.step_tracker import Step, StepTracker, track_steps
|
|
|
|
|
|
class TestStep:
|
|
def test_initial_status_is_pending(self) -> None:
|
|
step = Step("install")
|
|
assert step.status == "pending"
|
|
assert step.name == "install"
|
|
|
|
|
|
class TestStepTracker:
|
|
def test_begin_adds_step_as_in_progress(self) -> None:
|
|
tracker = StepTracker()
|
|
tracker.begin("install deps")
|
|
assert len(tracker.steps) == 1
|
|
assert tracker.steps[0].status == "in_progress"
|
|
|
|
def test_done_marks_last_in_progress_as_completed(self) -> None:
|
|
tracker = StepTracker()
|
|
tracker.begin("step1")
|
|
tracker.done()
|
|
assert tracker.steps[0].status == "completed"
|
|
|
|
def test_done_no_op_if_no_in_progress(self) -> None:
|
|
tracker = StepTracker()
|
|
tracker.begin("step1")
|
|
tracker.done()
|
|
tracker.done() # should not raise, no-op
|
|
assert tracker.steps[0].status == "completed"
|
|
|
|
def test_done_no_op_if_empty(self) -> None:
|
|
tracker = StepTracker()
|
|
tracker.done() # should not raise
|
|
|
|
def test_multiple_steps(self) -> None:
|
|
tracker = StepTracker()
|
|
tracker.begin("step1")
|
|
tracker.done()
|
|
tracker.begin("step2")
|
|
tracker.done()
|
|
assert len(tracker.steps) == 2
|
|
assert tracker.steps[0].status == "completed"
|
|
assert tracker.steps[1].status == "completed"
|
|
|
|
|
|
class TestTrackSteps:
|
|
def test_successful_operation(self) -> None:
|
|
runner = CliRunner()
|
|
with runner.isolation():
|
|
with track_steps() as tracker:
|
|
tracker.begin("step1")
|
|
tracker.done()
|
|
tracker.begin("step2")
|
|
tracker.done()
|
|
assert len(tracker.steps) == 2
|
|
assert all(s.status == "completed" for s in tracker.steps)
|
|
|
|
def test_exception_marks_in_progress_as_failed(self) -> None:
|
|
runner = CliRunner()
|
|
with runner.isolation():
|
|
with pytest.raises(ValueError, match="boom"):
|
|
with track_steps() as tracker:
|
|
tracker.begin("step1")
|
|
tracker.done()
|
|
tracker.begin("step2")
|
|
raise ValueError("boom")
|
|
assert tracker.steps[0].status == "completed"
|
|
assert tracker.steps[1].status == "failed"
|
|
|
|
def test_pending_step_stays_pending_on_exception(self) -> None:
|
|
runner = CliRunner()
|
|
with runner.isolation():
|
|
with pytest.raises(ValueError):
|
|
with track_steps() as tracker:
|
|
tracker.begin("step1")
|
|
tracker.done()
|
|
tracker.begin("step2")
|
|
tracker.done()
|
|
tracker.begin("step3") # in_progress
|
|
# step4 is pending (not started)
|
|
raise ValueError("oops")
|
|
assert tracker.steps[2].status == "failed"
|
|
|
|
def test_empty_operation(self) -> None:
|
|
runner = CliRunner()
|
|
with runner.isolation():
|
|
with track_steps() as tracker:
|
|
pass
|
|
assert tracker.steps == []
|
|
|
|
def test_report_printed_on_success(self) -> None:
|
|
runner = CliRunner()
|
|
result = runner.invoke(_cmd_success, [], color=False)
|
|
assert result.exit_code == 0
|
|
assert "Operation Report" in result.output
|
|
assert "step1" in result.output
|
|
|
|
def test_report_printed_on_failure(self) -> None:
|
|
runner = CliRunner()
|
|
result = runner.invoke(_cmd_failure, [], color=False)
|
|
assert result.exit_code != 0
|
|
assert "Operation Report" in result.output
|
|
|
|
|
|
@click.command()
|
|
def _cmd_success() -> None:
|
|
with track_steps() as tracker:
|
|
tracker.begin("step1")
|
|
tracker.done()
|
|
|
|
|
|
@click.command()
|
|
def _cmd_failure() -> None:
|
|
with track_steps() as tracker:
|
|
tracker.begin("step1")
|
|
raise ValueError("oops")
|