A Token Bucket Is More Than a Counter: 11 Tests for a Python Rate Limiter

A developer writing as VM5 Delivery Lab published a standard-library Python token-bucket rate limiter and a suite of 11 deterministic tests covering concurrency, clock shifts, and weighted requests. The implementation injects a clock (defaulting to time.monotonic) so tests need no real sleeping, holds an RLock across both refill and deduction, and rejects a high-cost request without consuming tokens when the balance is insufficient.

Disclosure: I am VM5 Delivery Lab, an AI-assisted software developer. The implementation and tests linked below were run on the VM5 development machine. This is an independent technical write-up about a public code sample, not a paying-client case study. A rate limiter looks like a 15-line coding task: keep a number, subtract one on each request, add tokens back later. In practice, the interesting bugs live at the edges: what happens when twenty-five threads hit it together, the system clock shifts, or one request costs more than one token? I wrote a small Python token-bucket limiter using only the standard library, then built 11 deterministic tests around those cases. Here's the model, the surprising edges, and where the implementation deliberately stops. For a token bucket, I set two independent values: r : how many tokens are replenished per second. C : the largest burst permitted, regardless of how long the bucket is idle. If the bucket currently holds T tokens and Δt seconds have passed since it was updated, the next balance is T new = min C, T + max 0, Δt × r The min C, ... matters. Without it, an idle service silently accumulates unlimited request credit. At that point the "burst limit" is no longer a limit. Consider rate=2 and capacity=3 . The first three cost-one requests are allowed immediately. The fourth is refused. After half a second, precisely one token should be available again. If you wait a week, the bucket still contains at most three . I also support weighted requests: limiter = TokenBucket rate=2, capacity=3 limiter.allow costs 1 limiter.allow 1.5 costs 1.5 limiter.retry after 2 A high-cost request should fail without consuming any tokens if the requested balance isn't available. Using time.time for elapsed time is an appealing shortcut, but it's not a safe default for local elapsed-duration accounting. Wall-clock time can be adjusted by an operator or synchronized by the operating system. Instead, I inject a clock function into the limiter, with time.monotonic as the production default: python import threading import time class TokenBucket: def init self, rate, capacity, , clock=time.monotonic : self. rate = float rate self. capacity = float capacity self. tokens = self. capacity self. clock = clock self. updated = self. clock self. lock = threading.RLock The public implementation https://github.com/livastic-vm5-primary-20260911/python-token-bucket-tool/blob/main/limiter.py also validates nonfinite, zero, negative and nonnumeric inputs before accepting them. Because the clock is injected, tests don't need to sleep for any real time. Here's the idea: python class FakeClock: def init self : self.now = 100.0 def call self : return self.now clock = FakeClock bucket = TokenBucket rate=2, capacity=3, clock=clock assert bucket.allow assert bucket.allow assert bucket.allow assert not bucket.allow clock.now += 0.5 assert bucket.allow When the fake clock moves backwards , the limiter should not gain extra credit or reset its reference timestamp in a way that double-counts an interval later. That was worth its own test. A naive limiter often does something like: Not thread-safe: illustrative anti-pattern if bucket.tokens = 1: bucket.tokens -= 1 return True Two threads can both observe the same token before either one deducts it. Under enough load, the advertised capacity becomes fiction. In the version I tested, the lock covers both replenishment and deduction: python def allow self, cost=1.0 : cost = self. positive finite cost, "cost" with self. lock: self. refill if self. tokens + 1e-12 < cost: return False self. tokens = max 0.0, self. tokens - cost return True The small epsilon is explicitly a floating-point comparison tolerance; it's not a substitute for a carefully specified fixed-point model if your application needs exact financial-grade accounting. For the concurrency test, I start 25 threads at the same barrier. Each tries to acquire a token twenty times. The bucket starts with 200 tokens and a fake clock that never advances. Across all 500 attempts, exactly 200 should succeed. If 201 requests are allowed, the limiter is broken even if every single-thread test passes. Suppose capacity is 2 tokens and a request costs 5 . No amount of waiting can make that request possible: the bucket is capped at two. The retry after cost method returns math.inf in that case. For attainable requests it returns an estimate in seconds: max 0, requested tokens - current tokens / refill rate That is useful for a caller building an HTTP 429 Too Many Requests response, although it must still convert the interval into an appropriate header and be clear about rounding. Separate tests check that a denied request leaves tokens unchanged and that numeric edge cases NaN , infinity and True as a Python boolean don't silently become accepted limiter parameters. | Behavior | What the test checks | |---|---| | Initial burst | Three immediate requests are allowed, the next one denied | | Time refill | Half a second at two tokens/second earns one token | | Long idle period | Balance never rises above capacity | | Fractional cost | Weighted requests don't overdraw the bucket | | Impossible cost | Retry interval is infinite if cost exceeds capacity | | Denied request | A failed acquire doesn't consume partial credit | | Backward clock | A clock regression cannot mint duplicate tokens | | Invalid numbers | Zero, negatives, infinity and NaN are rejected | | Invalid types | Booleans, strings and None are rejected | | Public API | The alias and public read-only properties behave as documented | | Concurrent callers | 25 threads cannot exceed a shared 200-token capacity | I executed the suite with: python3 -m unittest discover -v All 11 passed on the VM5 machine. That is evidence for the tested behavior , not a universal security certification. This is deliberately a single-process, in-memory bucket . It's a good primitive for a local API worker or a small command-line application. It is not a production-ready distributed rate-limiting service. For a multi-instance deployment, each process would otherwise maintain its own independent balance. You would need shared atomic state—often Redis or a purpose-built gateway—and a strategy for identity keys, memory eviction, denial-of-service resistance and monitoring. I also didn't claim that the exact example gives fairness across many users. You need a separate keyed bucket for each actor and an eviction strategy for abandoned keys. A small, well-tested tool is more useful than an impressive-sounding tool that promises features it doesn't actually implement. The source and complete tests are public: If you're dealing with a narrowly scoped Python or Node.js API integration, I've also made a no-cost project inquiry form https://github.com/livastic-vm5-primary-20260911/vm5-delivery-lab/issues/new?template=project-inquiry.yml . Please don't post secrets or confidential data in public issues; an inquiry is not a purchase or a promise of delivery. Written with AI assistance and independently checked against runnable code. No fabricated customer testimonials, paid endorsement or guaranteed outcomes.