1from ddtrace.internal.runtime.constants import GC_COUNT_GEN0
2from ddtrace.internal.runtime.constants import GC_RUNTIME_METRICS
3from ddtrace.internal.runtime.constants import PSUTIL_RUNTIME_METRICS
4from ddtrace.internal.runtime.metric_collectors import GCRuntimeMetricCollector
5from ddtrace.internal.runtime.metric_collectors import PSUtilRuntimeMetricCollector
6from ddtrace.internal.runtime.metric_collectors import RuntimeMetricCollector
7from tests.utils import BaseTestCase
8
9
10class TestRuntimeMetricCollector(BaseTestCase):
11    def test_failed_module_load_collect(self):
12        """Attempts to collect from a collector when it has failed to load its
13        module should return no metrics gracefully.
14        """
15
16        class A(RuntimeMetricCollector):
17            required_modules = ["moduleshouldnotexist"]
18
19            def collect_fn(self, keys):
20                return {"k": "v"}
21
22        self.assertIsNotNone(A().collect(), "collect should return valid metrics")
23
24
25class TestPSUtilRuntimeMetricCollector(BaseTestCase):
26    def test_metrics(self):
27        collector = PSUtilRuntimeMetricCollector()
28        for (key, value) in collector.collect(PSUTIL_RUNTIME_METRICS):
29            self.assertIsNotNone(value)
30
31
32class TestGCRuntimeMetricCollector(BaseTestCase):
33    def test_metrics(self):
34        collector = GCRuntimeMetricCollector()
35        for (key, value) in collector.collect(GC_RUNTIME_METRICS):
36            self.assertIsNotNone(value)
37
38    def test_gen1_changes(self):
39        # disable gc
40        import gc
41
42        gc.disable()
43
44        # start collector and get current gc counts
45        collector = GCRuntimeMetricCollector()
46        gc.collect()
47        start = gc.get_count()
48
49        # create reference
50        a = []
51        collected = collector.collect([GC_COUNT_GEN0])
52        self.assertGreater(collected[0][1], start[0])
53
54        # delete reference and collect
55        del a
56        gc.collect()
57        collected_after = collector.collect([GC_COUNT_GEN0])
58        assert len(collected_after) == 1
59        assert collected_after[0][0] == "runtime.python.gc.count.gen0"
60        assert isinstance(collected_after[0][1], int)
61