underwrite/tests/test_runtime_faults.py at master · sachncs/underwrite · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
"""Tests for Runtime — OTLP tracer, logging config, import errors, wiring."""

from __future__ import annotations

from typing import Any
from unittest.mock import MagicMock

from underwrite.__runtime__ import Runtime


class TestRuntimeOtlpTracer:
    def test_build_tracer_otlp_creates_tracer(self) -> None:
        config = config_with_otlp()
        rt = Runtime(config)
        assert rt.tracer is not None

    def test_build_tracer_otlp_disabled_returns_none(self) -> None:
        config = config_with_otlp()
        config.tracing.enabled = False
        rt = Runtime(config)
        assert rt.tracer is None

    def test_otlp_spans_are_exported(self) -> None:
        config = config_with_otlp()
        rt = Runtime(config)
        mock_exporter = MagicMock()
        assert rt.tracer is not None
        rt.tracer.exporter = mock_exporter

        span = rt.tracer.start_span("test-op", tags={"key": "val"})
        rt.tracer.end_span(span)

        mock_exporter.export.assert_called_once()
        exported_spans = mock_exporter.export.call_args[0][0]
        assert len(exported_spans) == 1
        assert exported_spans[0].operation == "test-op"
        assert exported_spans[0].tags["key"] == "val"

    def test_otlp_exporter_fallback_on_import_error(self) -> None:
        config = config_with_otlp()
        rt = Runtime(config)
        assert rt.tracer is not None
        span = rt.tracer.start_span("fallback-test")
        rt.tracer.end_span(span)
        assert span.operation == "fallback-test"


def config_with_otlp() -> Any:
    from underwrite.__config__ import Configuration

    config = Configuration.default()
    config.tracing.enabled = True
    config.tracing.exporter = "otlp"
    return config


class TestRuntimeRestartFailingServices:
    def test_restart_no_supervisor_returns_empty(self) -> None:
        from underwrite.__config__ import Configuration

        config = Configuration.default()
        config.recovery.auto_restart = False
        rt = Runtime(config)
        assert rt.restart_failing_services() == []

    def test_restart_returns_empty_when_no_failures(self) -> None:
        from underwrite.__config__ import Configuration

        config = Configuration.default()
        rt = Runtime(config)
        if rt.supervisor is not None:
            rt.supervisor.record_failure("svc-a")
            rt.supervisor.record_success("svc-a")
        assert rt.restart_failing_services() == []