154 lines
4.2 KiB
Python
154 lines
4.2 KiB
Python
import importlib
|
|
from contextlib import contextmanager
|
|
|
|
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("BROKER_TOKEN_KEY", "test-broker-token-key")
|
|
monkeypatch.setenv("RESET_OTP_SECRET", "test-reset-secret")
|
|
|
|
import app.main as app_main
|
|
|
|
importlib.reload(app_main)
|
|
return app_main.create_app()
|
|
|
|
|
|
class _FakeCursor:
|
|
def __init__(self, row):
|
|
self._row = row
|
|
|
|
def __enter__(self):
|
|
return self
|
|
|
|
def __exit__(self, exc_type, exc, tb):
|
|
return False
|
|
|
|
def execute(self, _query, _params=None):
|
|
return None
|
|
|
|
def fetchone(self):
|
|
return self._row
|
|
|
|
|
|
class _FakeConnection:
|
|
def __init__(self, row):
|
|
self._row = row
|
|
|
|
def __enter__(self):
|
|
return self
|
|
|
|
def __exit__(self, exc_type, exc, tb):
|
|
return False
|
|
|
|
def cursor(self, *args, **kwargs):
|
|
return _FakeCursor(self._row)
|
|
|
|
|
|
@contextmanager
|
|
def _fake_db_connection(row):
|
|
yield _FakeConnection(row)
|
|
|
|
|
|
def test_admin_access_requires_auth(monkeypatch):
|
|
app = _build_app(monkeypatch)
|
|
client = TestClient(app)
|
|
|
|
response = client.get("/api/admin/access")
|
|
|
|
assert response.status_code == 401
|
|
assert response.json() == {"detail": "Not authenticated"}
|
|
|
|
|
|
def test_admin_access_forbids_non_admin(monkeypatch):
|
|
app = _build_app(monkeypatch)
|
|
client = TestClient(app)
|
|
|
|
import app.admin_auth as admin_auth
|
|
|
|
monkeypatch.setattr(admin_auth, "get_user_for_session", lambda _sid: {"id": "user-1"})
|
|
monkeypatch.setattr(
|
|
admin_auth,
|
|
"db_connection",
|
|
lambda: _fake_db_connection(("user-1", "normal@example.com", "USER", False, False)),
|
|
)
|
|
|
|
response = client.get("/api/admin/access", cookies={"session_id": "session-1"})
|
|
|
|
assert response.status_code == 403
|
|
assert response.json() == {"detail": "Admin access required"}
|
|
|
|
|
|
def test_admin_access_returns_identity(monkeypatch):
|
|
app = _build_app(monkeypatch)
|
|
client = TestClient(app)
|
|
|
|
import app.admin_auth as admin_auth
|
|
|
|
monkeypatch.setattr(admin_auth, "get_user_for_session", lambda _sid: {"id": "admin-1"})
|
|
monkeypatch.setattr(
|
|
admin_auth,
|
|
"db_connection",
|
|
lambda: _fake_db_connection(("admin-1", "admin@example.com", "ADMIN", True, False)),
|
|
)
|
|
|
|
response = client.get("/api/admin/access", cookies={"session_id": "session-admin"})
|
|
|
|
assert response.status_code == 200
|
|
assert response.json() == {
|
|
"id": "admin-1",
|
|
"username": "admin@example.com",
|
|
"role": "ADMIN",
|
|
"can_manage_admins": False,
|
|
}
|
|
|
|
|
|
def test_admin_overview_returns_data(monkeypatch):
|
|
app = _build_app(monkeypatch)
|
|
client = TestClient(app)
|
|
|
|
import app.admin_auth as admin_auth
|
|
import app.admin_router as admin_router
|
|
|
|
monkeypatch.setattr(admin_auth, "get_user_for_session", lambda _sid: {"id": "admin-1"})
|
|
monkeypatch.setattr(
|
|
admin_auth,
|
|
"db_connection",
|
|
lambda: _fake_db_connection(("admin-1", "admin@example.com", "SUPER_ADMIN", True, True)),
|
|
)
|
|
monkeypatch.setattr(
|
|
admin_router,
|
|
"get_overview",
|
|
lambda: {
|
|
"total_users": 2,
|
|
"users_logged_in_last_24h": 1,
|
|
"total_runs": 3,
|
|
"running_runs": 1,
|
|
"stopped_runs": 1,
|
|
"error_runs": 1,
|
|
"live_runs_count": 1,
|
|
"paper_runs_count": 2,
|
|
"orders_last_24h": 5,
|
|
"trades_last_24h": 4,
|
|
"sip_executed_last_24h": 1,
|
|
"unresolved_orders": 2,
|
|
"blocked_strategies": 1,
|
|
"open_support_tickets": 3,
|
|
"top_errors": [],
|
|
},
|
|
)
|
|
|
|
response = client.get("/api/admin/overview", cookies={"session_id": "session-admin"})
|
|
|
|
assert response.status_code == 200
|
|
assert response.json()["unresolved_orders"] == 2
|
|
assert response.json()["blocked_strategies"] == 1
|
|
assert response.json()["open_support_tickets"] == 3
|