test_progress.py

Go to the documentation of this file.

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# tests/tests_core/test_progress.py
 
"""
Purpose
-------
Regression tests for :class:`oct.core.progress.ProgressTracker` and
:class:`oct.core.progress.Dashboard`.
 
Responsibilities
----------------
- Verify ProgressTracker lifecycle (start, complete, fail, skip) and markers.
- Verify Dashboard in-place updates, timer, quiet mode, and non-TTY fallback.
 
Diagnostics
-----------
Domain: CORE-TESTS
Levels:
    L2 — test lifecycle
    L3 — assertion details
    L4 — deep tracing
 
Contracts
---------
- All output is captured via ``io.StringIO``; no real terminal writes.
"""
 
import io
import threading
import time
 
from oct.core.progress import Dashboard, ProgressTracker
 
 
def _make_tracker(tasks, **kwargs):
    buf = io.StringIO()
    kwargs.setdefault("stream", buf)
    return ProgressTracker(tasks, **kwargs), buf
 
 
class TestBasicLifecycle:
 
    def test_basic_lifecycle(self):
        tracker, buf = _make_tracker(["A", "B", "C"])
        for name in ("A", "B", "C"):
            tracker.start_task(name)
            tracker.complete_task(name, summary=f"{name} done")
        output = buf.getvalue()
        assert "[x] A  A done" in output
        assert "[x] B  B done" in output
        assert "[x] C  C done" in output
 
    def test_fail_task(self):
        tracker, buf = _make_tracker(["X"])
        tracker.start_task("X")
        tracker.fail_task("X", summary="TIMEOUT")
        assert "[!] X  TIMEOUT" in buf.getvalue()
 
    def test_skip_task(self):
        tracker, buf = _make_tracker(["X"])
        tracker.start_task("X")
        tracker.skip_task("X", summary="skipped")
        assert "[-] X  skipped" in buf.getvalue()
 
 
class TestContextManager:
 
    def test_context_manager_success(self):
        tracker, buf = _make_tracker(["T"])
        with tracker.task("T"):
            pass
        assert "[x] T" in buf.getvalue()
 
    def test_context_manager_failure(self):
        tracker, buf = _make_tracker(["T"])
        try:
            with tracker.task("T"):
                raise ValueError("boom")
        except ValueError:
            pass
        assert "[!] T" in buf.getvalue()
 
 
class TestQuietMode:
 
    def test_quiet_mode(self):
        tracker, buf = _make_tracker(["A"], quiet=True)
        tracker.start_task("A")
        tracker.complete_task("A", summary="done")
        tracker.finish()
        assert buf.getvalue() == ""
 
 
class TestThreadSafety:
 
    def test_thread_safety(self):
        tracker, buf = _make_tracker(["Counter"])
        tracker.start_task("Counter", total=1000)
 
        def tick_100():
            for _ in range(100):
                tracker.tick()
 
        threads = [threading.Thread(target=tick_100) for _ in range(10)]
        for t in threads:
            t.start()
        for t in threads:
            t.join()
 
        assert tracker._progress == 1000
 
 
class TestIndeterminate:
 
    def test_indeterminate_mode(self):
        tracker, buf = _make_tracker(["Slow"])
        tracker.start_task("Slow", total=0)
        output = buf.getvalue()
        assert "[start] Slow" in output
 
    def test_tick_zero_no_advance(self):
        tracker, buf = _make_tracker(["X"])
        tracker.start_task("X", total=10)
        tracker.tick(0)
        assert tracker._progress == 0
 
 
class TestNonTTY:
 
    def test_non_tty_simple_lines(self):
        tracker, buf = _make_tracker(["A"])
        tracker.start_task("A")
        tracker.complete_task("A", summary="ok")
        output = buf.getvalue()
        assert "[start] A" in output
        assert "[x] A  ok" in output
        assert "\033" not in output
 
 
