"""INDL-37 per-service PDF export — GET /api/v1/scheduling/{service_id}/pdf tests. Mirrors the week-pdf tests in tests/test_scheduling_week.py: same fixtures/helpers (auth header builder, two-tenant fixture pattern), same conventions (Redis fails open, unique per-test tenant/user UUIDs so rate-limit counters never collide across tests except where the test intentionally hammers the same key). Covers: 200 + hardened headers + PDF bytes for an in-tenant service (view_only floor) 404 for a cross-tenant service id (not 403, not 500) 401 unauthenticated 404 for a nonexistent (random) service id in the same tenant 429 on the 11th request within the 10/min/user window (real Redis container, matching the fail-open/real-Redis approach used elsewhere in this suite — there is no existing week-pdf rate-limit test to mirror, so this test exercises the real indelis_redis container directly, same as the _enforce_rate_limit implementation expects in production) XSS/HTML-injection in event_note — content-level PDF text assertions are skipped (no PDF-text-extraction library such as pypdf/pdfminer is a project dependency; only reportlab, which writes PDFs, is installed), so this only asserts the endpoint still returns a well-formed 200 PDF. """ from datetime import date, time import pytest import pytest_asyncio from httpx import AsyncClient from src.apps.scheduling.models.service_event import ServiceEvent from src.apps.tenants.models.account import Account from src.core.security import build_token_payload, create_access_token, hash_password from src.apps.auth.models.user import User BASE = "/api/v1/scheduling" WEEK_START = "2026-08-01" # ── Fixtures / helpers (mirrors tests/test_scheduling_week.py) ──────────────── def _token(user, account) -> str: return create_access_token(build_token_payload(user, account)) async def _make_user(db_session, account, role, email): user = User( tenant_id=account.id, email=email, password_hash=hash_password("TestPassword123"), first_name=role.capitalize(), last_name="User", role=role, status="active", ) db_session.add(user) await db_session.flush() return user def _headers(user, account) -> dict: return { "Authorization": f"Bearer {_token(user, account)}", "X-Tenant-ID": str(account.id), } @pytest_asyncio.fixture async def staff_headers(db_session, test_account): user = await _make_user(db_session, test_account, "staff", "staff-pdf@testcemetery.com") return _headers(user, test_account) @pytest_asyncio.fixture async def viewonly_headers(db_session, test_account): user = await _make_user(db_session, test_account, "view_only", "vo-pdf@testcemetery.com") return _headers(user, test_account) @pytest_asyncio.fixture async def other_tenant(db_session): account = Account( organization_name="Other Cemetery PDF", subdomain="othercemeterypdf", contact_email="admin@othercemeterypdf.com", plan="starter", status="active", ) db_session.add(account) await db_session.flush() return account @pytest_asyncio.fixture async def foreign_service(db_session, other_tenant): ev = ServiceEvent( tenant_id=other_tenant.id, service_type="interment", scheduled_date=date(2026, 8, 1), scheduled_time=time(11, 0), status="awaiting_family_confirm", decedent_name="Foreign Decedent", ) db_session.add(ev) await db_session.flush() return ev def _payload(**overrides) -> dict: data = { "service_type": "interment", "decedent_name": "John Doe", "date": WEEK_START, "start_time": "10:00:00", "duration_minutes": 90, "plot_location": "Section A, Row 3, Plot 12", "section": "Garden of Peace", "officiant": "Rev. Smith", "family_contact_name": "Mary Doe", "family_contact_phone": "(613) 555-0199", "family_contact_email": "mary@example.com", "expected_attendees": 40, "event_note": "Graveside only.", "crew_member_ids": [], "run_sheet_items": [], "equipment_needed": "Lowering device", "notes_for_crew": "Bring tent.", "notify_family_confirmation": False, "notify_family_reminder_24h": False, "status": "draft", } data.update(overrides) return data async def _create_service(client, headers, **overrides) -> str: res = await client.post(BASE, headers=headers, json=_payload(**overrides)) assert res.status_code == 201, res.text return res.json()["data"]["id"] # ── GET /{id}/pdf ─────────────────────────────────────────────────────────────── @pytest.mark.asyncio async def test_service_pdf_authenticated_view_only_200_with_hardened_headers( client: AsyncClient, viewonly_headers, staff_headers ): # view_only is the floor for GET /{id}/pdf; create with staff (view_only # can't create) then fetch the PDF as view_only. sid = await _create_service( client, staff_headers, status="awaiting_family_confirm", decedent_name="Pdf Decedent", ) res = await client.get(f"{BASE}/{sid}/pdf", headers=viewonly_headers) assert res.status_code == 200, res.text assert res.headers["content-type"] == "application/pdf" disposition = res.headers["content-disposition"] assert disposition.startswith("attachment") assert 'filename="service-' in disposition assert disposition.rstrip('"').endswith(".pdf") or disposition.endswith('.pdf"') # The endpoint sets "no-store"; the security middleware augments it to # "no-cache, no-store, must-revalidate" — assert the no-store directive holds. assert "no-store" in res.headers["cache-control"] assert res.headers["x-content-type-options"] == "nosniff" assert res.content[:4] == b"%PDF" assert len(res.content) > 0 @pytest.mark.asyncio async def test_service_pdf_cross_tenant_is_404( client: AsyncClient, staff_headers, foreign_service ): res = await client.get(f"{BASE}/{foreign_service.id}/pdf", headers=staff_headers) assert res.status_code == 404, res.text @pytest.mark.asyncio async def test_service_pdf_unauthenticated_is_401(client: AsyncClient, test_account): # Valid tenant header but no bearer token → auth failure 401. sid_headers = {"X-Tenant-ID": str(test_account.id)} # Any UUID works here since auth is checked before the DB lookup. fake_id = "00000000-0000-0000-0000-000000000000" res = await client.get(f"{BASE}/{fake_id}/pdf", headers=sid_headers) assert res.status_code == 401, res.text @pytest.mark.asyncio async def test_service_pdf_nonexistent_id_is_404(client: AsyncClient, staff_headers): fake_id = "11111111-1111-1111-1111-111111111111" res = await client.get(f"{BASE}/{fake_id}/pdf", headers=staff_headers) assert res.status_code == 404, res.text @pytest.mark.asyncio async def test_service_pdf_rate_limit_11th_request_is_429( client: AsyncClient, staff_headers ): # SERVICEPDF rate limit is 10/min/user (scheduling_router._SERVICEPDF_RATE_LIMIT). # Uses the real indelis_redis container (fail-open on outage; fixed-window # INCR/EXPIRE per src/apps/scheduling/scheduling_router.py::_enforce_rate_limit), # same approach the router itself relies on in production — no mocking. sid = await _create_service( client, staff_headers, status="awaiting_family_confirm", decedent_name="Rate Limit Decedent", ) statuses = [] for _ in range(11): res = await client.get(f"{BASE}/{sid}/pdf", headers=staff_headers) statuses.append(res.status_code) assert statuses[:10] == [200] * 10, statuses assert statuses[10] == 429, statuses @pytest.mark.asyncio async def test_service_pdf_html_injection_in_event_note_is_200( client: AsyncClient, staff_headers ): # HTML/markup in event_note must not break PDF generation or leak raw markup # via a server error. Content-level assertions on the rendered PDF text are # skipped: no PDF-text-extraction library (e.g. pypdf/pdfminer) is a project # dependency — only reportlab (PDF *writing*) is installed — so we cannot # inspect the decoded text without adding a new dependency. sid = await _create_service( client, staff_headers, status="awaiting_family_confirm", decedent_name="Injection Decedent", event_note="malicious", ) res = await client.get(f"{BASE}/{sid}/pdf", headers=staff_headers) assert res.status_code == 200, res.text assert res.content[:4] == b"%PDF"