Latte
b82e2866cb
Adds the pydantic-settings Settings with validated GITEA_URL, OAuth scopes and toolsets, the argparse CLI skeleton, catalog.model, identity and GiteaAPIError. Test fixtures now isolate env, cwd and state paths in tmp.
56 lines
1.8 KiB
Python
56 lines
1.8 KiB
Python
"""Tests for the request rate limiter."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import pytest
|
|
|
|
from aegis_gitea_mcp.audit import validate_audit_log_integrity
|
|
from aegis_gitea_mcp.config import get_settings
|
|
from aegis_gitea_mcp.rate_limit import (
|
|
RequestRateLimiter,
|
|
SlidingWindowLimiter,
|
|
get_rate_limiter,
|
|
reset_rate_limiter,
|
|
)
|
|
|
|
|
|
def test_sliding_window_blocks_after_limit(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
now = [1000.0]
|
|
monkeypatch.setattr("aegis_gitea_mcp.rate_limit.time.time", lambda: now[0])
|
|
limiter = SlidingWindowLimiter(max_requests=2, window_seconds=60)
|
|
|
|
assert limiter.allow("k")
|
|
assert limiter.allow("k")
|
|
assert not limiter.allow("k")
|
|
assert limiter.allow("other")
|
|
|
|
now[0] += 61
|
|
assert limiter.allow("k")
|
|
|
|
|
|
def test_request_limiter_per_ip_and_token(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
monkeypatch.setenv("GITEA_URL", "https://g.example.com")
|
|
monkeypatch.setenv("RATE_LIMIT_PER_MINUTE", "2")
|
|
limiter = RequestRateLimiter()
|
|
|
|
assert limiter.check("1.1.1.1", "tok").allowed
|
|
assert limiter.check("1.1.1.1", None).allowed
|
|
assert not limiter.check("1.1.1.1", "tok").allowed
|
|
|
|
# A different IP with the same token trips the per-token limiter.
|
|
other = limiter.check("2.2.2.2", "tok")
|
|
assert other.allowed or other.reason == "Per-token rate limit exceeded"
|
|
third = limiter.check("3.3.3.3", "tok")
|
|
assert not third.allowed
|
|
assert third.reason == "Per-token rate limit exceeded"
|
|
|
|
valid, errors = validate_audit_log_integrity(get_settings().resolved_audit_log_path)
|
|
assert valid, errors
|
|
|
|
|
|
def test_global_limiter_singleton(mock_env: None) -> None:
|
|
first = get_rate_limiter()
|
|
assert get_rate_limiter() is first
|
|
reset_rate_limiter()
|
|
assert get_rate_limiter() is not first
|