software/tests/test_rate_limiter.py

1"""Tests for shared rate limiter."""23from __future__ import annotations45import threading6import time78import pytest910from src.orchestration.rate_limiter import DummyRateLimiter, SharedRateLimiter111213class TestSharedRateLimiter:14    """Tests for SharedRateLimiter."""1516    def test_default_values(self) -> None:17        """Test default configuration values."""18        limiter = SharedRateLimiter()19        assert limiter.max_concurrent == 520        assert limiter.delay == 1.221        limiter.shutdown()2223    def test_custom_values(self) -> None:24        """Test custom configuration values."""25        limiter = SharedRateLimiter(max_concurrent=3, delay=0.5)26        assert limiter.max_concurrent == 327        assert limiter.delay == 0.528        limiter.shutdown()2930    def test_acquire_release(self) -> None:31        """Test basic acquire and release."""32        limiter = SharedRateLimiter(max_concurrent=2, delay=0.01)33        try:34            limiter.acquire()35            limiter.release()36        finally:37            limiter.shutdown()3839    def test_context_manager_call(self) -> None:40        """Test using limiter as context manager via __call__."""41        limiter = SharedRateLimiter(max_concurrent=2, delay=0.01)42        try:43            with limiter():44                pass45        finally:46            limiter.shutdown()4748    def test_context_manager_direct(self) -> None:49        """Test using limiter as context manager directly."""50        limiter = SharedRateLimiter(max_concurrent=2, delay=0.01)51        try:52            with limiter:53                pass54        finally:55            limiter.shutdown()5657    def test_delay_enforcement(self) -> None:58        """Test that delay is enforced between calls."""59        limiter = SharedRateLimiter(max_concurrent=2, delay=0.1)60        try:61            start = time.time()6263            with limiter():64                pass6566            with limiter():67                pass6869            elapsed = time.time() - start70            # Should take at least delay seconds for second call71            assert elapsed >= 0.172        finally:73            limiter.shutdown()7475    def test_max_concurrent_limit(self) -> None:76        """Test that max concurrent limit is enforced."""77        limiter = SharedRateLimiter(max_concurrent=2, delay=0.01)78        lock = threading.Lock()79        concurrent_count = 080        max_observed = 08182        def worker() -> None:83            nonlocal concurrent_count, max_observed84            with limiter:85                with lock:86                    concurrent_count += 187                    if concurrent_count > max_observed:88                        max_observed = concurrent_count89                time.sleep(0.05)90                with lock:91                    concurrent_count -= 19293        try:94            threads = [threading.Thread(target=worker) for _ in range(5)]95            for t in threads:96                t.start()97            for t in threads:98                t.join()99100            # Max concurrent should not exceed limit101            assert max_observed <= 2102        finally:103            limiter.shutdown()104105    def test_shutdown_idempotent(self) -> None:106        """Test that shutdown can be called multiple times."""107        limiter = SharedRateLimiter(max_concurrent=2, delay=0.01)108        limiter.shutdown()109        limiter.shutdown()  # Should not raise110111class TestDummyRateLimiter:112    """Tests for DummyRateLimiter."""113114    def test_default_values(self) -> None:115        """Test default configuration values."""116        limiter = DummyRateLimiter()117        assert limiter.max_concurrent == 5118        assert limiter.delay == 1.2119120    def test_acquire_release(self) -> None:121        """Test basic acquire and release."""122        limiter = DummyRateLimiter(delay=0.01)123        limiter.acquire()124        limiter.release()125126    def test_context_manager_call(self) -> None:127        """Test using limiter as context manager via __call__."""128        limiter = DummyRateLimiter(delay=0.01)129        with limiter():130            pass131132    def test_context_manager_direct(self) -> None:133        """Test using limiter as context manager directly."""134        limiter = DummyRateLimiter(delay=0.01)135        with limiter:136            pass137138    def test_delay_enforcement(self) -> None:139        """Test that delay is enforced."""140        limiter = DummyRateLimiter(delay=0.1)141142        start = time.time()143144        with limiter():145            pass146147        with limiter():148            pass149150        elapsed = time.time() - start151        assert elapsed >= 0.1152153    def test_no_concurrency_limit(self) -> None:154        """Test that there's no actual concurrency limit."""155        limiter = DummyRateLimiter(delay=0.001)156        concurrent_count = 0157        max_concurrent = 0158        lock = threading.Lock()159160        def worker() -> None:161            nonlocal concurrent_count, max_concurrent162            with limiter:163                with lock:164                    concurrent_count += 1165                    max_concurrent = max(max_concurrent, concurrent_count)166                time.sleep(0.01)167                with lock:168                    concurrent_count -= 1169170        threads = [threading.Thread(target=worker) for _ in range(10)]171        for t in threads:172            t.start()173        for t in threads:174            t.join()175176        # With dummy limiter, all threads can run concurrently177        # (they still have delay but no semaphore)178        assert max_concurrent >= 1179180    def test_shutdown_noop(self) -> None:181        """Test that shutdown is a no-op."""182        limiter = DummyRateLimiter()183        limiter.shutdown()184        limiter.shutdown()  # Should not raise185186187class TestLimiterInteroperability:188    """Tests for using different limiters interchangeably."""189190    def test_same_interface(self) -> None:191        """Test that both limiters have the same interface."""192        for limiter_class in [SharedRateLimiter, DummyRateLimiter]:193            limiter = limiter_class(max_concurrent=2, delay=0.01)194195            # Should have same methods196            assert hasattr(limiter, "acquire")197            assert hasattr(limiter, "release")198            assert hasattr(limiter, "shutdown")199            assert hasattr(limiter, "__enter__")200            assert hasattr(limiter, "__exit__")201            assert hasattr(limiter, "__call__")202203            # Should have same attributes204            assert hasattr(limiter, "max_concurrent")205            assert hasattr(limiter, "delay")206207            limiter.shutdown()208209    def test_function_works_with_either(self) -> None:210        """Test that a function can use either limiter type."""211212        def do_work(limiter: SharedRateLimiter | DummyRateLimiter) -> str:213            with limiter:214                return "done"215216        shared = SharedRateLimiter(max_concurrent=2, delay=0.01)217        dummy = DummyRateLimiter(delay=0.01)218219        try:220            assert do_work(shared) == "done"221            assert do_work(dummy) == "done"222        finally:223            shared.shutdown()224            dummy.shutdown()225