mjalmoudy94
717e53a7ce
مخزن اختصاصی برنامهٔ وب؛ محتوای پوشهٔ web در ریشهٔ مخزن قرار گرفت. تاریخچهٔ پروژهٔ ربات (agy-bot) در این مخزن وجود ندارد و هیچ وابستگی runtime به آن نیست.
475 lines
17 KiB
Python
475 lines
17 KiB
Python
"""The error pipeline: from a raised exception to a row in the database.
|
|
|
|
These tests drive the real FastAPI app with the real middleware and the real
|
|
exception handlers, against the real recorder — only the database connection is
|
|
redirected. A unit test of the handlers would have missed the middleware, and a
|
|
unit test of the recorder would have missed the handlers; the value here is that
|
|
the whole path is exercised end to end.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from collections.abc import AsyncIterator
|
|
|
|
import pytest_asyncio
|
|
from fastapi import APIRouter, Depends
|
|
from httpx import ASGITransport, AsyncClient
|
|
from sqlalchemy import select
|
|
|
|
from app.core.config import settings
|
|
from app.core.database import get_session
|
|
from app.core.errors import ProviderUnavailableError
|
|
from app.infrastructure.db.models import ErrorEventRow
|
|
from app.interfaces.http.deps import OrgContext, get_org_context
|
|
from app.interfaces.http.request_context import REQUEST_ID_HEADER
|
|
from app.main import app
|
|
from tests.conftest import auth_header, signup
|
|
|
|
#: The password in this DSN must never reach the database.
|
|
LEAKY_DSN = "postgresql://agy:sup3rs3cret@db.internal:5432/agy"
|
|
|
|
TEST_PREFIX = "/api/__errortest__"
|
|
|
|
|
|
@pytest_asyncio.fixture
|
|
async def client(engine, session_factory):
|
|
"""The shared test client, except that app exceptions are not re-raised.
|
|
|
|
Starlette's ``ServerErrorMiddleware`` deliberately sends the 500 response
|
|
and *then* re-raises, so the ASGI server can log it. httpx mirrors that
|
|
re-raise by default, which would surface every intentional 500 as a test
|
|
error instead of a response. These tests care about what the caller
|
|
receives, so the re-raise is switched off.
|
|
"""
|
|
|
|
async def _override_session():
|
|
async with session_factory() as session:
|
|
try:
|
|
yield session
|
|
await session.commit()
|
|
except Exception:
|
|
await session.rollback()
|
|
raise
|
|
|
|
app.dependency_overrides[get_session] = _override_session
|
|
async with AsyncClient(
|
|
transport=ASGITransport(app=app, raise_app_exceptions=False),
|
|
base_url="http://test",
|
|
) as test_client:
|
|
yield test_client
|
|
app.dependency_overrides.clear()
|
|
|
|
|
|
@pytest_asyncio.fixture(autouse=True)
|
|
async def _recorder_uses_the_test_database(redirect_error_log) -> AsyncIterator[None]:
|
|
"""Redirect the recorder and the workers to the in-memory database.
|
|
|
|
See ``conftest.redirect_error_log``. Autouse here because every test in this
|
|
module deliberately triggers a failure path.
|
|
"""
|
|
yield
|
|
|
|
|
|
@pytest_asyncio.fixture
|
|
async def failing_routes() -> AsyncIterator[None]:
|
|
"""Mount throwaway routes that fail on purpose.
|
|
|
|
Registered on the real app so the real middleware, handlers and dependencies
|
|
run. Torn down afterwards so no other test can see them.
|
|
"""
|
|
router = APIRouter()
|
|
|
|
@router.get(f"{TEST_PREFIX}/unhandled")
|
|
async def _unhandled() -> None:
|
|
raise RuntimeError(f"kaboom while dialling {LEAKY_DSN}")
|
|
|
|
@router.get(f"{TEST_PREFIX}/provider-down")
|
|
async def _provider_down() -> None:
|
|
raise ProviderUnavailableError("سرویسدهنده در دسترس نیست.")
|
|
|
|
@router.get(f"{TEST_PREFIX}/value-error")
|
|
async def _value_error() -> None:
|
|
raise ValueError("chunk size must be positive")
|
|
|
|
@router.get(f"{TEST_PREFIX}/org/{{org_id}}")
|
|
async def _org_scoped(context: OrgContext = Depends(get_org_context)) -> None:
|
|
raise RuntimeError(f"failed inside org {context.org_id}")
|
|
|
|
app.include_router(router)
|
|
yield
|
|
app.router.routes = [
|
|
route
|
|
for route in app.router.routes
|
|
if not getattr(route, "path", "").startswith(TEST_PREFIX)
|
|
]
|
|
|
|
|
|
async def _rows(session_factory) -> list[ErrorEventRow]:
|
|
async with session_factory() as session:
|
|
result = await session.execute(select(ErrorEventRow).order_by(ErrorEventRow.created_at))
|
|
return list(result.scalars())
|
|
|
|
|
|
async def _only_row(session_factory) -> ErrorEventRow:
|
|
rows = await _rows(session_factory)
|
|
assert len(rows) == 1, f"expected exactly one error row, found {len(rows)}"
|
|
return rows[0]
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Request ids
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
async def test_every_response_carries_a_request_id(client: AsyncClient):
|
|
response = await client.get("/health")
|
|
assert response.status_code == 200
|
|
assert response.headers.get(REQUEST_ID_HEADER)
|
|
|
|
|
|
async def test_caller_supplied_request_id_is_honoured(client: AsyncClient):
|
|
"""Lets an upstream proxy or BFF correlate its logs with ours."""
|
|
response = await client.get("/health", headers={REQUEST_ID_HEADER: "trace-from-upstream"})
|
|
assert response.headers[REQUEST_ID_HEADER] == "trace-from-upstream"
|
|
|
|
|
|
async def test_an_absurdly_long_caller_request_id_is_capped(client: AsyncClient):
|
|
response = await client.get("/health", headers={REQUEST_ID_HEADER: "x" * 500})
|
|
assert len(response.headers[REQUEST_ID_HEADER]) <= 64
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Unhandled exceptions
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
async def test_unhandled_exception_is_persisted_with_full_context(
|
|
client: AsyncClient, failing_routes: None, session_factory
|
|
):
|
|
response = await client.get(f"{TEST_PREFIX}/unhandled", headers={REQUEST_ID_HEADER: "req-1"})
|
|
|
|
assert response.status_code == 500
|
|
# The caller is told the request id so a bug report is actionable — in the
|
|
# body *and* in the header. Starlette's ServerErrorMiddleware turns an
|
|
# unhandled exception into a response outside every user middleware, so the
|
|
# header has to be set by the handler; a middleware-only approach silently
|
|
# loses it on exactly the responses that matter most.
|
|
assert response.json()["error"]["request_id"] == "req-1"
|
|
assert response.headers.get(REQUEST_ID_HEADER) == "req-1"
|
|
|
|
row = await _only_row(session_factory)
|
|
assert row.kind == "unhandled_exception"
|
|
assert row.severity == "critical"
|
|
assert row.source == "http"
|
|
assert row.error_type == "RuntimeError"
|
|
assert row.status_code == 500
|
|
assert row.method == "GET"
|
|
assert row.path == f"{TEST_PREFIX}/unhandled"
|
|
assert row.request_id == "req-1"
|
|
assert row.fingerprint
|
|
assert row.stack, "a traceback is the single most useful field for fixing a bug"
|
|
assert row.resolved_at is None
|
|
|
|
|
|
async def test_error_responses_carry_the_header_for_every_handler(
|
|
client: AsyncClient, failing_routes: None
|
|
):
|
|
"""Each handler path must set the header, not just the body."""
|
|
checks = [
|
|
(f"{TEST_PREFIX}/unhandled", 500),
|
|
(f"{TEST_PREFIX}/provider-down", 503),
|
|
(f"{TEST_PREFIX}/value-error", 422),
|
|
]
|
|
for path, expected in checks:
|
|
response = await client.get(path, headers={REQUEST_ID_HEADER: "hdr-check"})
|
|
assert response.status_code == expected, path
|
|
assert response.headers.get(REQUEST_ID_HEADER) == "hdr-check", path
|
|
assert response.json()["error"]["request_id"] == "hdr-check", path
|
|
|
|
# A Pydantic validation failure, which uses FastAPI's own exception type.
|
|
response = await client.post(
|
|
"/api/auth/login", json={"email": "nope"}, headers={REQUEST_ID_HEADER: "hdr-check"}
|
|
)
|
|
assert response.status_code == 422
|
|
assert response.headers.get(REQUEST_ID_HEADER) == "hdr-check"
|
|
|
|
|
|
async def test_a_routing_404_still_carries_the_header(client: AsyncClient):
|
|
"""Handled by Starlette's own middleware, so it exercises a different path."""
|
|
response = await client.get("/api/definitely-not-a-route")
|
|
assert response.status_code == 404
|
|
assert response.headers.get(REQUEST_ID_HEADER)
|
|
|
|
|
|
async def test_a_secret_in_the_exception_message_never_reaches_the_database(
|
|
client: AsyncClient, failing_routes: None, session_factory
|
|
):
|
|
"""The whole reason the redactor exists.
|
|
|
|
The exception interpolates a DSN containing a password; the stored message
|
|
must not contain it, and neither must the stored stack.
|
|
"""
|
|
await client.get(f"{TEST_PREFIX}/unhandled")
|
|
|
|
row = await _only_row(session_factory)
|
|
assert "sup3rs3cret" not in row.message
|
|
assert "sup3rs3cret" not in row.stack
|
|
assert "db.internal" in row.message # the useful part survives
|
|
|
|
|
|
async def test_org_and_user_are_attributed_through_the_real_dependency(
|
|
client: AsyncClient, failing_routes: None, session_factory
|
|
):
|
|
token = (await signup(client, "errlog-org@example.com"))["access_token"]
|
|
org = (
|
|
await client.post(
|
|
"/api/organizations", json={"name": "سازمان آزمایشی"}, headers=auth_header(token)
|
|
)
|
|
).json()
|
|
|
|
response = await client.get(
|
|
f"{TEST_PREFIX}/org/{org['id']}", headers=auth_header(token)
|
|
)
|
|
assert response.status_code == 500
|
|
|
|
row = await _only_row(session_factory)
|
|
assert row.organization_id == org["id"]
|
|
assert row.user_id == org["owner_user_id"]
|
|
|
|
|
|
async def test_a_failure_outside_any_tenant_still_records(
|
|
client: AsyncClient, failing_routes: None, session_factory
|
|
):
|
|
await client.get(f"{TEST_PREFIX}/unhandled")
|
|
row = await _only_row(session_factory)
|
|
assert row.organization_id is None
|
|
assert row.user_id is None
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Which failures get recorded
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
async def test_server_side_app_error_is_always_recorded(
|
|
client: AsyncClient, failing_routes: None, session_factory
|
|
):
|
|
"""A 503 is an operational fault, not caller error — the switch must not
|
|
suppress it."""
|
|
response = await client.get(f"{TEST_PREFIX}/provider-down")
|
|
assert response.status_code == 503
|
|
|
|
row = await _only_row(session_factory)
|
|
assert row.kind == "provider_unavailable"
|
|
assert row.severity == "error"
|
|
assert row.status_code == 503
|
|
|
|
|
|
async def test_client_errors_are_not_recorded_by_default(
|
|
client: AsyncClient, session_factory
|
|
):
|
|
"""A public API takes constant 404s from scanners; recording them would
|
|
bury real bugs."""
|
|
assert settings.error_log_client_errors is False
|
|
response = await client.get("/api/organizations/does-not-exist")
|
|
assert response.status_code in (401, 404)
|
|
assert await _rows(session_factory) == []
|
|
|
|
|
|
async def test_client_errors_are_recorded_when_enabled(
|
|
client: AsyncClient, session_factory, monkeypatch
|
|
):
|
|
monkeypatch.setattr(settings, "error_log_client_errors", True)
|
|
token = (await signup(client, "errlog-404@example.com"))["access_token"]
|
|
response = await client.get("/api/organizations/nope", headers=auth_header(token))
|
|
assert response.status_code == 404
|
|
|
|
row = await _only_row(session_factory)
|
|
assert row.kind == "not_found"
|
|
assert row.severity == "warning", "a 404 must never look like a server fault"
|
|
assert row.status_code == 404
|
|
|
|
|
|
async def test_a_bare_value_error_is_recorded_even_with_client_errors_off(
|
|
client: AsyncClient, failing_routes: None, session_factory
|
|
):
|
|
"""``ValueError`` escapes are a typing gap, not caller input: the codebase
|
|
has a typed ``ValidationError`` for that. They are always worth seeing."""
|
|
assert settings.error_log_client_errors is False
|
|
response = await client.get(f"{TEST_PREFIX}/value-error")
|
|
assert response.status_code == 422
|
|
|
|
row = await _only_row(session_factory)
|
|
assert row.kind == "value_error"
|
|
assert row.severity == "warning"
|
|
assert row.error_type == "ValueError"
|
|
|
|
|
|
async def test_validation_failures_use_the_standard_envelope(
|
|
client: AsyncClient, session_factory
|
|
):
|
|
"""The web client reads ``error.code``/``error.message``; FastAPI's default
|
|
``{"detail": [...]}`` is unreadable to it."""
|
|
response = await client.post("/api/auth/login", json={"email": "not-an-email"})
|
|
assert response.status_code == 422
|
|
body = response.json()
|
|
assert body["error"]["code"] == "validation_error"
|
|
assert body["error"]["message"]
|
|
assert body["error"]["request_id"]
|
|
assert isinstance(body["error"]["details"], list)
|
|
|
|
|
|
async def test_the_switch_can_turn_recording_off_entirely(
|
|
client: AsyncClient, failing_routes: None, session_factory, monkeypatch
|
|
):
|
|
monkeypatch.setattr(settings, "error_log_enabled", False)
|
|
response = await client.get(f"{TEST_PREFIX}/unhandled")
|
|
|
|
# The response is unchanged — only the persistence is skipped.
|
|
assert response.status_code == 500
|
|
assert response.json()["error"]["request_id"]
|
|
assert await _rows(session_factory) == []
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Resilience — recording must never make things worse
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
async def test_a_broken_recorder_does_not_break_the_response(
|
|
client: AsyncClient, failing_routes: None, monkeypatch
|
|
):
|
|
"""If the database is unreachable, the caller must still get a clean 500.
|
|
|
|
This is the failure mode that matters most: the errors you cannot record are
|
|
exactly the database errors, and they must not cascade into a hang or an
|
|
unhandled exception inside the exception handler.
|
|
"""
|
|
import app.infrastructure.error_log as error_log
|
|
|
|
class _Exploding:
|
|
def __call__(self, *args, **kwargs):
|
|
raise RuntimeError("database is unreachable")
|
|
|
|
monkeypatch.setattr(error_log, "SessionFactory", _Exploding())
|
|
|
|
response = await client.get(f"{TEST_PREFIX}/unhandled", headers={REQUEST_ID_HEADER: "req-x"})
|
|
assert response.status_code == 500
|
|
assert response.json()["error"]["code"] == "internal_error"
|
|
assert response.json()["error"]["request_id"] == "req-x"
|
|
|
|
|
|
async def test_metadata_that_is_not_serialisable_does_not_raise(
|
|
client: AsyncClient, failing_routes: None, session_factory
|
|
):
|
|
"""``_safe_metadata`` must degrade, not explode, on odd payloads."""
|
|
from app.infrastructure.error_log import _safe_metadata
|
|
|
|
class _Weird:
|
|
def __repr__(self) -> str:
|
|
return "<weird>"
|
|
|
|
result = _safe_metadata({"obj": _Weird(), "password": "hunter2"})
|
|
assert result["password"] == "[REDACTED]"
|
|
assert "weird" in str(result["obj"])
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Worker and startup failures
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
async def test_a_failed_worker_pass_is_persisted(seeded, session_factory):
|
|
"""A worker that only logged to stderr would lose the failure the moment the
|
|
container was replaced."""
|
|
import asyncio
|
|
|
|
from app.workers.base import run_periodically
|
|
|
|
async def exploding_pass():
|
|
raise RuntimeError("billing rollup blew up")
|
|
|
|
outcomes = await run_periodically(
|
|
exploding_pass,
|
|
interval_seconds=1,
|
|
stop=asyncio.Event(),
|
|
run_once=True,
|
|
worker_name="billing",
|
|
)
|
|
|
|
# The pass is reported as failed, so a cron entry exits non-zero instead of
|
|
# silently succeeding.
|
|
assert len(outcomes) == 1
|
|
assert outcomes[0].failed == 1
|
|
assert outcomes[0].name == "billing"
|
|
|
|
row = await _only_row(session_factory)
|
|
assert row.kind == "worker_pass_failed"
|
|
assert row.source == "worker"
|
|
assert row.severity == "error"
|
|
assert row.error_type == "RuntimeError"
|
|
assert row.metadata_json["worker"] == "billing"
|
|
assert row.stack
|
|
# A worker pass has no HTTP request behind it.
|
|
assert row.path == ""
|
|
assert row.status_code is None
|
|
|
|
|
|
async def test_a_failed_task_run_is_persisted_with_its_tenant(
|
|
seeded, client: AsyncClient, session_factory, monkeypatch
|
|
):
|
|
"""A task that cannot run is a tenant-visible problem, so the row must name
|
|
the organization — otherwise nobody can find it in the console."""
|
|
from app.application.services.workforce_service import WorkforceService
|
|
from app.workers.ai_runtime import AiRuntimeWorker
|
|
from tests.test_workers import _make_org_with_employee
|
|
|
|
org_id, headers, project_id, employee_id = await _make_org_with_employee(
|
|
client, "errlog-worker@example.com"
|
|
)
|
|
task_id = (
|
|
await client.post(
|
|
f"/api/organizations/{org_id}/tasks",
|
|
json={
|
|
"title": "کاری که شکست میخورد",
|
|
"project_id": project_id,
|
|
"assigned_employee_id": employee_id,
|
|
},
|
|
headers=headers,
|
|
)
|
|
).json()["id"]
|
|
|
|
async def _boom(*args, **kwargs):
|
|
raise RuntimeError("provider returned a malformed response")
|
|
|
|
monkeypatch.setattr(WorkforceService, "start_run", _boom)
|
|
|
|
outcome = await AiRuntimeWorker().drain()
|
|
|
|
assert outcome.failed == 1, outcome.notes
|
|
row = await _only_row(session_factory)
|
|
assert row.kind == "task_run_failed"
|
|
assert row.source == "worker"
|
|
assert row.organization_id == org_id
|
|
assert row.metadata_json["task_id"] == task_id
|
|
assert row.metadata_json["employee_id"] == employee_id
|
|
|
|
|
|
async def test_a_startup_failure_is_persisted_as_critical(
|
|
seeded, session_factory
|
|
):
|
|
"""Boot failures are the hardest to diagnose after the fact: the process is
|
|
gone and stderr is usually discarded."""
|
|
from app.main import _record_startup_failure
|
|
|
|
try:
|
|
raise RuntimeError("could not create the schema")
|
|
except RuntimeError as exc:
|
|
await _record_startup_failure(exc)
|
|
|
|
row = await _only_row(session_factory)
|
|
assert row.kind == "startup_failed"
|
|
assert row.source == "startup"
|
|
assert row.severity == "critical"
|
|
assert row.metadata_json["app_env"] == settings.app_env
|