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

1from prisma import Prisma 

2from prisma._compat import model_json 

3 

4 

5def test_prometheus(client: Prisma) -> None: 

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

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

8 

9 metrics = client.get_metrics(format='prometheus') 

10 

11 assert 'prisma_client_queries_total' in metrics 

12 assert 'prisma_datasource_queries_total' in metrics 

13 assert 'prisma_client_queries_active' in metrics 

14 assert 'prisma_client_queries_duration_histogram_ms_bucket' in metrics 

15 

16 

17def test_json_string(client: Prisma) -> None: 

18 """Metrics can be serlialized to JSON""" 

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

20 

21 metrics = client.get_metrics() 

22 assert isinstance(model_json(metrics), str) 

23 

24 

25def test_json(client: Prisma) -> None: 

26 """Metrics returned in the JSON format""" 

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

28 

29 metrics = client.get_metrics(format='json') 

30 

31 assert len(metrics.counters) > 0 

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

33 

34 assert len(metrics.gauges) > 0 

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

36 assert gauge.value > 0 

37 

38 assert len(metrics.histograms) > 0 

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

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

41 

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

43 

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

45 assert bucket.max_value >= 0 

46 assert bucket.total_count >= 0 

47 

48 

49def test_global_labels(client: Prisma) -> None: 

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

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