leocrm/tests/test_report_generator.py

"""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