""" Unit tests for the Garmin request pacer. Nothing here touches the network, and the interval is zero (see conftest) so the accounting is tested without the waiting. """ import time import pytest from services import garmin_throttle as throttle class Recorder: """Stands in for a garminconnect client.""" def __init__(self): self.calls = [] self.display_name = "Tester" self.garth = object() def get_user_summary(self, date): self.calls.append(date) return {"date": date} def get_sleep_data(self, date): self.calls.append(date) return {} class TestPacedClient: def test_calls_reach_the_wrapped_client(self): inner = Recorder() client = throttle.pace(inner, "u") assert client.get_user_summary("2026-09-01") == {"date": "2026-09-01"} assert inner.calls == ["2026-09-01"] def test_every_call_is_counted(self): client = throttle.pace(Recorder(), "u") client.get_user_summary("d") client.get_sleep_data("d") assert client.limiter.used == 2 def test_non_callables_pass_through(self): """`display_name` is read by the library on every request, and `garth` carries the login flow — neither is an API call.""" inner = Recorder() client = throttle.pace(inner, "u") assert client.display_name == "Tester" assert client.garth is inner.garth assert client.limiter.used == 0 def test_attribute_writes_reach_the_client(self): inner = Recorder() client = throttle.pace(inner, "u") client.display_name = "Someone" assert inner.display_name == "Someone" def test_wrapping_twice_does_not_stack_two_waits(self): once = throttle.pace(Recorder(), "u") twice = throttle.pace(once, "u") assert twice is once def test_a_missing_method_still_raises_attribute_error(self): client = throttle.pace(Recorder(), "u") with pytest.raises(AttributeError): client.get_something_that_does_not_exist class TestBudget: def test_the_budget_stops_the_run(self): client = throttle.pace(Recorder(), "u", budget=3) for _ in range(3): client.get_sleep_data("d") with pytest.raises(throttle.BudgetExhausted): client.get_sleep_data("d") def test_remaining_counts_down(self): client = throttle.pace(Recorder(), "u", budget=5) client.get_sleep_data("d") assert client.limiter.remaining == 4 def test_a_new_run_resets_the_budget(self): client = throttle.pace(Recorder(), "u", budget=2) client.get_sleep_data("d") client.get_sleep_data("d") throttle.pace(client, "u", budget=2) client.get_sleep_data("d") # must not raise assert client.limiter.used == 1 def test_no_budget_means_unlimited(self): client = throttle.pace(Recorder(), "u", budget=0) # 0 is falsy but explicit; only None means unlimited. with pytest.raises(throttle.BudgetExhausted): client.get_sleep_data("d") limiter = throttle.limiter_for("u") limiter.start_run(None) client.get_sleep_data("d") assert limiter.remaining is None class TestSpacing: def test_calls_are_spaced_by_the_interval(self, monkeypatch): monkeypatch.setenv("GARMIN_MIN_INTERVAL_SECONDS", "0.05") client = throttle.pace(Recorder(), f"spacing-{time.monotonic()}") started = time.monotonic() for _ in range(4): client.get_sleep_data("d") # Three gaps between four calls; the first claims a free slot. assert time.monotonic() - started >= 0.05 * 3 def test_the_interval_is_read_per_call(self, monkeypatch): monkeypatch.setenv("GARMIN_MIN_INTERVAL_SECONDS", "2.5") assert throttle.min_interval() == 2.5 monkeypatch.delenv("GARMIN_MIN_INTERVAL_SECONDS") assert throttle.min_interval() == 0.5, "the researched default" def test_accounts_are_paced_independently(self): a = throttle.limiter_for("account-a") b = throttle.limiter_for("account-b") assert a is not b assert throttle.limiter_for("account-a") is a