svcforge/tests/unit/test_obs.py

"""obs.py: the three things that are wrong by default.

Not tested here: that structlog logs, that prometheus counts, that OTEL traces. Those are
the libraries' tests. What is tested is every place where the default is a bug — the
histogram buckets, the context that does not cross a queue, and the contextvars that leak
between tasks.
"""

from __future__ import annotations

import io
import json
import logging
import re
from collections.abc import Iterator
from pathlib import Path
from typing import Any
from uuid import uuid4

import pytest
import structlog
import yaml
from opentelemetry import trace
from opentelemetry.sdk.trace import TracerProvider
from prometheus_client import REGISTRY

from svcforge_core import obs
from svcforge_core.settings import Settings

# W3C traceparent: version-traceid-spanid-flags.
#
# The flags byte is matched loosely and the sampled bit checked separately, on purpose.
# Module 7's acceptance line says `00-<32 hex>-<16 hex>-01`, and current SDKs emit `-03`:
# bit 0x01 is `sampled`, and bit 0x02 is `random-trace-id` from trace-context level 2. An
# assertion pinned to `-01` fails on a spec revision that changed nothing we care about.
# What we care about is the trace being sampled, which is bit 0x01 and nothing else.
TRACEPARENT_RE = re.compile(
    r"^00-(?P<trace_id>[0-9a-f]{32})-(?P<span_id>[0-9a-f]{16})-(?P<flags>[0-9a-f]{2})$"
)
SAMPLED_BIT = 0x01

# A provision takes minutes. prometheus_client's default buckets end at 10 seconds.
DEFAULT_TOP_BUCKET = 10.0


def _settings(**over: object) -> Settings:
    return Settings(pg_dsn="postgresql://u:p@localhost:5432/db", **over)  # type: ignore[arg-type]


@pytest.fixture
def logs() -> Iterator[io.StringIO]:
    """setup() against a captured stream, restoring the process-wide config afterwards.

    setup() is deliberately global and deliberately once-only, which makes it deliberately
    awkward to test. Reaching into the module flag is the honest price of that; the
    alternative is a seam that exists only for tests.
    """
    stream = io.StringIO()
    saved_handlers = logging.getLogger().handlers[:]
    saved_level = logging.getLogger().level
    saved_config = structlog.get_config()

    obs._configured = False
    obs.setup("test-service", _settings())

    # Re-point the handler setup() installed at our buffer; everything else it configured —
    # processors, formatter, the JSON renderer last — is exactly what production gets.
    handler = logging.getLogger().handlers[0]
    assert isinstance(handler, logging.StreamHandler)
    handler.setStream(stream)

    try:
        yield stream
    finally:
        structlog.contextvars.clear_contextvars()
        structlog.configure(**saved_config)
        logging.getLogger().handlers = saved_handlers
        logging.getLogger().setLevel(saved_level)
        obs._configured = False


def _lines(stream: io.StringIO) -> list[dict[str, Any]]:
    return [json.loads(line) for line in stream.getvalue().splitlines() if line.strip()]


# --- Buckets ------------------------------------------------------------------------------


def test_provision_histogram_has_a_bucket_for_a_thirty_minute_provision() -> None:
    """The acceptance check, as a unit test: `le="1800"` exists.

    With the library defaults the top finite bucket is 10s, every provision lands in +Inf,
    and histogram_quantile interpolates across a bucket spanning 10s to infinity. The p95
    it returns is meaningless, and it is meaningless silently, which is why this is
    asserted rather than eyeballed on a dashboard.
    """
    bounds = obs.PROVISION_TIME._upper_bounds
    assert 1800.0 in bounds
    assert bounds[-1] == float("inf")
    assert max(b for b in bounds if b != float("inf")) > DEFAULT_TOP_BUCKET


def test_provision_histogram_exports_the_1800_bucket() -> None:
    """Same claim, checked through the exposition format the scrape actually reads."""
    obs.PROVISION_TIME.observe(42.0)
    bucket = REGISTRY.get_sample_value("svcforge_provision_duration_seconds_bucket", {"le": "1800.0"})
    assert bucket is not None


def test_every_metric_the_alerts_query_actually_exists() -> None:
    """Parse the alerts out of values.yaml and check each metric is really registered.

    Reading the chart is the whole point. An earlier version of this test asserted against
    a hardcoded list, which meant it could only catch a rename on the *code* side — rename
    a metric in values.yaml and it stayed green while the alert silently never fired again.
    A test that duplicates the thing it is meant to cross-check is not a cross-check.
    """
    values = yaml.safe_load(
        (Path(__file__).resolve().parents[2] / "deploy" / "chart" / "values.yaml").read_text()
    )
    rules = (values.get("prometheusRule") or {}).get("rules") or []
    assert rules, "no alert rules found under prometheusRule.rules — this test would silently pass"

    # Prometheus appends _bucket/_sum/_count to a histogram's series; the collector is
    # registered under the base name.
    suffixes = ("_bucket", "_sum", "_count")
    registered = set(REGISTRY._names_to_collectors)

    checked = 0
    for rule in rules:
        for token in re.findall(r"\bsvcforge_[a-z0-9_]+\b", str(rule.get("expr", ""))):
            base = next((token[: -len(s)] for s in suffixes if token.endswith(s)), token)
            assert base in registered, (
                f"alert {rule.get('alert')!r} queries {token!r}, but no such metric is registered"
            )
            checked += 1
    assert checked >= len(rules), "expected at least one metric reference per alert"


# --- Trace context across the queue -------------------------------------------------------


