tests/test_observability.py

import logging
import time
from pathlib import Path

from src.observability import (
    JobLogCapture,
    ProgressBridge,
    ProgressReporter,
    StructuredTqdmCapture,
    read_log_tail,
)


class FakeProgress:
    def __init__(self):
        self.calls = []

    def __call__(self, value, **kwargs):
        self.calls.append((value, kwargs.get("desc")))


def test_progress_bridge_emits_stages_and_completion():
    target = FakeProgress()
    bridge = ProgressBridge(target)
    bridge.update(0.25, "Loading model")
    bridge.finish("Done")
    assert target.calls[0] == (0.25, "Loading model")
    assert target.calls[-1] == (1.0, "Done")


def test_progress_reporter_is_shared_ui_and_log_source(tmp_path):
    path = tmp_path / "job.log"
    target = FakeProgress()
    bridge = ProgressBridge(target)
    with JobLogCapture(path, "abc123", level="INFO") as capture:
        reporter = ProgressReporter(bridge, capture.event, capture.debug_event, level="INFO")
        reporter.update(0.25, "Loading model", stage="model-load", force_info=True)
        reporter.finish("Done", stage="job")
    text = path.read_text(encoding="utf-8")
    assert target.calls[0] == (0.25, "Loading model")
    assert target.calls[-1] == (1.0, "Done")
    assert "stage=progress" in text
    assert "progress_stage=model-load" in text
    assert "percent=25.0" in text
    assert "progress_stage=job" in text
    assert "percent=100.0" in text


def test_info_progress_is_throttled_but_debug_keeps_updates(tmp_path):
    info_path = tmp_path / "info.log"
    with JobLogCapture(info_path, "info123", level="INFO") as capture:
        reporter = ProgressReporter(
            ProgressBridge(), capture.event, capture.debug_event, level="INFO", info_interval=999
        )
        reporter.update(0.01, "start", stage="separation")
        reporter.update(0.02, "tiny update", stage="separation")
        reporter.update(0.25, "quarter", stage="separation")
    info_text = info_path.read_text(encoding="utf-8")
    assert info_text.count("stage=progress |") == 2
    assert "stage=progress-update" not in info_text

    debug_path = tmp_path / "debug.log"
    with JobLogCapture(debug_path, "debug123", level="DEBUG") as capture:
        reporter = ProgressReporter(
            ProgressBridge(), capture.event, capture.debug_event, level="DEBUG", info_interval=999
        )
        reporter.update(0.01, "start", stage="separation")
        reporter.update(0.02, "tiny update", stage="separation")
    debug_text = debug_path.read_text(encoding="utf-8")
    assert debug_text.count("stage=progress-update") == 2


def test_job_log_capture_writes_structured_tail(tmp_path):
    path = tmp_path / "job.log"
    with JobLogCapture(path, "abc123", level="DEBUG") as capture:
        capture.event("test-stage", count=2)
        logging.getLogger("separator").debug("package detail")
    text = path.read_text(encoding="utf-8")
    assert "job=abc123" in text
    assert "stage=test-stage" in text
    assert "package detail" in text
    assert "job-log-close" in text
    assert "stage=test-stage" in read_log_tail(path)


def test_progress_heartbeat_from_helper_thread_is_captured(tmp_path):
    path = tmp_path / "heartbeat.log"
    with JobLogCapture(path, "heartbeat123", level="INFO") as capture:
        reporter = ProgressReporter(
            ProgressBridge(), capture.event, capture.debug_event, level="INFO", info_interval=0.01
        )
        with reporter.long_operation(0.4, "Separating stems", stage="separation", heartbeat_interval=0.1):
            time.sleep(0.24)
    text = path.read_text(encoding="utf-8")
    assert "stage=progress-heartbeat" in text
    assert "progress_stage=separation" in text


def test_structured_tqdm_capture_logs_numeric_debug_progress(tmp_path):
    from io import StringIO
    from tqdm import tqdm

    path = tmp_path / "tqdm.log"
    with JobLogCapture(path, "tqdm123", level="DEBUG") as capture:
        with StructuredTqdmCapture(
            capture.debug_event, min_percent_delta=0.0, min_interval=0.0
        ):
            for _ in tqdm(range(4), file=StringIO(), mininterval=0, miniters=1):
                pass
    text = path.read_text(encoding="utf-8")
    assert "stage=tqdm-progress" in text
    assert "current=4" in text
    assert "total=4" in text
    assert "percent=100.0" in text


def test_structured_events_are_always_mirrored_to_space_console(tmp_path, capsys):
    path = tmp_path / "console.log"
    with JobLogCapture(path, "console123", level="INFO") as capture:
        capture.event("console-check", value=1)
    stderr = capsys.readouterr().err
    assert "job=console123" in stderr
    assert "stage=console-check" in stderr