writing tests much fun

Author: sl0thentr0py

sentry_sdk/monitor.py

@@ -20,9 +20,9 @@ class Monitor(object):
  name = "sentry.monitor"
 
  def __init__(self, transport, interval=10):
- # type: (sentry_sdk.transport.Transport, int) -> None
+ # type: (sentry_sdk.transport.Transport, float) -> None
  self.transport = transport # type: sentry_sdk.transport.Transport
- self.interval = interval # type: int
+ self.interval = interval # type: float
 
  self._healthy = True
  self._downsample_factor = 1 # type: int

tests/test_monitor.py

@@ -0,0 +1,54 @@
+import pytest
+from time import sleep
+
+from sentry_sdk import Hub
+from sentry_sdk.transport import Transport
+
+
+class HealthyTestTransport(Transport):
+ def _send_event(self, event):
+ pass
+
+ def _send_envelope(self, envelope):
+ pass
+
+ def is_worker_full(self):
+ return False
+
+class UnhealthyTestTransport(HealthyTestTransport):
+ def is_worker_full(self):
+ return True
+
+
+def test_no_monitor_if_disabled(sentry_init):
+ sentry_init(transport=HealthyTestTransport())
+ assert Hub.current.client.monitor is None
+
+
+def test_monitor_if_enabled(sentry_init):
+ sentry_init(transport=HealthyTestTransport(), _experiments={"enable_backpressure_handling": True})
+
+ monitor = Hub.current.client.monitor
+ assert monitor is not None
+ assert monitor._thread is None
+
+ assert monitor.is_healthy() is True
+ assert monitor.downsample_factor() == 1
+ assert monitor._thread is not None
+ assert monitor._thread.name == "sentry.monitor"
+
+
+def test_monitor_unhealthy(sentry_init, monkeypatch):
+ sentry_init(transport=UnhealthyTestTransport(), _experiments={"enable_backpressure_handling": True})
+
+ client = Hub.current.client
+ transport = client.transport
+ monitor = client.monitor
+ monitor.interval = 0.1
+
+ assert monitor.is_healthy() is True
+ sleep(0.1) # wait for monitor to run
+ assert monitor.is_healthy() is False
+ assert monitor.downsample_factor() == 2
+ sleep(0.1) # wait for monitor to run
+ assert monitor.downsample_factor() == 4