office_translator/tests/test_metrics.py

"""
Tests for Track C3 — Prometheus quality metrics.

Covers the new counters and helpers added in middleware/metrics.py:
  - quality_l0_checks_total{result, file_type}
  - quality_l1_judge_total{verdict, model}
  - quality_l1_judge_duration_seconds{model}
  - quality_l1_judge_cost_usd{model}
  - translation_retry_total{reason, tier}
  - format_elements_lost_total{format, element_type}
"""
import sys
import importlib.util
from pathlib import Path

import pytest
from prometheus_client import (
    REGISTRY,
    CollectorRegistry,
    Counter,
    Histogram,
)


# ---------------------------------------------------------------------------
# The middleware package __init__ imports `magic` (libmagic), which is not
# always available in CI. We load the metrics module directly to test it in
# isolation — the production code uses the same import path.
#
# prometheus_client uses a global default REGISTRY. To avoid duplicate
# registration errors when the test module is collected multiple times,
# we re-create the metrics on a fresh per-fixture registry.
# ---------------------------------------------------------------------------

_REPO_ROOT = Path(__file__).resolve().parent.parent


@pytest.fixture(scope="module")
def metrics():
    """Load the metrics module ONCE per test module.

    Because the global REGISTRY is shared across the test session,
    we register the counters/histograms on a fresh CollectorRegistry
    the first time the module loads, then reuse the same module
    instance across tests.

    Counters retain their values across tests within this module, so
    the tests check for monotonic increase (>=) rather than equality.
    """
    return _load_metrics_module_with_fresh_registry()


def _load_metrics_module_with_registry(registry: CollectorRegistry):
    """Load middleware/metrics.py with ALL its counters/histograms
    registered on the supplied fresh registry.

    metrics.py does ``from prometheus_client import Counter, Histogram`` at
    module top — injecting patched classes into the module dict BEFORE exec
    does not survive that import. The only reliable interception point is
    the ``prometheus_client`` module itself: we swap its Counter/Histogram
    attributes for subclasses that default to ``registry``, exec the
    module source, and restore the originals.
    """
    import types
    import prometheus_client

    orig_counter = prometheus_client.Counter
    orig_histogram = prometheus_client.Histogram

    class _RegistryCounter(orig_counter):
        def __new__(cls, *args, **kwargs):
            kwargs.setdefault("registry", registry)
            return super().__new__(cls)

        def __init__(self, *args, **kwargs):
            kwargs.setdefault("registry", registry)
            super().__init__(*args, **kwargs)

    class _RegistryHistogram(orig_histogram):
        def __new__(cls, *args, **kwargs):
            kwargs.setdefault("registry", registry)
            return super().__new__(cls)

        def __init__(self, *args, **kwargs):
            kwargs.setdefault("registry", registry)
            super().__init__(*args, **kwargs)

    source = (_REPO_ROOT / "middleware" / "metrics.py").read_text(encoding="utf-8")
    mod = types.ModuleType("metrics_under_test")
    mod.__file__ = str(_REPO_ROOT / "middleware" / "metrics.py")
    try:
        prometheus_client.Counter = _RegistryCounter
        prometheus_client.Histogram = _RegistryHistogram
        exec(compile(source, mod.__file__, "exec"), mod.__dict__)
    finally:
        prometheus_client.Counter = orig_counter
        prometheus_client.Histogram = orig_histogram
    return mod


def _load_metrics_module_with_fresh_registry():
    """Load metrics module with its counters/histograms attached to a
    fresh CollectorRegistry (never the shared default one — the app may
    already have registered the same names earlier in the session)."""
    return _load_metrics_module_with_registry(CollectorRegistry())


def _counter_value(counter, **labels):
    """Read the current value of a labelled counter. 0 if unlabelled."""
    try:
        return counter.labels(**labels)._value.get()
    except (KeyError, AttributeError):
        return 0


# ============================================================================
# L0 counter
# ============================================================================

class TestL0Counter:
    def test_l0_pass_increments(self, metrics):
        before = _counter_value(metrics.quality_l0_checks_total, result="pass", file_type="docx")
        metrics.record_l0_result(passed=True, file_type="docx")
        after = _counter_value(metrics.quality_l0_checks_total, result="pass", file_type="docx")
        assert after >= before + 1

    def test_l0_fail_increments(self, metrics):
        before = _counter_value(metrics.quality_l0_checks_total, result="fail", file_type="pptx")
        metrics.record_l0_result(passed=False, file_type="pptx")
        after = _counter_value(metrics.quality_l0_checks_total, result="fail", file_type="pptx")
        assert after >= before + 1

    def test_l0_error_increments(self, metrics):
        before = _counter_value(metrics.quality_l0_checks_total, result="error", file_type="xlsx")
        metrics.record_l0_error(file_type="xlsx")
        after = _counter_value(metrics.quality_l0_checks_total, result="error", file_type="xlsx")
        assert after >= before + 1

    def test_l0_file_type_unknown_default(self, metrics):
        # Default file_type is "unknown" — make sure that doesn't blow up
        metrics.record_l0_result(passed=True)
        metrics.record_l0_error()


# ============================================================================
# L1 counter
# ============================================================================

