saksham
1a4c80958f
FastAPI + Celery + Next.js + Postgres/Redis app with company monitoring, source collection, LLM-based change analysis, enrichment, and account security (Turnstile, escalating lockout, email verification).
122 lines
4.9 KiB
Python
122 lines
4.9 KiB
Python
"""/system/status and /system/logs - especially that /system/logs is
|
|
admin-only (Phase 19 - it exposes operational internals, not something any
|
|
registered user should read) and that the live log feed actually captures
|
|
what the app logs. Server-wide secret management (Turnstile site
|
|
key/secret) moved to /system/secrets - see test_system_secrets.py."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import uuid
|
|
|
|
from fastapi.testclient import TestClient
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from app.core.config import get_settings
|
|
from app.core.logging import get_logger
|
|
from app.main import app
|
|
from app.repositories.user_repository import UserRepository
|
|
|
|
|
|
def _register_and_login(client: TestClient) -> dict[str, str]:
|
|
email = f"user-{uuid.uuid4().hex[:12]}@example.com"
|
|
client.post(
|
|
"/api/v1/auth/register",
|
|
json={"email": email, "password": "correct-horse-1", "display_name": "Test User"},
|
|
)
|
|
tokens = client.post(
|
|
"/api/v1/auth/login", json={"email": email, "password": "correct-horse-1"}
|
|
).json()
|
|
return {"Authorization": f"Bearer {tokens['access_token']}"}
|
|
|
|
|
|
async def _register_admin_and_login(client: TestClient, db_session: AsyncSession) -> dict[str, str]:
|
|
"""Registration never accepts is_admin from the client - promote
|
|
directly in the DB, the same way a real operator would via a one-off
|
|
script/console, not through any HTTP-exposed path."""
|
|
email = f"admin-{uuid.uuid4().hex[:12]}@example.com"
|
|
client.post(
|
|
"/api/v1/auth/register",
|
|
json={"email": email, "password": "correct-horse-1", "display_name": "Admin User"},
|
|
)
|
|
user = await UserRepository(db_session).get_by_email(email)
|
|
user.is_admin = True
|
|
await db_session.commit()
|
|
tokens = client.post(
|
|
"/api/v1/auth/login", json={"email": email, "password": "correct-horse-1"}
|
|
).json()
|
|
return {"Authorization": f"Bearer {tokens['access_token']}"}
|
|
|
|
|
|
def test_system_status_reports_not_localhost_for_default_test_client(client: TestClient):
|
|
# Starlette's TestClient defaults its ASGI scope client to
|
|
# ("testclient", 50000), not a loopback address - this is the "someone
|
|
# not on this machine" case.
|
|
resp = client.get("/api/v1/system/status")
|
|
assert resp.status_code == 200
|
|
assert resp.json()["is_localhost"] is False
|
|
|
|
|
|
def test_system_status_reports_localhost_for_loopback_client():
|
|
with TestClient(app, client=("127.0.0.1", 54321)) as loopback_client:
|
|
resp = loopback_client.get("/api/v1/system/status")
|
|
assert resp.status_code == 200
|
|
assert resp.json()["is_localhost"] is True
|
|
|
|
|
|
def test_system_status_treats_configured_extra_ip_as_localhost():
|
|
# Docker Desktop's bridge networking means host-originated traffic
|
|
# never arrives as literal loopback - additional_trusted_local_ips is
|
|
# the opt-in escape hatch for that, see app.core.security.is_localhost.
|
|
test_settings = get_settings().model_copy(
|
|
update={"additional_trusted_local_ips": "172.18.0.1, 10.0.0.5"}
|
|
)
|
|
app.dependency_overrides[get_settings] = lambda: test_settings
|
|
try:
|
|
with TestClient(app, client=("172.18.0.1", 54321)) as bridge_client:
|
|
resp = bridge_client.get("/api/v1/system/status")
|
|
assert resp.status_code == 200
|
|
assert resp.json()["is_localhost"] is True
|
|
finally:
|
|
app.dependency_overrides.pop(get_settings, None)
|
|
|
|
|
|
def test_system_status_does_not_trust_an_unlisted_ip():
|
|
test_settings = get_settings().model_copy(update={"additional_trusted_local_ips": "172.18.0.1"})
|
|
app.dependency_overrides[get_settings] = lambda: test_settings
|
|
try:
|
|
with TestClient(app, client=("203.0.113.9", 54321)) as stranger_client:
|
|
resp = stranger_client.get("/api/v1/system/status")
|
|
assert resp.status_code == 200
|
|
assert resp.json()["is_localhost"] is False
|
|
finally:
|
|
app.dependency_overrides.pop(get_settings, None)
|
|
|
|
|
|
async def test_system_logs_captures_and_categorizes_real_log_calls(
|
|
client: TestClient, db_session: AsyncSession
|
|
):
|
|
headers = await _register_admin_and_login(client, db_session)
|
|
marker = f"phase17-test-marker-{uuid.uuid4().hex[:8]}"
|
|
|
|
logger = get_logger("tests.system_logs")
|
|
logger.warning("test_api_style_failure", event_marker=marker)
|
|
|
|
resp = client.get("/api/v1/system/logs", headers=headers)
|
|
assert resp.status_code == 200
|
|
entries = resp.json()
|
|
match = next(e for e in entries if e["context"].get("event_marker") == marker)
|
|
assert match["category"] == "api_error"
|
|
assert match["level"] == "warning"
|
|
assert match["event"] == "test_api_style_failure"
|
|
|
|
|
|
def test_system_logs_requires_auth(client: TestClient):
|
|
resp = client.get("/api/v1/system/logs")
|
|
assert resp.status_code == 401
|
|
|
|
|
|
def test_system_logs_non_admin_forbidden(client: TestClient):
|
|
headers = _register_and_login(client)
|
|
resp = client.get("/api/v1/system/logs", headers=headers)
|
|
assert resp.status_code == 403
|