-
Notifications
You must be signed in to change notification settings - Fork 2
/
conftest.py
140 lines (100 loc) · 4.11 KB
/
conftest.py
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
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
from __future__ import annotations
import os
import platform
import tempfile
from typing import Any, Callable, Generator
import pytest
from opentelemetry.metrics import set_meter_provider
from opentelemetry.sdk.metrics import MeterProvider
from opentelemetry.sdk.metrics.export import InMemoryMetricReader
from opentelemetry.sdk.resources import Resource
from opentelemetry.sdk.trace import ReadableSpan, TracerProvider
from opentelemetry.sdk.trace.export import SimpleSpanProcessor
from opentelemetry.sdk.trace.export.in_memory_span_exporter import InMemorySpanExporter
from opentelemetry.trace import set_tracer_provider
from appsignal import probes
from appsignal.agent import agent
from appsignal.check_in.heartbeat import (
_kill_continuous_heartbeats,
_reset_heartbeat_continuous_interval_seconds,
)
from appsignal.check_in.scheduler import _reset_scheduler
from appsignal.client import _reset_client
from appsignal.heartbeat import _heartbeat_class_warning, _heartbeat_helper_warning
from appsignal.internal_logger import _reset_logger
from appsignal.opentelemetry import METRICS_PREFERRED_TEMPORALITY
@pytest.fixture(scope="function", autouse=True)
def disable_start_opentelemetry(mocker: Any) -> Any:
mocker.patch("appsignal.opentelemetry._start_tracer")
mocker.patch("appsignal.opentelemetry._start_metrics")
@pytest.fixture(scope="session", autouse=True)
def start_in_memory_metric_reader() -> Generator[InMemoryMetricReader, None, None]:
metric_reader = InMemoryMetricReader(
preferred_temporality=METRICS_PREFERRED_TEMPORALITY
)
provider = MeterProvider(resource=Resource({}), metric_readers=[metric_reader])
set_meter_provider(provider)
yield metric_reader
@pytest.fixture(scope="session", autouse=True)
def start_in_memory_span_exporter() -> Generator[InMemorySpanExporter, None, None]:
span_exporter = InMemorySpanExporter()
exporter_processor = SimpleSpanProcessor(span_exporter)
provider = TracerProvider()
provider.add_span_processor(exporter_processor)
set_tracer_provider(provider)
yield span_exporter
@pytest.fixture(scope="function")
def metrics(
start_in_memory_metric_reader: InMemoryMetricReader,
) -> Generator[Callable[[], Any], None, None]:
# Getting the metrics data implicitly wipes its state
start_in_memory_metric_reader.get_metrics_data()
yield start_in_memory_metric_reader.get_metrics_data
@pytest.fixture(scope="function")
def spans(
start_in_memory_span_exporter: InMemorySpanExporter,
) -> Generator[Callable[[], tuple[ReadableSpan, ...]], None, None]:
start_in_memory_span_exporter.clear()
def get_and_clear_spans() -> tuple[ReadableSpan, ...]:
spans = start_in_memory_span_exporter.get_finished_spans()
start_in_memory_span_exporter.clear()
return spans
yield get_and_clear_spans
@pytest.fixture(scope="function", autouse=True)
def reset_environment_between_tests() -> Any:
old_environ = dict(os.environ)
yield
os.environ.clear()
os.environ.update(old_environ)
@pytest.fixture(scope="function", autouse=True)
def reset_internal_logger_after_tests() -> Any:
yield
_reset_logger()
@pytest.fixture(scope="function", autouse=True)
def stop_and_clear_probes_after_tests() -> Any:
yield
probes.stop()
probes.clear()
@pytest.fixture(scope="function", autouse=True)
def reset_agent_active_state() -> Any:
agent.active = False
@pytest.fixture(scope="function", autouse=True)
def reset_global_client() -> Any:
_reset_client()
@pytest.fixture(scope="function", autouse=True)
def reset_checkins() -> Any:
yield
_reset_heartbeat_continuous_interval_seconds()
_kill_continuous_heartbeats()
_reset_scheduler()
@pytest.fixture(scope="function", autouse=True)
def stop_agent() -> Any:
tmp_path = "/tmp" if platform.system() == "Darwin" else tempfile.gettempdir()
working_dir = os.path.join(tmp_path, "appsignal")
if os.path.isdir(working_dir):
os.system(f"rm -rf {working_dir}")
@pytest.fixture(scope="function")
def reset_heartbeat_warnings() -> Any:
_heartbeat_class_warning.reset()
_heartbeat_helper_warning.reset()
yield