class TestL1Counter:
    def test_l1_pass_increments(self, metrics):
        before = _counter_value(
            metrics.quality_l1_judge_total,
            verdict="pass", model="deepseek-chat",
        )
        metrics.record_l1_verdict(verdict="pass", model="deepseek-chat")
        after = _counter_value(
            metrics.quality_l1_judge_total,
            verdict="pass", model="deepseek-chat",
        )
        assert after >= before + 1

    def test_l1_fail_increments(self, metrics):
        before = _counter_value(
            metrics.quality_l1_judge_total,
            verdict="fail", model="gpt-4o-mini",
        )
        metrics.record_l1_verdict(verdict="fail", model="gpt-4o-mini")
        after = _counter_value(
            metrics.quality_l1_judge_total,
            verdict="fail", model="gpt-4o-mini",
        )
        assert after >= before + 1

    def test_l1_skip_increments(self, metrics):
        before = _counter_value(
            metrics.quality_l1_judge_total,
            verdict="skip", model="none",
        )
        metrics.record_l1_verdict(verdict="skip", model="none")
        after = _counter_value(
            metrics.quality_l1_judge_total,
            verdict="skip", model="none",
        )
        assert after >= before + 1

    def test_l1_with_duration_and_cost(self, metrics):
        # Should not raise
        metrics.record_l1_verdict(
            verdict="pass",
            model="deepseek-chat",
            duration_seconds=0.42,
            cost_usd=0.0003,
        )
        # And counter should have incremented
        after = _counter_value(
            metrics.quality_l1_judge_total,
            verdict="pass", model="deepseek-chat",
        )
        assert after >= 1


# ============================================================================
# Translation retry counter
# ============================================================================

class TestRetryCounter:
    def test_retry_l0_fail(self, metrics):
        before = _counter_value(
            metrics.translation_retry_total,
            reason="l0_fail", tier="free",
        )
        metrics.record_translation_retry(reason="l0_fail", tier="free")
        after = _counter_value(
            metrics.translation_retry_total,
            reason="l0_fail", tier="free",
        )
        assert after >= before + 1

    def test_retry_l1_fail_pro(self, metrics):
        before = _counter_value(
            metrics.translation_retry_total,
            reason="l1_fail", tier="pro",
        )
        metrics.record_translation_retry(reason="l1_fail", tier="pro")
        after = _counter_value(
            metrics.translation_retry_total,
            reason="l1_fail", tier="pro",
        )
        assert after >= before + 1

    def test_retry_format_loss(self, metrics):
        before = _counter_value(
            metrics.translation_retry_total,
            reason="format_loss", tier="enterprise",
        )
        metrics.record_translation_retry(reason="format_loss", tier="enterprise")
        after = _counter_value(
            metrics.translation_retry_total,
            reason="format_loss", tier="enterprise",
        )
        assert after >= before + 1

    def test_retry_default_tier(self, metrics):
        # Default tier is "free"
        before = _counter_value(
            metrics.translation_retry_total,
            reason="user_request", tier="free",
        )
        metrics.record_translation_retry(reason="user_request")
        after = _counter_value(
            metrics.translation_retry_total,
            reason="user_request", tier="free",
        )
        assert after >= before + 1


# ============================================================================
# Format loss counter
# ============================================================================

class TestFormatLossCounter:
    def test_docx_hyperlink(self, metrics):
        before = _counter_value(
            metrics.format_elements_lost_total,
            format="docx", element_type="hyperlink",
        )
        metrics.record_format_loss(format="docx", element_type="hyperlink")
        after = _counter_value(
            metrics.format_elements_lost_total,
            format="docx", element_type="hyperlink",
        )
        assert after >= before + 1

    def test_pptx_diagram(self, metrics):
        before = _counter_value(
            metrics.format_elements_lost_total,
            format="pptx", element_type="diagram",
        )
        metrics.record_format_loss(format="pptx", element_type="diagram")
        after = _counter_value(
            metrics.format_elements_lost_total,
            format="pptx", element_type="diagram",
        )
        assert after >= before + 1

    def test_pdf_image(self, metrics):
        before = _counter_value(
            metrics.format_elements_lost_total,
            format="pdf", element_type="image",
        )
        metrics.record_format_loss(format="pdf", element_type="image")
        after = _counter_value(
            metrics.format_elements_lost_total,
            format="pdf", element_type="image",
        )
        assert after >= before + 1


# ============================================================================
# Helper robustness
# ============================================================================

class TestMetricsResilience:
    """The metric helpers must never raise — a failing metrics call
    must not break a translation job."""

    def test_all_helpers_accept_none(self, metrics):
        # No kwargs at all → defaults kick in
        metrics.record_l0_result(passed=True)
        metrics.record_l1_verdict(verdict="pass")
        metrics.record_translation_retry(reason="user_request")
        metrics.record_format_loss(format="docx", element_type="hyperlink")

    def test_duration_zero_is_ok(self, metrics):
        metrics.record_l1_verdict(
            verdict="pass", model="deepseek-chat",
            duration_seconds=0.0, cost_usd=0.0,
        )


# ============================================================================
# Pipeline integration smoke tests
# ============================================================================

class TestPipelineIntegration:
    """Make sure the L0 + L1 pipeline still records metrics correctly."""

    def test_l0_check_records_metric(self, metrics):
        from services.quality import run_l0_check

        # Use clear, easy-to-detect text that should pass L0
        source = ["Hello world"] * 3
        translated = ["Bonjour le monde"] * 3

        before = _counter_value(
            metrics.quality_l0_checks_total, result="pass", file_type="docx",
        )
        result = run_l0_check(
            source, translated, "fr",
            job_id="test_job", file_extension="docx",
        )
        after = _counter_value(
            metrics.quality_l0_checks_total, result="pass", file_type="docx",
        )

        # The pipeline should have called record_l0_result() at least once
        assert after >= before

    def test_l0_check_handles_invalid_input(self, metrics):
        from services.quality import run_l0_check

        # Empty chunks: should not crash, just return a benign result
        result = run_l0_check([], [], "fr", file_extension="docx")
        assert result is not None