Coverage for databases/tests/test_metrics.py: 100%

1import pytest 

2 

3from prisma import Prisma 

4from prisma._compat import model_json 

5 

6 

7@pytest.mark.asyncio 

8async def test_prometheus(client: Prisma) -> None: 

9 """Metrics can be returned in Prometheus format""" 

10 await client.user.create(data={'name': 'Robert'}) 

11 

12 metrics = await client.get_metrics(format='prometheus') 

13 

14 assert 'prisma_client_queries_total' in metrics 

15 assert 'prisma_datasource_queries_total' in metrics 

16 assert 'prisma_client_queries_active' in metrics 

17 assert 'prisma_client_queries_duration_histogram_ms_bucket' in metrics 

18 

19 

20@pytest.mark.asyncio 

21async def test_json_string(client: Prisma) -> None: 

22 """Metrics can be serlialized to JSON""" 

23 await client.user.create(data={'name': 'Robert'}) 

24 

25 metrics = await client.get_metrics() 

26 assert isinstance(model_json(metrics), str) 

27 

28 

29@pytest.mark.asyncio 

30async def test_json(client: Prisma) -> None: 

31 """Metrics returned in the JSON format""" 

32 await client.user.create(data={'name': 'Robert'}) 

33 

34 metrics = await client.get_metrics(format='json') 

35 

36 assert len(metrics.counters) > 0 

37 assert metrics.counters[0].value > 0 

38 

39 assert len(metrics.gauges) > 0 

40 gauge = next(filter(lambda g: g.key == 'prisma_pool_connections_open', metrics.gauges)) 

41 assert gauge.value > 0 

42 

43 assert len(metrics.histograms) > 0 

44 assert metrics.histograms[0].value.sum > 0 

45 assert metrics.histograms[0].value.count > 0 

46 

47 assert len(metrics.histograms[0].value.buckets) > 0 

48 

49 for bucket in metrics.histograms[0].value.buckets: 

50 assert bucket.max_value >= 0 

51 assert bucket.total_count >= 0 

52 

53 

54@pytest.mark.asyncio 

55async def test_global_labels(client: Prisma) -> None: 

56 metrics = await client.get_metrics(global_labels={'foo': 'bar'}) 

57 assert metrics.counters[0].labels == {'foo': 'bar'}