mirror of
https://github.com/ruvnet/RuView.git
synced 2026-04-28 05:59:32 +00:00
5bd0d59aa6
P1 fixes (this sprint): - P1-6: Extract sensing-server modules (cli, types, csi, pose) from main.rs - P1-7: DDA ray march for tomography — O(max(n)) replaces O(n^3) voxel scan - P1-8: Batch neural inference — Tensor::stack/split for single GPU call - P1-10: Eliminate 112KB/frame alloc — islice replaces deque→list copy P2 fixes (this quarter): - P2-11: Python unit tests for 8 modules (rate_limit, auth, error_handler, pose_service, stream_service, hardware_service, health_check, metrics) - P2-13: MAT simulated data safety guard — blocking overlay + pulsing banner - P2-14: Wire token blacklist into auth verification + logout endpoint - P2-15: Frame budget benchmark — confirms pipeline well under 50ms budget Addresses 8 of 10 remaining issues from QE analysis (ADR-080). Co-Authored-By: claude-flow <ruv@ruv.net>
62 lines
2.8 KiB
Python
62 lines
2.8 KiB
Python
"""Tests for rate limiting middleware."""
|
|
|
|
import pytest
|
|
from unittest.mock import MagicMock, AsyncMock, patch
|
|
|
|
|
|
class TestRateLimitMiddleware:
|
|
def test_init(self, mock_settings):
|
|
with patch("src.api.middleware.rate_limit.get_settings", return_value=mock_settings):
|
|
from src.api.middleware.rate_limit import RateLimitMiddleware
|
|
app = MagicMock()
|
|
mw = RateLimitMiddleware(app)
|
|
assert "anonymous" in mw.rate_limits
|
|
assert "authenticated" in mw.rate_limits
|
|
assert "admin" in mw.rate_limits
|
|
|
|
def test_exempt_paths(self, mock_settings):
|
|
with patch("src.api.middleware.rate_limit.get_settings", return_value=mock_settings):
|
|
from src.api.middleware.rate_limit import RateLimitMiddleware
|
|
app = MagicMock()
|
|
mw = RateLimitMiddleware(app)
|
|
assert "/health" in mw.exempt_paths
|
|
assert "/metrics" in mw.exempt_paths
|
|
|
|
def test_is_exempt(self, mock_settings):
|
|
with patch("src.api.middleware.rate_limit.get_settings", return_value=mock_settings):
|
|
from src.api.middleware.rate_limit import RateLimitMiddleware
|
|
app = MagicMock()
|
|
mw = RateLimitMiddleware(app)
|
|
assert mw._is_exempt_path("/health") is True
|
|
assert mw._is_exempt_path("/api/v1/pose/current") is False
|
|
|
|
def test_path_specific_limits(self, mock_settings):
|
|
with patch("src.api.middleware.rate_limit.get_settings", return_value=mock_settings):
|
|
from src.api.middleware.rate_limit import RateLimitMiddleware
|
|
app = MagicMock()
|
|
mw = RateLimitMiddleware(app)
|
|
assert "/api/v1/pose/current" in mw.path_limits
|
|
assert mw.path_limits["/api/v1/pose/current"]["requests"] == 60
|
|
|
|
def test_trusted_proxies_not_blocked(self, mock_settings):
|
|
with patch("src.api.middleware.rate_limit.get_settings", return_value=mock_settings):
|
|
from src.api.middleware.rate_limit import RateLimitMiddleware
|
|
app = MagicMock()
|
|
mw = RateLimitMiddleware(app)
|
|
assert not mw._is_client_blocked("new-client-id")
|
|
|
|
|
|
class TestRateLimitConfig:
|
|
def test_anonymous_limit(self, mock_settings):
|
|
with patch("src.api.middleware.rate_limit.get_settings", return_value=mock_settings):
|
|
from src.api.middleware.rate_limit import RateLimitMiddleware
|
|
app = MagicMock()
|
|
mw = RateLimitMiddleware(app)
|
|
assert mw.rate_limits["anonymous"]["burst"] == 10
|
|
|
|
def test_admin_limit(self, mock_settings):
|
|
with patch("src.api.middleware.rate_limit.get_settings", return_value=mock_settings):
|
|
from src.api.middleware.rate_limit import RateLimitMiddleware
|
|
app = MagicMock()
|
|
mw = RateLimitMiddleware(app)
|
|
assert mw.rate_limits["admin"]["requests"] == 10000
|