from __future__ import annotations from django.core.cache import cache from django.test import override_settings from yksa_kit import host_rate_limit def test_host_from_url(): assert host_rate_limit.host_from_url("https://Db.SatNOGS.org/api/") == "db.satnogs.org" assert host_rate_limit.host_from_url("not a url") == "" def test_development_never_blocks(): cache.clear() host_rate_limit.acquire("example.com", per_minute=1) host_rate_limit.acquire("example.com", per_minute=1) assert cache.get("kit:host_rl:example.com") is None @override_settings(PRODUCTION=True) def test_production_counts_against_the_minute_window(): cache.clear() host_rate_limit.acquire("example.com", per_minute=5) assert cache.get("kit:host_rl:example.com") == 1 @override_settings(PRODUCTION=True) def test_a_zero_budget_means_unlimited(): """per_minute=0 is how a source says 'no published limit', not 'no requests'.""" cache.clear() host_rate_limit.acquire("example.com", per_minute=0) assert cache.get("kit:host_rl:example.com") is None @override_settings(PRODUCTION=True) def test_the_wait_gives_up_rather_than_stalling_the_worker(): cache.clear() cache.set("kit:host_rl:example.com", 99, timeout=60) host_rate_limit.acquire("example.com", per_minute=1, wait_seconds=0.0)