Phase 5 Batch 5 Task 5.18: Report Generator PDF-Support & Druck-Funktionen
- Installed WeasyPrint 69.0 for PDF generation - Created 5 Jinja2 HTML templates: contact_list, calendar_week, calendar_month, company_list, audit_log - All templates: A4 landscape, print-optimized CSS (@media print, page margins) - Extended output_format in schemas.py: added pdf and print - Created pdf_generator.py: Jinja2 + WeasyPrint pipeline with preset support - Modified routes.py: PDF/print generation, preset endpoints (/presets, /presets/generate) - Added RBAC (require_permission) to all report endpoints - Generate endpoints return StreamingResponse (file download) directly - 7 backend tests: presets listing, PDF/CSV generation, template CRUD, RBAC
This commit is contained in:
Agent Zero
+6
-1
@@ -74,6 +74,11 @@ from app.plugins.builtins.mail.models import ( # noqa: F401
|
||||
)
|
||||
from app.plugins.builtins.permissions import PermissionsPlugin # noqa: F401
|
||||
from app.plugins.builtins.permissions.models import Permission, ShareLink # noqa: F401
|
||||
from app.plugins.builtins.report_generator import ReportGeneratorPlugin # noqa: F401
|
||||
from app.plugins.builtins.report_generator.models import ( # noqa: F401
|
||||
ReportInstance,
|
||||
ReportTemplate,
|
||||
)
|
||||
from app.plugins.builtins.tags.models import Tag, TagAssignment # noqa: F401
|
||||
from app.plugins.registry import reset_registry_for_testing # noqa: F401
|
||||
from app.services.plugin_service import reset_plugin_service_for_testing # noqa: F401
|
||||
@@ -135,7 +140,7 @@ def clean_tables(db_setup):
|
||||
# TRUNCATE all tables with CASCADE — fast and reliable isolation
|
||||
conn.execute(
|
||||
text(
|
||||
"TRUNCATE TABLE contact_pgp_keys, pgp_keys, mail_account_send_permissions, mail_account_delegates, mail_seen_by, vacation_sent_log, mail_signatures, mail_templates, mail_rules, mail_label_assignments, mail_labels, mail_attachments, mails, mail_folders, mail_accounts, resource_bookings, resources, subtasks, user_calendar_visibility, calendar_shares, calendar_entry_links, calendar_entries, calendars, files, folders, entity_links, share_links, permissions, tag_assignments, tags, workflow_step_history, workflow_instances, workflows, ai_messages, ai_conversations, plugin_migrations, plugins, contacts, api_tokens, password_reset_tokens, notifications, deletion_log, audit_log, sessions, roles, user_tenants, users, tenants CASCADE;"
|
||||
"TRUNCATE TABLE report_instances, report_templates, contact_pgp_keys, pgp_keys, mail_account_send_permissions, mail_account_delegates, mail_seen_by, vacation_sent_log, mail_signatures, mail_templates, mail_rules, mail_label_assignments, mail_labels, mail_attachments, mails, mail_folders, mail_accounts, resource_bookings, resources, subtasks, user_calendar_visibility, calendar_shares, calendar_entry_links, calendar_entries, calendars, files, folders, entity_links, share_links, permissions, tag_assignments, tags, workflow_step_history, workflow_instances, workflows, ai_messages, ai_conversations, plugin_migrations, plugins, contacts, api_tokens, password_reset_tokens, notifications, deletion_log, audit_log, sessions, roles, user_tenants, users, tenants CASCADE;"
|
||||
)
|
||||
)
|
||||
conn.commit()
|
||||
|
||||
@@ -0,0 +1,257 @@
|
||||
"""Tests for the Report Generator plugin — PDF support, presets, templates, RBAC."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
import shutil
|
||||
|
||||
import pytest
|
||||
import pytest_asyncio
|
||||
from httpx import ASGITransport, AsyncClient
|
||||
from sqlalchemy.ext.asyncio import AsyncEngine, async_sessionmaker, AsyncSession
|
||||
|
||||
from app.core.db import close_engine, reset_engine_for_testing
|
||||
from app.core.service_container import get_container
|
||||
from app.main import create_app
|
||||
from app.plugins.builtins.permissions import PermissionsPlugin
|
||||
from app.plugins.builtins.report_generator import ReportGeneratorPlugin
|
||||
from app.plugins.registry import reset_registry_for_testing
|
||||
from app.services.plugin_service import reset_plugin_service_for_testing
|
||||
|
||||
from tests.conftest import (
|
||||
ORIGIN_HEADER,
|
||||
login_client,
|
||||
seed_tenant_and_users,
|
||||
)
|
||||
|
||||
REPORT_TEST_STORAGE = "/tmp/report_test"
|
||||
|
||||
|
||||
@pytest_asyncio.fixture
|
||||
async def report_app(engine: AsyncEngine, redis_client):
|
||||
"""FastAPI app with Report Generator + Permissions plugins registered, installed, and activated."""
|
||||
os.environ["REPORT_STORAGE_BASE"] = REPORT_TEST_STORAGE
|
||||
reset_engine_for_testing(engine)
|
||||
app = create_app()
|
||||
|
||||
registry = reset_registry_for_testing()
|
||||
registry.initialize(engine, app)
|
||||
|
||||
container = get_container()
|
||||
await container.initialize()
|
||||
|
||||
registry.register_plugin(PermissionsPlugin())
|
||||
registry.register_plugin(ReportGeneratorPlugin())
|
||||
reset_plugin_service_for_testing(registry)
|
||||
|
||||
_sf = async_sessionmaker(bind=engine, expire_on_commit=False, class_=AsyncSession)
|
||||
async with _sf() as session:
|
||||
await registry.install(session, "permissions")
|
||||
await registry.activate(session, "permissions")
|
||||
await registry.install(session, "report_generator")
|
||||
await registry.activate(session, "report_generator")
|
||||
await session.commit()
|
||||
|
||||
yield app
|
||||
await close_engine()
|
||||
if os.path.exists(REPORT_TEST_STORAGE):
|
||||
shutil.rmtree(REPORT_TEST_STORAGE, ignore_errors=True)
|
||||
|
||||
|
||||
@pytest_asyncio.fixture
|
||||
async def report_client(report_app) -> AsyncClient:
|
||||
transport = ASGITransport(app=report_app)
|
||||
async with AsyncClient(transport=transport, base_url="http://test") as c:
|
||||
yield c
|
||||
|
||||
|
||||
@pytest_asyncio.fixture
|
||||
async def report_authed_client(
|
||||
report_client: AsyncClient, db_session: AsyncSession
|
||||
) -> tuple[AsyncClient, dict]:
|
||||
"""Authenticated admin client with seeded data and report generator plugin activated."""
|
||||
seed = await seed_tenant_and_users(db_session)
|
||||
login_resp = await report_client.post(
|
||||
"/api/v1/auth/login",
|
||||
json={"email": "admin@tenanta.com", "password": "TestPass123!"},
|
||||
headers=ORIGIN_HEADER,
|
||||
)
|
||||
assert login_resp.status_code == 200, f"Login failed: {login_resp.text}"
|
||||
csrf_token = login_resp.json().get("csrf_token", "")
|
||||
report_client.headers.update({"X-CSRF-Token": csrf_token})
|
||||
|
||||
return report_client, seed
|
||||
|
||||
|
||||
# ─── Tests ───
|
||||
|
||||
|
||||
class TestReportPresets:
|
||||
"""Test preset report listing and generation."""
|
||||
|
||||
async def test_list_presets(self, report_authed_client):
|
||||
"""GET /presets returns all 5 preset templates."""
|
||||
client, _ = report_authed_client
|
||||
resp = await client.get("/api/v1/reports/presets", headers=ORIGIN_HEADER)
|
||||
assert resp.status_code == 200
|
||||
data = resp.json()
|
||||
assert isinstance(data, list)
|
||||
assert len(data) == 5
|
||||
keys = {item["key"] for item in data}
|
||||
assert keys == {
|
||||
"contact_list",
|
||||
"calendar_week",
|
||||
"calendar_month",
|
||||
"company_list",
|
||||
"audit_log",
|
||||
}
|
||||
# Each preset should have required fields
|
||||
for item in data:
|
||||
assert "name" in item
|
||||
assert "description" in item
|
||||
assert "icon" in item
|
||||
assert "output_formats" in item
|
||||
assert "pdf" in item["output_formats"]
|
||||
|
||||
async def test_generate_preset_pdf(self, report_authed_client):
|
||||
"""POST /presets/generate returns a PDF file for contact_list preset."""
|
||||
client, _ = report_authed_client
|
||||
resp = await client.post(
|
||||
"/api/v1/reports/presets/generate",
|
||||
json={
|
||||
"preset": "contact_list",
|
||||
"output_format": "pdf",
|
||||
"parameters": {
|
||||
"title": "Test Kontaktliste",
|
||||
"contacts": [
|
||||
{"name": "Max Mustermann", "email": "max@test.com", "phone": "+49 123 456789", "type": "customer", "company": "Test GmbH"},
|
||||
{"name": "Anna Schmidt", "email": "anna@test.com", "phone": "+49 987 654321", "type": "supplier", "company": "AG GmbH"},
|
||||
],
|
||||
},
|
||||
},
|
||||
headers=ORIGIN_HEADER,
|
||||
)
|
||||
assert resp.status_code == 200, f"Generate failed: {resp.text[:300]}"
|
||||
assert resp.headers["content-type"].startswith("application/pdf")
|
||||
assert "attachment" in resp.headers.get("content-disposition", "")
|
||||
# Verify it's a valid PDF (starts with %PDF)
|
||||
content = resp.content
|
||||
assert content[:5] == b"%PDF-"
|
||||
assert len(content) > 100 # Should have meaningful content
|
||||
|
||||
async def test_generate_preset_csv(self, report_authed_client):
|
||||
"""POST /presets/generate returns a CSV file for company_list preset."""
|
||||
client, _ = report_authed_client
|
||||
resp = await client.post(
|
||||
"/api/v1/reports/presets/generate",
|
||||
json={
|
||||
"preset": "company_list",
|
||||
"output_format": "csv",
|
||||
"parameters": {
|
||||
"companies": [
|
||||
{"name": "Test GmbH", "address": "Teststr. 1", "zip": "12345", "city": "Berlin", "phone": "+49 123", "email": "info@test.de", "contact_person": "Max"},
|
||||
],
|
||||
},
|
||||
},
|
||||
headers=ORIGIN_HEADER,
|
||||
)
|
||||
assert resp.status_code == 200, f"Generate failed: {resp.text[:300]}"
|
||||
assert resp.headers["content-type"].startswith("text/csv")
|
||||
content = resp.content
|
||||
assert b"Test GmbH" in content
|
||||
assert b"Firmenname" in content
|
||||
|
||||
|
||||
class TestReportTemplates:
|
||||
"""Test template CRUD and report generation with PDF output."""
|
||||
|
||||
async def test_create_and_generate_pdf_template(self, report_authed_client):
|
||||
"""Create a template with output_format=pdf and generate a report (file download)."""
|
||||
client, _ = report_authed_client
|
||||
|
||||
# Create a PDF template
|
||||
create_resp = await client.post(
|
||||
"/api/v1/reports/templates",
|
||||
json={
|
||||
"name": "Custom PDF Report",
|
||||
"description": "A custom HTML template for PDF",
|
||||
"template_type": "jinja2",
|
||||
"content": "<html><body><h1>{{ title }}</h1><p>{{ message }}</p></body></html>",
|
||||
"output_format": "pdf",
|
||||
},
|
||||
headers=ORIGIN_HEADER,
|
||||
)
|
||||
assert create_resp.status_code == 201, f"Create failed: {create_resp.text}"
|
||||
template = create_resp.json()
|
||||
assert template["output_format"] == "pdf"
|
||||
template_id = template["id"]
|
||||
|
||||
# Generate a report from this template — returns file directly
|
||||
gen_resp = await client.post(
|
||||
"/api/v1/reports/generate",
|
||||
json={
|
||||
"template_id": template_id,
|
||||
"data": {"title": "Hello World", "message": "This is a test PDF."},
|
||||
},
|
||||
headers=ORIGIN_HEADER,
|
||||
)
|
||||
assert gen_resp.status_code == 200, f"Generate failed: {gen_resp.text[:300]}"
|
||||
assert gen_resp.headers["content-type"].startswith("application/pdf")
|
||||
assert gen_resp.content[:5] == b"%PDF-"
|
||||
|
||||
async def test_output_format_validation(self, report_authed_client):
|
||||
"""Template creation rejects invalid output_format values."""
|
||||
client, _ = report_authed_client
|
||||
resp = await client.post(
|
||||
"/api/v1/reports/templates",
|
||||
json={
|
||||
"name": "Bad Format",
|
||||
"content": "test",
|
||||
"output_format": "invalid_format",
|
||||
},
|
||||
headers=ORIGIN_HEADER,
|
||||
)
|
||||
assert resp.status_code == 422 # Validation error
|
||||
|
||||
|
||||
class TestReportRBAC:
|
||||
"""Test RBAC enforcement on report endpoints."""
|
||||
|
||||
async def test_unauthenticated_access_blocked(self, report_client):
|
||||
"""Unauthenticated requests to reports endpoints are rejected."""
|
||||
resp = await report_client.get("/api/v1/reports/presets", headers=ORIGIN_HEADER)
|
||||
assert resp.status_code == 401
|
||||
|
||||
async def test_viewer_cannot_manage_templates(self, report_authed_client, db_session):
|
||||
"""Viewer role cannot create templates (requires manage_templates) and cannot generate (requires generate)."""
|
||||
client, seed = report_authed_client
|
||||
|
||||
# Login as viewer
|
||||
viewer_resp = await client.post(
|
||||
"/api/v1/auth/login",
|
||||
json={"email": "viewer@tenanta.com", "password": "TestPass123!"},
|
||||
headers=ORIGIN_HEADER,
|
||||
)
|
||||
assert viewer_resp.status_code == 200
|
||||
csrf_token = viewer_resp.json().get("csrf_token", "")
|
||||
client.headers.update({"X-CSRF-Token": csrf_token})
|
||||
|
||||
# Viewer does not have reports:read in legacy permissions → 403 on presets
|
||||
resp = await client.get("/api/v1/reports/presets", headers=ORIGIN_HEADER)
|
||||
assert resp.status_code == 403
|
||||
|
||||
# Viewer should NOT be able to create templates
|
||||
resp = await client.post(
|
||||
"/api/v1/reports/templates",
|
||||
json={"name": "Forbidden", "content": "test", "output_format": "pdf"},
|
||||
headers=ORIGIN_HEADER,
|
||||
)
|
||||
assert resp.status_code == 403
|
||||
|
||||
# Viewer should NOT be able to generate preset reports
|
||||
resp = await client.post(
|
||||
"/api/v1/reports/presets/generate",
|
||||
json={"preset": "contact_list", "output_format": "pdf", "parameters": {}},
|
||||
headers=ORIGIN_HEADER,
|
||||
)
|
||||
assert resp.status_code == 403
|
||||
Reference in New Issue
Block a user