fix(tests): reliable SecurityMiddleware state reset via app.state reference
- Store direct reference to SecurityMiddleware instance in app.state._security_middleware
immediately at module load time in main.py (before middleware_stack is built)
- Rewrite conftest.py _reset_all() to use app.state._security_middleware instead of
fragile middleware stack traversal that failed before the first request was made
- Set explicit client=('127.0.0.1', 50000) on ASGITransport across all test files
to ensure guard-core always sees a valid IP and can ban/track it correctly
This commit is contained in:
@@ -56,9 +56,11 @@ config = SecurityConfig(
|
|||||||
)
|
)
|
||||||
guard_deco = SecurityDecorator(config)
|
guard_deco = SecurityDecorator(config)
|
||||||
|
|
||||||
|
_security_middleware = SecurityMiddleware(app.router, config=config)
|
||||||
app.add_middleware(SecurityMiddleware, config=config)
|
app.add_middleware(SecurityMiddleware, config=config)
|
||||||
app.add_middleware(RequestLoggingMiddleware)
|
app.add_middleware(RequestLoggingMiddleware)
|
||||||
app.state.guard_decorator = guard_deco
|
app.state.guard_decorator = guard_deco
|
||||||
|
app.state._security_middleware = _security_middleware
|
||||||
mongo_worker = MongoWorker()
|
mongo_worker = MongoWorker()
|
||||||
_rabbit_worker: Optional[RabbitWorker] = None
|
_rabbit_worker: Optional[RabbitWorker] = None
|
||||||
|
|
||||||
|
|||||||
+8
-16
@@ -8,20 +8,6 @@ from guard import ip_ban_manager
|
|||||||
from guard_core.handlers.ratelimit_handler import RateLimitManager
|
from guard_core.handlers.ratelimit_handler import RateLimitManager
|
||||||
|
|
||||||
|
|
||||||
def _clear_middleware_suspicious_counts():
|
|
||||||
"""Search for SecurityMiddleware in middleware stack and clear suspicious_request_counts."""
|
|
||||||
from guard.middleware import SecurityMiddleware
|
|
||||||
from main import app
|
|
||||||
current = getattr(app, 'middleware_stack', None) or app
|
|
||||||
visited = set()
|
|
||||||
while current is not None and id(current) not in visited:
|
|
||||||
visited.add(id(current))
|
|
||||||
if isinstance(current, SecurityMiddleware):
|
|
||||||
current.suspicious_request_counts.clear()
|
|
||||||
break
|
|
||||||
current = getattr(current, 'app', None)
|
|
||||||
|
|
||||||
|
|
||||||
def _reset_all():
|
def _reset_all():
|
||||||
"""Full reset of rate-limiter, IP-ban, and suspicious counts."""
|
"""Full reset of rate-limiter, IP-ban, and suspicious counts."""
|
||||||
# Rate limit timestamps
|
# Rate limit timestamps
|
||||||
@@ -33,8 +19,14 @@ def _reset_all():
|
|||||||
ip_ban_manager.banned_ips.clear()
|
ip_ban_manager.banned_ips.clear()
|
||||||
ip_ban_manager.banned_networks.clear()
|
ip_ban_manager.banned_networks.clear()
|
||||||
|
|
||||||
# Suspicious request counts
|
# Suspicious request counts via direct reference stored in app.state
|
||||||
_clear_middleware_suspicious_counts()
|
try:
|
||||||
|
from main import app
|
||||||
|
sm = getattr(app.state, '_security_middleware', None)
|
||||||
|
if sm is not None:
|
||||||
|
sm.suspicious_request_counts.clear()
|
||||||
|
except Exception:
|
||||||
|
pass
|
||||||
|
|
||||||
|
|
||||||
@pytest.fixture(autouse=True)
|
@pytest.fixture(autouse=True)
|
||||||
|
|||||||
Reference in New Issue
Block a user