statlite/examples/python-fastapi-demo/test_statlite_metrics.py at main · PVRLabs/statlite · 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
75
76
"""Integration checks for the FastAPI StatLite Metrics demo."""

from __future__ import annotations

import importlib
import sys
import time
import unittest
from datetime import datetime
from pathlib import Path

from fastapi.testclient import TestClient


sys.path.insert(0, str(Path(__file__).parent))

import app as demo_app
from statlite_metrics import SCHEMA


class StatLiteMetricsTests(unittest.TestCase):
    def setUp(self) -> None:
        self.demo = importlib.reload(demo_app)
        self.client = TestClient(self.demo.app, raise_server_exceptions=False)

    def test_snapshot_is_application_and_process_profile_without_host_fields(self) -> None:
        snapshot = self.client.get("/statlite/metrics").json()

        self.assertEqual(snapshot["schema"], SCHEMA)
        self.assertEqual(snapshot["integration"], "fastapi")
        self.assertEqual(snapshot["status"], "UP")
        self.assertIsNotNone(datetime.fromisoformat(snapshot["started_at"].replace("Z", "+00:00")))
        metrics = snapshot["metrics"]
        self.assertIn("process_cpu_usage", metrics)
        self.assertNotIn("cpu_usage", metrics)
        self.assertIsInstance(metrics["runtime_heap_used_bytes"], int)
        self.assertGreaterEqual(metrics["runtime_heap_used_bytes"], 0)
        self.assertGreaterEqual(metrics["uptime_seconds"], 0)
        self.assertNotIn("request_duration_seconds_max", metrics)
        for key in (
            "host_cpu_usage",
            "host_memory_used_bytes",
            "host_memory_total_bytes",
            "host_disk_used_bytes",
            "host_disk_total_bytes",
        ):
            self.assertNotIn(key, metrics)

    def test_real_responses_are_counted_and_metrics_endpoint_is_excluded(self) -> None:
        self.assertEqual(self.client.get("/").status_code, 200)
        self.assertEqual(self.client.get("/missing").status_code, 404)
        self.assertEqual(self.client.get("/failure").status_code, 500)

        first = self.client.get("/statlite/metrics").json()["metrics"]
        second = self.client.get("/statlite/metrics?source=test").json()["metrics"]

        for snapshot in (first, second):
            self.assertEqual(snapshot["requests_total"], 3)
            self.assertEqual(snapshot["responses_404_total"], 1)
            self.assertEqual(snapshot["responses_4xx_total"], 1)
            self.assertEqual(snapshot["responses_5xx_total"], 1)
            self.assertGreater(snapshot["request_duration_seconds_total"], 0)

    def test_process_metrics_remain_valid_across_snapshots(self) -> None:
        first = self.client.get("/statlite/metrics").json()
        time.sleep(0.001)
        second = self.client.get("/statlite/metrics").json()

        self.assertEqual(second["started_at"], first["started_at"])
        self.assertGreaterEqual(
            second["metrics"]["uptime_seconds"], first["metrics"]["uptime_seconds"]
        )
        for snapshot in (first, second):
            memory = snapshot["metrics"]["runtime_heap_used_bytes"]
            self.assertIsInstance(memory, int)
            self.assertGreaterEqual(memory, 0)