132 lines
4.3 KiB
Python
132 lines
4.3 KiB
Python
import importlib
|
|
import logging
|
|
|
|
from fastapi.testclient import TestClient
|
|
|
|
|
|
def _build_app(monkeypatch):
|
|
monkeypatch.setenv("APP_ENV", "test")
|
|
monkeypatch.setenv("DISABLE_STARTUP_TASKS", "1")
|
|
monkeypatch.setenv("DB_HOST", "localhost")
|
|
monkeypatch.setenv("DB_NAME", "trading_db")
|
|
monkeypatch.setenv("DB_USER", "trader")
|
|
monkeypatch.setenv("DB_PASSWORD", "test-password")
|
|
monkeypatch.setenv("CORS_ORIGINS", "http://localhost:3000")
|
|
monkeypatch.setenv("SUPPORT_GUARD_BACKEND", "memory")
|
|
monkeypatch.setenv("SUPPORT_GUARD_WINDOW_SECONDS", "900")
|
|
monkeypatch.setenv("SUPPORT_CREATE_LIMIT", "2")
|
|
monkeypatch.setenv("SUPPORT_STATUS_LIMIT", "3")
|
|
monkeypatch.setenv("SUPPORT_STATUS_TICKET_LIMIT", "2")
|
|
|
|
import app.main as app_main
|
|
|
|
importlib.reload(app_main)
|
|
return app_main.create_app()
|
|
|
|
|
|
def test_support_ticket_creation_is_throttled(monkeypatch, caplog):
|
|
app = _build_app(monkeypatch)
|
|
client = TestClient(app)
|
|
|
|
import app.services.support_abuse as support_abuse
|
|
import app.routers.support_ticket as support_router
|
|
|
|
support_abuse.reset_memory_support_guard_state()
|
|
monkeypatch.setattr(
|
|
support_router,
|
|
"create_ticket",
|
|
lambda **kwargs: {"ticket_id": "ticket-1", "status": "NEW", "created_at": "2026-04-08T00:00:00+00:00"},
|
|
)
|
|
|
|
payload = {
|
|
"name": "Trader",
|
|
"email": "trader@example.com",
|
|
"subject": "Need help",
|
|
"message": "Something happened",
|
|
}
|
|
|
|
with caplog.at_level(logging.WARNING):
|
|
first = client.post("/api/support/ticket", json=payload)
|
|
second = client.post("/api/support/ticket", json=payload)
|
|
third = client.post("/api/support/ticket", json=payload)
|
|
|
|
assert first.status_code == 200
|
|
assert second.status_code == 200
|
|
assert third.status_code == 429
|
|
assert "Support request blocked" in caplog.text
|
|
|
|
|
|
def test_invalid_ticket_probing_is_throttled(monkeypatch):
|
|
app = _build_app(monkeypatch)
|
|
client = TestClient(app)
|
|
|
|
import app.services.support_abuse as support_abuse
|
|
import app.routers.support_ticket as support_router
|
|
|
|
support_abuse.reset_memory_support_guard_state()
|
|
monkeypatch.setattr(support_router, "get_ticket_status", lambda ticket_id, email: None)
|
|
|
|
payload = {"email": "trader@example.com"}
|
|
|
|
first = client.post("/api/support/ticket/status/unknown-ticket", json=payload)
|
|
second = client.post("/api/support/ticket/status/unknown-ticket", json=payload)
|
|
third = client.post("/api/support/ticket/status/unknown-ticket", json=payload)
|
|
|
|
assert first.status_code == 404
|
|
assert second.status_code == 404
|
|
assert third.status_code == 429
|
|
|
|
|
|
def test_legitimate_status_lookup_still_works(monkeypatch):
|
|
app = _build_app(monkeypatch)
|
|
client = TestClient(app)
|
|
|
|
import app.services.support_abuse as support_abuse
|
|
import app.routers.support_ticket as support_router
|
|
|
|
support_abuse.reset_memory_support_guard_state()
|
|
monkeypatch.setattr(
|
|
support_router,
|
|
"get_ticket_status",
|
|
lambda ticket_id, email: {
|
|
"ticket_id": ticket_id,
|
|
"status": "NEW",
|
|
"created_at": "2026-04-08T00:00:00+00:00",
|
|
"updated_at": "2026-04-08T00:00:00+00:00",
|
|
},
|
|
)
|
|
|
|
response = client.post("/api/support/ticket/status/ticket-1", json={"email": "trader@example.com"})
|
|
|
|
assert response.status_code == 200
|
|
assert response.json()["ticket_id"] == "ticket-1"
|
|
|
|
|
|
def test_support_captcha_hook_blocks_without_matching_header(monkeypatch):
|
|
monkeypatch.setenv("SUPPORT_CAPTCHA_SECRET", "expected-captcha")
|
|
app = _build_app(monkeypatch)
|
|
client = TestClient(app)
|
|
|
|
import app.services.support_abuse as support_abuse
|
|
import app.routers.support_ticket as support_router
|
|
|
|
support_abuse.reset_memory_support_guard_state()
|
|
monkeypatch.setattr(
|
|
support_router,
|
|
"create_ticket",
|
|
lambda **kwargs: {"ticket_id": "ticket-1", "status": "NEW", "created_at": "2026-04-08T00:00:00+00:00"},
|
|
)
|
|
|
|
response = client.post(
|
|
"/api/support/ticket",
|
|
json={
|
|
"name": "Trader",
|
|
"email": "trader@example.com",
|
|
"subject": "Need help",
|
|
"message": "Something happened",
|
|
},
|
|
)
|
|
|
|
assert response.status_code == 403
|
|
assert response.json() == {"detail": "Support verification failed"}
|