1
0

test_ratelimiting.py 1.2 KB

123456789101112131415161718192021222324252627282930313233343536373839
  1. from synapse.api.ratelimiting import Ratelimiter
  2. from tests import unittest
  3. class TestRatelimiter(unittest.TestCase):
  4. def test_allowed(self):
  5. limiter = Ratelimiter()
  6. allowed, time_allowed = limiter.can_do_action(
  7. key="test_id", time_now_s=0, rate_hz=0.1, burst_count=1
  8. )
  9. self.assertTrue(allowed)
  10. self.assertEquals(10.0, time_allowed)
  11. allowed, time_allowed = limiter.can_do_action(
  12. key="test_id", time_now_s=5, rate_hz=0.1, burst_count=1
  13. )
  14. self.assertFalse(allowed)
  15. self.assertEquals(10.0, time_allowed)
  16. allowed, time_allowed = limiter.can_do_action(
  17. key="test_id", time_now_s=10, rate_hz=0.1, burst_count=1
  18. )
  19. self.assertTrue(allowed)
  20. self.assertEquals(20.0, time_allowed)
  21. def test_pruning(self):
  22. limiter = Ratelimiter()
  23. allowed, time_allowed = limiter.can_do_action(
  24. key="test_id_1", time_now_s=0, rate_hz=0.1, burst_count=1
  25. )
  26. self.assertIn("test_id_1", limiter.message_counts)
  27. allowed, time_allowed = limiter.can_do_action(
  28. key="test_id_2", time_now_s=10, rate_hz=0.1, burst_count=1
  29. )
  30. self.assertNotIn("test_id_1", limiter.message_counts)