class TestMisc:
 
    def test_unknown_task_name_ignored(self):
        tracker, buf = _make_tracker(["A"])
        tracker.complete_task("nonexistent")
        tracker.fail_task("nonexistent")
        tracker.skip_task("nonexistent")
        assert buf.getvalue() == ""
 
    def test_title_printed(self):
        tracker, buf = _make_tracker(["A"], title="oct health")
        assert "oct health" in buf.getvalue()
 
    def test_finish_clears_pending(self):
        tracker, buf = _make_tracker(["A", "B"])
        tracker.start_task("A")
        tracker.complete_task("A")
        tracker.finish()
        output = buf.getvalue()
        assert "[x] A" in output
        assert "[ ] B" in output
 
 
# ============================================================================
# Dashboard tests
# ============================================================================
 
 
def _make_dashboard(**kwargs):
    buf = io.StringIO()
    kwargs.setdefault("stream", buf)
    return Dashboard(**kwargs), buf
 
 
class TestDashboardRender:
 
    def test_render_stores_lines(self):
        db, buf = _make_dashboard()
        db.render(["line0", "line1", "line2"], tasks_total=1)
        assert len(db._lines) == 3
 
    def test_non_tty_no_output_on_render(self):
        db, buf = _make_dashboard()
        db.render(["aaa", "bbb"], tasks_total=1)
        assert buf.getvalue() == ""
 
    def test_non_tty_prints_final_state_on_finish(self):
        db, buf = _make_dashboard()
        db.render(["[ ] A", "footer", "==="], footer_line=1, tasks_total=1)
        db.update_line(0, "[x] A done")
        db.complete_task()
        db.finish()
        output = buf.getvalue()
        assert "[x] A done" in output
        assert "===" in output
 
 
class TestDashboardUpdateLine:
 
    def test_update_line_changes_internal_state(self):
        db, buf = _make_dashboard()
        db.render(["old0", "old1", "old2"], tasks_total=1)
        db.update_line(1, "new1")
        assert db._lines[1] == "new1"
 
    def test_update_out_of_range_ignored(self):
        db, buf = _make_dashboard()
        db.render(["a", "b"], tasks_total=1)
        db.update_line(5, "nope")
        db.update_line(-1, "nope")
        assert db._lines == ["a", "b"]
 
 
class TestDashboardQuiet:
 
    def test_quiet_suppresses_all_output(self):
        db, buf = _make_dashboard(quiet=True)
        db.render(["a", "b", "c"], tasks_total=1)
        db.set_task("X")
        db.update_line(0, "changed")
        db.complete_task()
        db.finish()
        assert buf.getvalue() == ""
 
 
class TestDashboardTaskTracking:
 
    def test_tasks_done_increments(self):
        db, buf = _make_dashboard()
        db.render(["a", "footer", "end"], footer_line=1, tasks_total=3)
        db.complete_task()
        db.complete_task()
        assert db._tasks_done == 2
 
    def test_set_task_stores_name(self):
        db, buf = _make_dashboard()
        db.render(["a", "footer", "end"], footer_line=1, tasks_total=1)
        db.set_task("Test suite")
        assert db._task_name == "Test suite"
 
    def test_complete_task_clears_name(self):
        db, buf = _make_dashboard()
        db.render(["a", "footer", "end"], footer_line=1, tasks_total=1)
        db.set_task("X")
        db.complete_task()
        assert db._task_name is None
 
 
class TestDashboardFinish:
 
    def test_finish_updates_footer(self):
        db, buf = _make_dashboard()
        db.render(["header", "footer", "==="], footer_line=1, tasks_total=2)
        db.complete_task()
        db.complete_task()
        db.finish()
        output = buf.getvalue()
        assert "All 2 checks completed" in output
 
    def test_finish_stops_timer_thread(self):
        db, buf = _make_dashboard()
        db.render(["a", "b", "c"], tasks_total=1)
        db.finish()
        assert db._timer_thread is None
 
 
class TestDashboardTimer:
 
    def test_footer_includes_elapsed(self):
        db, buf = _make_dashboard()
        db.render(["header", "footer", "==="], footer_line=1, tasks_total=1)
        db.set_task("Slow")
        time.sleep(0.05)
        db._update_footer()
        assert "Slow" in db._lines[1]
        assert "elapsed" in db._lines[1]
        db.finish()