56 lines
1.8 KiB
Python
56 lines
1.8 KiB
Python
import re
|
|
from time import sleep
|
|
|
|
import pytest
|
|
import requests
|
|
from fastapi import FastAPI
|
|
|
|
from pyinfra.webserver.prometheus import (
|
|
add_prometheus_endpoint,
|
|
make_prometheus_processing_time_decorator_from_settings,
|
|
)
|
|
from pyinfra.webserver.utils import create_webserver_thread_from_settings
|
|
|
|
|
|
@pytest.fixture(scope="class")
|
|
def app_with_prometheus_endpoint(settings):
|
|
app = FastAPI()
|
|
app = add_prometheus_endpoint(app)
|
|
thread = create_webserver_thread_from_settings(app, settings)
|
|
thread.daemon = True
|
|
thread.start()
|
|
sleep(1)
|
|
yield
|
|
thread.join(timeout=1)
|
|
|
|
|
|
@pytest.fixture
|
|
def monitored_function(settings):
|
|
@make_prometheus_processing_time_decorator_from_settings(settings)
|
|
def process(*args, **kwargs):
|
|
sleep(0.5)
|
|
|
|
return process
|
|
|
|
|
|
class TestPrometheusMonitor:
|
|
def test_prometheus_endpoint_is_available(self, app_with_prometheus_endpoint, settings):
|
|
resp = requests.get(f"http://{settings.webserver.host}:{settings.webserver.port}/prometheus")
|
|
assert resp.status_code == 200
|
|
|
|
def test_processing_with_a_monitored_fn_increases_parameter_counter(
|
|
self, app_with_prometheus_endpoint, monitored_function, settings
|
|
):
|
|
pattern = re.compile(rf".*{settings.metrics.prometheus.prefix}_processing_time_count (\d\.\d).*")
|
|
|
|
resp = requests.get(f"http://{settings.webserver.host}:{settings.webserver.port}/prometheus")
|
|
assert pattern.search(resp.text).group(1) == "0.0"
|
|
|
|
monitored_function()
|
|
resp = requests.get(f"http://{settings.webserver.host}:{settings.webserver.port}/prometheus")
|
|
assert pattern.search(resp.text).group(1) == "1.0"
|
|
|
|
monitored_function()
|
|
resp = requests.get(f"http://{settings.webserver.host}:{settings.webserver.port}/prometheus")
|
|
assert pattern.search(resp.text).group(1) == "2.0"
|