Phase 4: API Endpoint
- Add RandomNumberOutcomeResponse schema to schemas.py - Add GET /api/audit/random-jobs endpoint to routes/audit.py - Returns list of outcomes (newest first, no pagination) - Requires VIEW_AUDIT permission - Add tests: admin can access, regular user forbidden, unauthenticated 401
This commit is contained in:
parent
7beb213cf5
commit
b3ed81e8fd
3 changed files with 71 additions and 1 deletions
|
|
@ -10,11 +10,12 @@ from sqlalchemy.ext.asyncio import AsyncSession
|
||||||
|
|
||||||
from auth import require_permission
|
from auth import require_permission
|
||||||
from database import get_db
|
from database import get_db
|
||||||
from models import CounterRecord, Permission, SumRecord, User
|
from models import CounterRecord, Permission, RandomNumberOutcome, SumRecord, User
|
||||||
from schemas import (
|
from schemas import (
|
||||||
CounterRecordResponse,
|
CounterRecordResponse,
|
||||||
PaginatedCounterRecords,
|
PaginatedCounterRecords,
|
||||||
PaginatedSumRecords,
|
PaginatedSumRecords,
|
||||||
|
RandomNumberOutcomeResponse,
|
||||||
SumRecordResponse,
|
SumRecordResponse,
|
||||||
)
|
)
|
||||||
|
|
||||||
|
|
@ -115,3 +116,28 @@ async def get_sum_records(
|
||||||
per_page=per_page,
|
per_page=per_page,
|
||||||
total_pages=total_pages,
|
total_pages=total_pages,
|
||||||
)
|
)
|
||||||
|
|
||||||
|
|
||||||
|
@router.get("/random-jobs", response_model=list[RandomNumberOutcomeResponse])
|
||||||
|
async def get_random_job_outcomes(
|
||||||
|
db: AsyncSession = Depends(get_db),
|
||||||
|
_current_user: User = Depends(require_permission(Permission.VIEW_AUDIT)),
|
||||||
|
) -> list[RandomNumberOutcomeResponse]:
|
||||||
|
"""Get all random number job outcomes, newest first."""
|
||||||
|
query = select(RandomNumberOutcome).order_by(desc(RandomNumberOutcome.created_at))
|
||||||
|
result = await db.execute(query)
|
||||||
|
outcomes = result.scalars().all()
|
||||||
|
|
||||||
|
return [
|
||||||
|
RandomNumberOutcomeResponse(
|
||||||
|
id=outcome.id,
|
||||||
|
job_id=outcome.job_id,
|
||||||
|
triggered_by_user_id=outcome.triggered_by_user_id,
|
||||||
|
triggered_by_email=outcome.triggered_by.email,
|
||||||
|
value=outcome.value,
|
||||||
|
duration_ms=outcome.duration_ms,
|
||||||
|
status=outcome.status,
|
||||||
|
created_at=outcome.created_at,
|
||||||
|
)
|
||||||
|
for outcome in outcomes
|
||||||
|
]
|
||||||
|
|
|
||||||
|
|
@ -257,6 +257,24 @@ class AppointmentResponse(BaseModel):
|
||||||
PaginatedAppointments = PaginatedResponse[AppointmentResponse]
|
PaginatedAppointments = PaginatedResponse[AppointmentResponse]
|
||||||
|
|
||||||
|
|
||||||
|
# =============================================================================
|
||||||
|
# Random Number Job Schemas
|
||||||
|
# =============================================================================
|
||||||
|
|
||||||
|
|
||||||
|
class RandomNumberOutcomeResponse(BaseModel):
|
||||||
|
"""Response model for a random number job outcome."""
|
||||||
|
|
||||||
|
id: int
|
||||||
|
job_id: int
|
||||||
|
triggered_by_user_id: int
|
||||||
|
triggered_by_email: str
|
||||||
|
value: int
|
||||||
|
duration_ms: int
|
||||||
|
status: str
|
||||||
|
created_at: datetime
|
||||||
|
|
||||||
|
|
||||||
# =============================================================================
|
# =============================================================================
|
||||||
# Meta/Constants Schemas
|
# Meta/Constants Schemas
|
||||||
# =============================================================================
|
# =============================================================================
|
||||||
|
|
|
||||||
|
|
@ -272,6 +272,32 @@ class TestAuditAccess:
|
||||||
response = await client.get("/api/audit/sum")
|
response = await client.get("/api/audit/sum")
|
||||||
assert response.status_code == 401
|
assert response.status_code == 401
|
||||||
|
|
||||||
|
@pytest.mark.asyncio
|
||||||
|
async def test_admin_can_view_random_jobs(self, client_factory, admin_user):
|
||||||
|
"""Admin should be able to view random job outcomes."""
|
||||||
|
async with client_factory.create(cookies=admin_user["cookies"]) as client:
|
||||||
|
response = await client.get("/api/audit/random-jobs")
|
||||||
|
|
||||||
|
assert response.status_code == 200
|
||||||
|
# Returns a list (no pagination)
|
||||||
|
assert isinstance(response.json(), list)
|
||||||
|
|
||||||
|
@pytest.mark.asyncio
|
||||||
|
async def test_regular_user_cannot_view_random_jobs(
|
||||||
|
self, client_factory, regular_user
|
||||||
|
):
|
||||||
|
"""Regular users should be forbidden from random-jobs endpoint."""
|
||||||
|
async with client_factory.create(cookies=regular_user["cookies"]) as client:
|
||||||
|
response = await client.get("/api/audit/random-jobs")
|
||||||
|
|
||||||
|
assert response.status_code == 403
|
||||||
|
|
||||||
|
@pytest.mark.asyncio
|
||||||
|
async def test_unauthenticated_cannot_view_random_jobs(self, client):
|
||||||
|
"""Unauthenticated users should get 401."""
|
||||||
|
response = await client.get("/api/audit/random-jobs")
|
||||||
|
assert response.status_code == 401
|
||||||
|
|
||||||
|
|
||||||
# =============================================================================
|
# =============================================================================
|
||||||
# Offensive Security Tests - Bypass Attempts
|
# Offensive Security Tests - Bypass Attempts
|
||||||
|
|
|
||||||
Loading…
Add table
Add a link
Reference in a new issue