def test_traceparent_round_trips_through_a_string() -> None:
    """inject -> a W3C string -> extract -> the same trace. This is the queue crossing."""
    provider = TracerProvider()
    tracer = provider.get_tracer("test")

    with tracer.start_as_current_span("api.post") as span:
        traceparent = obs.inject_traceparent()
        api_trace_id = span.get_span_context().trace_id

    assert traceparent is not None
    match = TRACEPARENT_RE.match(traceparent)
    assert match is not None, traceparent
    assert match["trace_id"] == format(api_trace_id, "032x")
    assert int(match["flags"], 16) & SAMPLED_BIT, "unsampled: the worker's span would be dropped"

    # The worker's side: a fresh context, minutes later, in another process.
    ctx = obs.context_from_traceparent(traceparent)
    with tracer.start_as_current_span("worker.claim", context=ctx) as worker_span:
        assert worker_span.get_span_context().trace_id == api_trace_id


def test_inject_returns_none_without_a_span() -> None:
    """A task the reconciler enqueued has no inbound request. Null column, not an error."""
    assert obs.inject_traceparent() is None


def test_context_from_traceparent_survives_none_and_garbage() -> None:
    """A malformed traceparent must start a new trace, never fail a provision.

    Extract does not raise on bad input — it returns a context with no span. Asserted here
    because the alternative would be a tenant's provision failing over a telemetry header.
    """
    for bad in (None, "", "not-a-traceparent", "00-tooshort-01"):
        ctx = obs.context_from_traceparent(bad)
        assert not trace.get_current_span(ctx).get_span_context().is_valid


# --- Contextvars --------------------------------------------------------------------------


def test_every_line_carries_instance_id_task_id_and_team(logs: io.StringIO) -> None:
    """The acceptance check: bind once at claim, and the keys ride on every line after."""
    instance_id = uuid4()
    obs.bind_task_context(instance_id, 42, "platform")

    obs.get_logger("t").info("provision.started")
    obs.get_logger("t").warning("helm.slow")

    for line in _lines(logs):
        assert line["instance_id"] == str(instance_id)
        assert line["task_id"] == 42
        assert line["team"] == "platform"


def test_foreign_stdlib_logs_are_json_and_carry_the_context(logs: io.StringIO) -> None:
    """psycopg and uvicorn log through stdlib `logging`, and their lines must parse too.

    Without the ProcessorFormatter bridge these arrive as bare text on the same stdout, and
    every one of them is a parse failure in the collector.
    """
    obs.bind_task_context(uuid4(), 7, "payments")
    logging.getLogger("some.library").warning("connection reset")

    line = _lines(logs)[-1]
    assert line["event"] == "connection reset"
    assert line["task_id"] == 7
    assert line["team"] == "payments"


def test_bind_task_context_clears_the_previous_task(logs: io.StringIO) -> None:
    """The bug this prevents: task 2's log line naming task 1's tenant.

    A worker coroutine reuses its context across claim-loop iterations. Bind without
    clearing and the stale instance_id survives into the next task, which means the log for
    the incident you are debugging points at the wrong customer.
    """
    obs.bind_task_context(uuid4(), 1, "team-a")
    second = uuid4()
    obs.bind_task_context(second, 2, "team-b")

    obs.get_logger("t").info("claimed")

    line = _lines(logs)[-1]
    assert line["instance_id"] == str(second)
    assert line["task_id"] == 2
    assert line["team"] == "team-b"


def test_json_renderer_is_last_and_output_is_one_object_per_line(logs: io.StringIO) -> None:
    """JSONRenderer last in the chain, and nothing after it.

    A processor appended after the renderer receives a `str` where it expects a dict and
    raises. The symptom is not a crash — structlog's default is to fail the log call, so
    the line simply never appears.
    """
    obs.get_logger("t").info("hello", extra_key="value")

    lines = _lines(logs)
    assert len(lines) == 1
    assert lines[0]["event"] == "hello"
    assert lines[0]["extra_key"] == "value"
    assert lines[0]["level"] == "info"
    assert lines[0]["service"] == "test-service"
    assert "timestamp" in lines[0]


def test_setup_is_idempotent(logs: io.StringIO) -> None:
    """Called twice, one handler, one line. Two handlers would be two of every line.

    Each service calls setup() from its entrypoint, and an entrypoint that imports another
    entrypoint (the CLI shelling into the reconciler) calls it twice.
    """
    obs.setup("test-service", _settings())
    obs.setup("test-service", _settings())

    # Count ours, not pytest's — its capture handler is on the root logger too.
    ours = [
        h
        for h in logging.getLogger().handlers
        if isinstance(h.formatter, structlog.stdlib.ProcessorFormatter)
    ]
    assert len(ours) == 1

    obs.get_logger("t").info("once")
    assert len(_lines(logs)) == 1


def test_metrics_are_single_process_in_memory() -> None:
    """One process per pod, scale with replicas. Multiprocess mode is not in use.

    `ValueClass` is prometheus_client's fork in the road, chosen at import from
    PROMETHEUS_MULTIPROC_DIR: MutexValue keeps counters in memory, MultiProcessValue mmaps
    them into a shared directory. This repo takes the other fix for `uvicorn --workers 4`
    corrupting counters — one process per pod — so MutexValue is the correct answer, and a
    stray PROMETHEUS_MULTIPROC_DIR in a Deployment's env would silently change it to the
    other one along with the meaning of every gauge.
    """
    from prometheus_client import values

    assert values.ValueClass is values.MutexValue

    # And the registry the services expose is the default in-memory one, not a
    # MultiProcessCollector reading files off disk.
    obs.QUEUE_DEPTH.set(3)
    assert REGISTRY.get_sample_value("svcforge_queue_depth") == 3.0