760 lines
27 KiB
Python
760 lines
27 KiB
Python
|
"""Documents Generator tests (Phase L1-L3) — letterheads, print templates,
|
||
|
|
block registry, module contributions, preview/render, RBAC, isolation.
|
||
|
|
|
||
|
|
Contract-Muster (like importexport_entities): plugins contribute
|
||
|
|
``document_blocks()``, ``document_placeholders(entity_type)`` and
|
||
|
|
``document_data(db, tenant_id, entity_id, entity_type)``. contacts is the
|
||
|
|
first contributor (W4a-style contribution, #359 philosophy).
|
||
|
|
"""
|
||
|
|
|
||
|
|
from __future__ import annotations
|
||
|
|
|
||
|
|
import uuid as uuid_mod
|
||
|
|
from collections.abc import AsyncGenerator
|
||
|
|
|
||
|
|
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.permission_registry import init_permission_registry
|
||
|
|
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,
|
||
|
|
)
|
||
|
|
|
||
|
|
DOC_BASE = "/api/v1/reports"
|
||
|
|
|
||
|
|
|
||
|
|
@pytest_asyncio.fixture
|
||
|
|
async def docs_app(engine: AsyncEngine, redis_client):
|
||
|
|
"""App with permissions + report_generator installed & activated."""
|
||
|
|
reset_engine_for_testing(engine)
|
||
|
|
app = create_app()
|
||
|
|
|
||
|
|
registry = reset_registry_for_testing()
|
||
|
|
registry.initialize(engine, app)
|
||
|
|
init_permission_registry(active_plugin_names={"permissions", "report_generator"})
|
||
|
|
|
||
|
|
container = get_container()
|
||
|
|
await container.initialize()
|
||
|
|
|
||
|
|
registry.register_plugin(PermissionsPlugin())
|
||
|
|
registry.register_plugin(ReportGeneratorPlugin())
|
||
|
|
# contacts contributes document placeholders/data via its contract —
|
||
|
|
# register (not install) so list_discovered() includes it and the
|
||
|
|
# contract registry can lazy-load contacts.contracts.
|
||
|
|
from app.plugins.builtins.contacts.plugin import ContactsPlugin
|
||
|
|
|
||
|
|
registry.register_plugin(ContactsPlugin())
|
||
|
|
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()
|
||
|
|
|
||
|
|
|
||
|
|
@pytest_asyncio.fixture
|
||
|
|
async def docs_client(docs_app) -> AsyncGenerator[AsyncClient, None]:
|
||
|
|
transport = ASGITransport(app=docs_app)
|
||
|
|
async with AsyncClient(transport=transport, base_url="http://test") as c:
|
||
|
|
yield c
|
||
|
|
|
||
|
|
|
||
|
|
def _letterhead_payload(**overrides) -> dict:
|
||
|
|
payload = {
|
||
|
|
"name": "Standard Briefpapier",
|
||
|
|
"description": "Firmenbriefkopf mit Logo",
|
||
|
|
"config": {
|
||
|
|
"page": {
|
||
|
|
"size": "A4",
|
||
|
|
"orientation": "portrait",
|
||
|
|
"margins": {"top": 25, "right": 20, "bottom": 25, "left": 20},
|
||
|
|
},
|
||
|
|
"header": {
|
||
|
|
"enabled": True,
|
||
|
|
"blocks": [
|
||
|
|
{
|
||
|
|
"id": "h1",
|
||
|
|
"type": "text",
|
||
|
|
"config": {"content": "Meine Firma GmbH"},
|
||
|
|
}
|
||
|
|
],
|
||
|
|
},
|
||
|
|
"footer": {"enabled": True, "blocks": []},
|
||
|
|
"watermark": {"enabled": False},
|
||
|
|
},
|
||
|
|
"is_default": False,
|
||
|
|
}
|
||
|
|
payload.update(overrides)
|
||
|
|
return payload
|
||
|
|
|
||
|
|
|
||
|
|
def _template_payload(letterhead_id: str | None = None, **overrides) -> dict:
|
||
|
|
payload = {
|
||
|
|
"name": "Angebot",
|
||
|
|
"description": "Angebot mit Briefpapier",
|
||
|
|
"letterhead_id": letterhead_id,
|
||
|
|
"entity_type": "company",
|
||
|
|
"blocks": [
|
||
|
|
{
|
||
|
|
"id": "b1",
|
||
|
|
"type": "text",
|
||
|
|
"config": {"content": "Sehr geehrte Damen und Herren,\n\nFirma: {{name}}"},
|
||
|
|
},
|
||
|
|
{
|
||
|
|
"id": "b2",
|
||
|
|
"type": "shape",
|
||
|
|
"config": {"shape": "line", "width": "100%", "height": 1, "color": "#111827"},
|
||
|
|
},
|
||
|
|
],
|
||
|
|
"output_format": "pdf",
|
||
|
|
}
|
||
|
|
payload.update(overrides)
|
||
|
|
return payload
|
||
|
|
|
||
|
|
|
||
|
|
# ─── Letterhead CRUD ─────────────────────────────────────────────────────────
|
||
|
|
|
||
|
|
|
||
|
|
@pytest.mark.asyncio
|
||
|
|
class TestLetterheadCRUD:
|
||
|
|
async def test_create_letterhead_201(
|
||
|
|
self, docs_client: AsyncClient, db_session
|
||
|
|
):
|
||
|
|
await seed_tenant_and_users(db_session)
|
||
|
|
await login_client(docs_client, "admin@tenanta.com")
|
||
|
|
resp = await docs_client.post(
|
||
|
|
f"{DOC_BASE}/letterheads",
|
||
|
|
json=_letterhead_payload(),
|
||
|
|
headers=ORIGIN_HEADER,
|
||
|
|
)
|
||
|
|
assert resp.status_code == 201, f"{resp.status_code} {resp.text}"
|
||
|
|
body = resp.json()
|
||
|
|
assert body["name"] == "Standard Briefpapier"
|
||
|
|
assert body["config"]["header"]["enabled"] is True
|
||
|
|
assert body["id"]
|
||
|
|
|
||
|
|
async def test_create_letterhead_empty_name_422(
|
||
|
|
self, docs_client: AsyncClient, db_session
|
||
|
|
):
|
||
|
|
await seed_tenant_and_users(db_session)
|
||
|
|
await login_client(docs_client, "admin@tenanta.com")
|
||
|
|
resp = await docs_client.post(
|
||
|
|
f"{DOC_BASE}/letterheads",
|
||
|
|
json=_letterhead_payload(name=""),
|
||
|
|
headers=ORIGIN_HEADER,
|
||
|
|
)
|
||
|
|
assert resp.status_code == 422
|
||
|
|
|
||
|
|
async def test_list_letterheads_tenant_scoped(
|
||
|
|
self, docs_client: AsyncClient, db_session
|
||
|
|
):
|
||
|
|
await seed_tenant_and_users(db_session)
|
||
|
|
await login_client(docs_client, "admin@tenanta.com")
|
||
|
|
resp = await docs_client.post(
|
||
|
|
f"{DOC_BASE}/letterheads",
|
||
|
|
json=_letterhead_payload(),
|
||
|
|
headers=ORIGIN_HEADER,
|
||
|
|
)
|
||
|
|
assert resp.status_code == 201
|
||
|
|
lh_a = resp.json()["id"]
|
||
|
|
|
||
|
|
# Tenant B admin creates their own letterhead
|
||
|
|
await login_client(docs_client, "admin@tenantb.com")
|
||
|
|
resp = await docs_client.post(
|
||
|
|
f"{DOC_BASE}/letterheads",
|
||
|
|
json=_letterhead_payload(name="Tenant B Kopf"),
|
||
|
|
headers=ORIGIN_HEADER,
|
||
|
|
)
|
||
|
|
assert resp.status_code == 201
|
||
|
|
|
||
|
|
resp = await docs_client.get(f"{DOC_BASE}/letterheads", headers=ORIGIN_HEADER)
|
||
|
|
assert resp.status_code == 200
|
||
|
|
items = resp.json()["items"]
|
||
|
|
ids = [i["id"] for i in items]
|
||
|
|
assert lh_a not in ids # cross-tenant isolation
|
||
|
|
assert len(items) == 1
|
||
|
|
assert items[0]["name"] == "Tenant B Kopf"
|
||
|
|
|
||
|
|
async def test_get_letterhead_404(self, docs_client: AsyncClient, db_session):
|
||
|
|
await seed_tenant_and_users(db_session)
|
||
|
|
await login_client(docs_client, "admin@tenanta.com")
|
||
|
|
resp = await docs_client.get(
|
||
|
|
f"{DOC_BASE}/letterheads/{uuid_mod.uuid4()}", headers=ORIGIN_HEADER
|
||
|
|
)
|
||
|
|
assert resp.status_code == 404
|
||
|
|
|
||
|
|
async def test_update_letterhead(
|
||
|
|
self, docs_client: AsyncClient, db_session
|
||
|
|
):
|
||
|
|
await seed_tenant_and_users(db_session)
|
||
|
|
await login_client(docs_client, "admin@tenanta.com")
|
||
|
|
resp = await docs_client.post(
|
||
|
|
f"{DOC_BASE}/letterheads",
|
||
|
|
json=_letterhead_payload(),
|
||
|
|
headers=ORIGIN_HEADER,
|
||
|
|
)
|
||
|
|
lh_id = resp.json()["id"]
|
||
|
|
resp = await docs_client.put(
|
||
|
|
f"{DOC_BASE}/letterheads/{lh_id}",
|
||
|
|
json={"name": "Neuer Name"},
|
||
|
|
headers=ORIGIN_HEADER,
|
||
|
|
)
|
||
|
|
assert resp.status_code == 200
|
||
|
|
assert resp.json()["name"] == "Neuer Name"
|
||
|
|
|
||
|
|
async def test_delete_letterhead_soft(
|
||
|
|
self, docs_client: AsyncClient, db_session
|
||
|
|
):
|
||
|
|
await seed_tenant_and_users(db_session)
|
||
|
|
await login_client(docs_client, "admin@tenanta.com")
|
||
|
|
resp = await docs_client.post(
|
||
|
|
f"{DOC_BASE}/letterheads",
|
||
|
|
json=_letterhead_payload(),
|
||
|
|
headers=ORIGIN_HEADER,
|
||
|
|
)
|
||
|
|
lh_id = resp.json()["id"]
|
||
|
|
resp = await docs_client.delete(
|
||
|
|
f"{DOC_BASE}/letterheads/{lh_id}", headers=ORIGIN_HEADER
|
||
|
|
)
|
||
|
|
assert resp.status_code == 204
|
||
|
|
resp = await docs_client.get(
|
||
|
|
f"{DOC_BASE}/letterheads/{lh_id}", headers=ORIGIN_HEADER
|
||
|
|
)
|
||
|
|
assert resp.status_code == 404
|
||
|
|
|
||
|
|
async def test_viewer_cannot_manage_403(
|
||
|
|
self, docs_client: AsyncClient, db_session
|
||
|
|
):
|
||
|
|
await seed_tenant_and_users(db_session)
|
||
|
|
await login_client(docs_client, "viewer@tenanta.com")
|
||
|
|
resp = await docs_client.post(
|
||
|
|
f"{DOC_BASE}/letterheads",
|
||
|
|
json=_letterhead_payload(),
|
||
|
|
headers=ORIGIN_HEADER,
|
||
|
|
)
|
||
|
|
assert resp.status_code == 403
|
||
|
|
|
||
|
|
|
||
|
|
# ─── Print Template CRUD ─────────────────────────────────────────────────────
|
||
|
|
|
||
|
|
|
||
|
|
@pytest.mark.asyncio
|
||
|
|
class TestPrintTemplateCRUD:
|
||
|
|
async def test_create_template_201(
|
||
|
|
self, docs_client: AsyncClient, db_session
|
||
|
|
):
|
||
|
|
await seed_tenant_and_users(db_session)
|
||
|
|
await login_client(docs_client, "admin@tenanta.com")
|
||
|
|
resp = await docs_client.post(
|
||
|
|
f"{DOC_BASE}/print-templates",
|
||
|
|
json=_template_payload(),
|
||
|
|
headers=ORIGIN_HEADER,
|
||
|
|
)
|
||
|
|
assert resp.status_code == 201, f"{resp.status_code} {resp.text}"
|
||
|
|
body = resp.json()
|
||
|
|
assert body["name"] == "Angebot"
|
||
|
|
assert len(body["blocks"]) == 2
|
||
|
|
assert body["entity_type"] == "company"
|
||
|
|
|
||
|
|
async def test_create_template_with_letterhead_ref(
|
||
|
|
self, docs_client: AsyncClient, db_session
|
||
|
|
):
|
||
|
|
await seed_tenant_and_users(db_session)
|
||
|
|
await login_client(docs_client, "admin@tenanta.com")
|
||
|
|
resp = await docs_client.post(
|
||
|
|
f"{DOC_BASE}/letterheads",
|
||
|
|
json=_letterhead_payload(),
|
||
|
|
headers=ORIGIN_HEADER,
|
||
|
|
)
|
||
|
|
lh_id = resp.json()["id"]
|
||
|
|
resp = await docs_client.post(
|
||
|
|
f"{DOC_BASE}/print-templates",
|
||
|
|
json=_template_payload(letterhead_id=lh_id),
|
||
|
|
headers=ORIGIN_HEADER,
|
||
|
|
)
|
||
|
|
assert resp.status_code == 201
|
||
|
|
assert resp.json()["letterhead_id"] == lh_id
|
||
|
|
|
||
|
|
async def test_create_template_invalid_block_422(
|
||
|
|
self, docs_client: AsyncClient, db_session
|
||
|
|
):
|
||
|
|
await seed_tenant_and_users(db_session)
|
||
|
|
await login_client(docs_client, "admin@tenanta.com")
|
||
|
|
resp = await docs_client.post(
|
||
|
|
f"{DOC_BASE}/print-templates",
|
||
|
|
json=_template_payload(blocks=[{"id": "x", "type": "not_a_block"}]),
|
||
|
|
headers=ORIGIN_HEADER,
|
||
|
|
)
|
||
|
|
assert resp.status_code == 422
|
||
|
|
assert "invalid_block" in resp.text
|
||
|
|
|
||
|
|
async def test_create_template_text_block_requires_content_422(
|
||
|
|
self, docs_client: AsyncClient, db_session
|
||
|
|
):
|
||
|
|
await seed_tenant_and_users(db_session)
|
||
|
|
await login_client(docs_client, "admin@tenanta.com")
|
||
|
|
resp = await docs_client.post(
|
||
|
|
f"{DOC_BASE}/print-templates",
|
||
|
|
json=_template_payload(
|
||
|
|
blocks=[{"id": "x", "type": "text", "config": {}}]
|
||
|
|
),
|
||
|
|
headers=ORIGIN_HEADER,
|
||
|
|
)
|
||
|
|
assert resp.status_code == 422
|
||
|
|
|
||
|
|
async def test_create_template_bad_shape_422(
|
||
|
|
self, docs_client: AsyncClient, db_session
|
||
|
|
):
|
||
|
|
await seed_tenant_and_users(db_session)
|
||
|
|
await login_client(docs_client, "admin@tenanta.com")
|
||
|
|
resp = await docs_client.post(
|
||
|
|
f"{DOC_BASE}/print-templates",
|
||
|
|
json=_template_payload(
|
||
|
|
blocks=[
|
||
|
|
{
|
||
|
|
"id": "x",
|
||
|
|
"type": "shape",
|
||
|
|
"config": {"shape": "triangle"},
|
||
|
|
}
|
||
|
|
]
|
||
|
|
),
|
||
|
|
headers=ORIGIN_HEADER,
|
||
|
|
)
|
||
|
|
assert resp.status_code == 422
|
||
|
|
|
||
|
|
async def test_list_and_update_and_delete_template(
|
||
|
|
self, docs_client: AsyncClient, db_session
|
||
|
|
):
|
||
|
|
await seed_tenant_and_users(db_session)
|
||
|
|
await login_client(docs_client, "admin@tenanta.com")
|
||
|
|
resp = await docs_client.post(
|
||
|
|
f"{DOC_BASE}/print-templates",
|
||
|
|
json=_template_payload(),
|
||
|
|
headers=ORIGIN_HEADER,
|
||
|
|
)
|
||
|
|
tid = resp.json()["id"]
|
||
|
|
|
||
|
|
resp = await docs_client.get(f"{DOC_BASE}/print-templates", headers=ORIGIN_HEADER)
|
||
|
|
assert resp.status_code == 200
|
||
|
|
assert any(t["id"] == tid for t in resp.json()["items"])
|
||
|
|
|
||
|
|
resp = await docs_client.put(
|
||
|
|
f"{DOC_BASE}/print-templates/{tid}",
|
||
|
|
json={"name": "Angebot v2"},
|
||
|
|
headers=ORIGIN_HEADER,
|
||
|
|
)
|
||
|
|
assert resp.status_code == 200
|
||
|
|
assert resp.json()["name"] == "Angebot v2"
|
||
|
|
|
||
|
|
resp = await docs_client.delete(
|
||
|
|
f"{DOC_BASE}/print-templates/{tid}", headers=ORIGIN_HEADER
|
||
|
|
)
|
||
|
|
assert resp.status_code == 204
|
||
|
|
resp = await docs_client.get(
|
||
|
|
f"{DOC_BASE}/print-templates/{tid}", headers=ORIGIN_HEADER
|
||
|
|
)
|
||
|
|
assert resp.status_code == 404
|
||
|
|
|
||
|
|
|
||
|
|
# ─── Block Registry + Module Contributions ──────────────────────────────────
|
||
|
|
|
||
|
|
|
||
|
|
@pytest.mark.asyncio
|
||
|
|
class TestBlockRegistry:
|
||
|
|
async def test_document_blocks_contains_builtin_types(
|
||
|
|
self, docs_client: AsyncClient, db_session
|
||
|
|
):
|
||
|
|
await seed_tenant_and_users(db_session)
|
||
|
|
await login_client(docs_client, "admin@tenanta.com")
|
||
|
|
resp = await docs_client.get(f"{DOC_BASE}/document-blocks", headers=ORIGIN_HEADER)
|
||
|
|
assert resp.status_code == 200
|
||
|
|
types = {b["type"] for b in resp.json()}
|
||
|
|
assert {
|
||
|
|
"text",
|
||
|
|
"image",
|
||
|
|
"shape",
|
||
|
|
"table",
|
||
|
|
"spacer",
|
||
|
|
"divider",
|
||
|
|
"placeholder",
|
||
|
|
} <= types
|
||
|
|
# builtin entries carry a label for the palette UI
|
||
|
|
assert all(b.get("label") for b in resp.json())
|
||
|
|
|
||
|
|
async def test_document_placeholders_contains_contacts_contribution(
|
||
|
|
self, docs_client: AsyncClient, db_session
|
||
|
|
):
|
||
|
|
"""contacts plugin contributes placeholders via contract (L1)."""
|
||
|
|
await seed_tenant_and_users(db_session)
|
||
|
|
await login_client(docs_client, "admin@tenanta.com")
|
||
|
|
resp = await docs_client.get(
|
||
|
|
f"{DOC_BASE}/document-placeholders", headers=ORIGIN_HEADER
|
||
|
|
)
|
||
|
|
assert resp.status_code == 200
|
||
|
|
data = resp.json()
|
||
|
|
assert "contact" in data, f"contact placeholders missing: {list(data)}"
|
||
|
|
keys = [p["key"] for p in data["contact"]]
|
||
|
|
assert "firstname" in keys
|
||
|
|
assert "displayname" in keys
|
||
|
|
# every placeholder has label + example for the editor UI
|
||
|
|
assert all(p.get("label") and p.get("example") is not None for p in data["contact"])
|
||
|
|
|
||
|
|
|
||
|
|
# ─── Preview (HTML) ─────────────────────────────────────────────────────────
|
||
|
|
|
||
|
|
|
||
|
|
@pytest.mark.asyncio
|
||
|
|
class TestPreview:
|
||
|
|
async def test_preview_renders_blocks_with_data(
|
||
|
|
self, docs_client: AsyncClient, db_session
|
||
|
|
):
|
||
|
|
await seed_tenant_and_users(db_session)
|
||
|
|
await login_client(docs_client, "admin@tenanta.com")
|
||
|
|
resp = await docs_client.post(
|
||
|
|
f"{DOC_BASE}/documents/preview",
|
||
|
|
json={
|
||
|
|
"blocks": [
|
||
|
|
{
|
||
|
|
"id": "b1",
|
||
|
|
"type": "text",
|
||
|
|
"config": {"content": "Hallo {{displayname}}!"},
|
||
|
|
}
|
||
|
|
],
|
||
|
|
"letterhead_config": None,
|
||
|
|
"data": {"displayname": "Test AG"},
|
||
|
|
},
|
||
|
|
headers=ORIGIN_HEADER,
|
||
|
|
)
|
||
|
|
assert resp.status_code == 200, f"{resp.status_code} {resp.text}"
|
||
|
|
html = resp.json()["html"]
|
||
|
|
assert "Hallo Test AG!" in html
|
||
|
|
|
||
|
|
async def test_preview_with_letterhead_header(
|
||
|
|
self, docs_client: AsyncClient, db_session
|
||
|
|
):
|
||
|
|
await seed_tenant_and_users(db_session)
|
||
|
|
await login_client(docs_client, "admin@tenanta.com")
|
||
|
|
resp = await docs_client.post(
|
||
|
|
f"{DOC_BASE}/documents/preview",
|
||
|
|
json={
|
||
|
|
"blocks": [
|
||
|
|
{
|
||
|
|
"id": "b1",
|
||
|
|
"type": "text",
|
||
|
|
"config": {"content": "Inhalt"},
|
||
|
|
}
|
||
|
|
],
|
||
|
|
"letterhead_config": _letterhead_payload()["config"],
|
||
|
|
"data": {},
|
||
|
|
},
|
||
|
|
headers=ORIGIN_HEADER,
|
||
|
|
)
|
||
|
|
assert resp.status_code == 200
|
||
|
|
html = resp.json()["html"]
|
||
|
|
assert "Meine Firma GmbH" in html
|
||
|
|
assert "@page" in html or "running(" in html
|
||
|
|
|
||
|
|
async def test_preview_placeholder_defaults_without_data(
|
||
|
|
self, docs_client: AsyncClient, db_session
|
||
|
|
):
|
||
|
|
"""Missing data keys fall back to placeholder example values."""
|
||
|
|
await seed_tenant_and_users(db_session)
|
||
|
|
await login_client(docs_client, "admin@tenanta.com")
|
||
|
|
resp = await docs_client.post(
|
||
|
|
f"{DOC_BASE}/documents/preview",
|
||
|
|
json={
|
||
|
|
"blocks": [
|
||
|
|
{
|
||
|
|
"id": "b1",
|
||
|
|
"type": "text",
|
||
|
|
"config": {"content": "{{firstname}}"},
|
||
|
|
}
|
||
|
|
],
|
||
|
|
"letterhead_config": None,
|
||
|
|
"entity_type": "contact",
|
||
|
|
"data": None,
|
||
|
|
},
|
||
|
|
headers=ORIGIN_HEADER,
|
||
|
|
)
|
||
|
|
assert resp.status_code == 200
|
||
|
|
# must render the example value, not crash with StrictUndefined
|
||
|
|
assert "{{firstname}}" not in resp.json()["html"]
|
||
|
|
|
||
|
|
async def test_preview_invalid_block_422(
|
||
|
|
self, docs_client: AsyncClient, db_session
|
||
|
|
):
|
||
|
|
await seed_tenant_and_users(db_session)
|
||
|
|
await login_client(docs_client, "admin@tenanta.com")
|
||
|
|
resp = await docs_client.post(
|
||
|
|
f"{DOC_BASE}/documents/preview",
|
||
|
|
json={
|
||
|
|
"blocks": [{"id": "x", "type": "nope"}],
|
||
|
|
"letterhead_config": None,
|
||
|
|
},
|
||
|
|
headers=ORIGIN_HEADER,
|
||
|
|
)
|
||
|
|
assert resp.status_code == 422
|
||
|
|
|
||
|
|
|
||
|
|
# ─── Render (PDF) ────────────────────────────────────────────────────────────
|
||
|
|
|
||
|
|
|
||
|
|
@pytest.mark.asyncio
|
||
|
|
class TestRender:
|
||
|
|
async def _create_template(self, client: AsyncClient, blocks=None) -> str:
|
||
|
|
payload = _template_payload()
|
||
|
|
if blocks is not None:
|
||
|
|
payload["blocks"] = blocks
|
||
|
|
resp = await client.post(
|
||
|
|
f"{DOC_BASE}/print-templates",
|
||
|
|
json=payload,
|
||
|
|
headers=ORIGIN_HEADER,
|
||
|
|
)
|
||
|
|
assert resp.status_code == 201, f"{resp.status_code} {resp.text}"
|
||
|
|
return resp.json()["id"]
|
||
|
|
|
||
|
|
async def test_render_pdf_bytes(
|
||
|
|
self, docs_client: AsyncClient, db_session
|
||
|
|
):
|
||
|
|
seed = await seed_tenant_and_users(db_session)
|
||
|
|
await login_client(docs_client, "admin@tenanta.com")
|
||
|
|
tid = await self._create_template(docs_client)
|
||
|
|
resp = await docs_client.post(
|
||
|
|
f"{DOC_BASE}/print-templates/{tid}/render",
|
||
|
|
json={
|
||
|
|
"entity_type": "company",
|
||
|
|
"entity_id": str(seed["company_a"].id),
|
||
|
|
},
|
||
|
|
headers=ORIGIN_HEADER,
|
||
|
|
)
|
||
|
|
assert resp.status_code == 200, f"{resp.status_code} {resp.text}"
|
||
|
|
assert resp.headers["content-type"].startswith("application/pdf")
|
||
|
|
assert resp.content.startswith(b"%PDF")
|
||
|
|
|
||
|
|
async def test_render_with_letterhead_pdf(
|
||
|
|
self, docs_client: AsyncClient, db_session
|
||
|
|
):
|
||
|
|
seed = await seed_tenant_and_users(db_session)
|
||
|
|
await login_client(docs_client, "admin@tenanta.com")
|
||
|
|
resp = await docs_client.post(
|
||
|
|
f"{DOC_BASE}/letterheads",
|
||
|
|
json=_letterhead_payload(),
|
||
|
|
headers=ORIGIN_HEADER,
|
||
|
|
)
|
||
|
|
lh_id = resp.json()["id"]
|
||
|
|
resp = await docs_client.post(
|
||
|
|
f"{DOC_BASE}/print-templates",
|
||
|
|
json=_template_payload(letterhead_id=lh_id),
|
||
|
|
headers=ORIGIN_HEADER,
|
||
|
|
)
|
||
|
|
tid = resp.json()["id"]
|
||
|
|
resp = await docs_client.post(
|
||
|
|
f"{DOC_BASE}/print-templates/{tid}/render",
|
||
|
|
json={
|
||
|
|
"entity_type": "company",
|
||
|
|
"entity_id": str(seed["company_a"].id),
|
||
|
|
},
|
||
|
|
headers=ORIGIN_HEADER,
|
||
|
|
)
|
||
|
|
assert resp.status_code == 200
|
||
|
|
assert resp.content.startswith(b"%PDF")
|
||
|
|
|
||
|
|
async def test_render_unknown_entity_404(
|
||
|
|
self, docs_client: AsyncClient, db_session
|
||
|
|
):
|
||
|
|
await seed_tenant_and_users(db_session)
|
||
|
|
await login_client(docs_client, "admin@tenanta.com")
|
||
|
|
tid = await self._create_template(docs_client)
|
||
|
|
resp = await docs_client.post(
|
||
|
|
f"{DOC_BASE}/print-templates/{tid}/render",
|
||
|
|
json={"entity_type": "company", "entity_id": str(uuid_mod.uuid4())},
|
||
|
|
headers=ORIGIN_HEADER,
|
||
|
|
)
|
||
|
|
assert resp.status_code == 404
|
||
|
|
|
||
|
|
async def test_render_template_not_found_404(
|
||
|
|
self, docs_client: AsyncClient, db_session
|
||
|
|
):
|
||
|
|
await seed_tenant_and_users(db_session)
|
||
|
|
await login_client(docs_client, "admin@tenanta.com")
|
||
|
|
resp = await docs_client.post(
|
||
|
|
f"{DOC_BASE}/print-templates/{uuid_mod.uuid4()}/render",
|
||
|
|
json={"entity_type": "company", "entity_id": str(uuid_mod.uuid4())},
|
||
|
|
headers=ORIGIN_HEADER,
|
||
|
|
)
|
||
|
|
assert resp.status_code == 404
|
||
|
|
|
||
|
|
async def test_render_viewer_forbidden_403(
|
||
|
|
self, docs_client: AsyncClient, db_session
|
||
|
|
):
|
||
|
|
seed = await seed_tenant_and_users(db_session)
|
||
|
|
await login_client(docs_client, "admin@tenanta.com")
|
||
|
|
tid = await self._create_template(docs_client)
|
||
|
|
await login_client(docs_client, "viewer@tenanta.com")
|
||
|
|
resp = await docs_client.post(
|
||
|
|
f"{DOC_BASE}/print-templates/{tid}/render",
|
||
|
|
json={
|
||
|
|
"entity_type": "company",
|
||
|
|
"entity_id": str(seed["company_a"].id),
|
||
|
|
},
|
||
|
|
headers=ORIGIN_HEADER,
|
||
|
|
)
|
||
|
|
assert resp.status_code == 403
|
||
|
|
|
||
|
|
async def test_render_image_block_missing_asset_no_crash(
|
||
|
|
self, docs_client: AsyncClient, db_session
|
||
|
|
):
|
||
|
|
"""image block referencing a non-existent asset renders empty."""
|
||
|
|
seed = await seed_tenant_and_users(db_session)
|
||
|
|
await login_client(docs_client, "admin@tenanta.com")
|
||
|
|
tid = await self._create_template(
|
||
|
|
docs_client,
|
||
|
|
blocks=[
|
||
|
|
{
|
||
|
|
"id": "img1",
|
||
|
|
"type": "image",
|
||
|
|
"config": {"asset_id": str(uuid_mod.uuid4()), "width": 100},
|
||
|
|
},
|
||
|
|
{
|
||
|
|
"id": "t1",
|
||
|
|
"type": "text",
|
||
|
|
"config": {"content": "{{name}}"},
|
||
|
|
},
|
||
|
|
],
|
||
|
|
)
|
||
|
|
resp = await docs_client.post(
|
||
|
|
f"{DOC_BASE}/print-templates/{tid}/render",
|
||
|
|
json={
|
||
|
|
"entity_type": "company",
|
||
|
|
"entity_id": str(seed["company_a"].id),
|
||
|
|
},
|
||
|
|
headers=ORIGIN_HEADER,
|
||
|
|
)
|
||
|
|
assert resp.status_code == 200
|
||
|
|
assert resp.content.startswith(b"%PDF")
|
||
|
|
|
||
|
|
|
||
|
|
# ─── Assets (Logo/Bild-Upload) ───────────────────────────────────────────────
|
||
|
|
|
||
|
|
|
||
|
|
@pytest.mark.asyncio
|
||
|
|
class TestAssets:
|
||
|
|
async def test_upload_logo_asset_201(
|
||
|
|
self, docs_client: AsyncClient, db_session
|
||
|
|
):
|
||
|
|
await seed_tenant_and_users(db_session)
|
||
|
|
await login_client(docs_client, "admin@tenanta.com")
|
||
|
|
resp = await docs_client.post(
|
||
|
|
f"{DOC_BASE}/letterheads",
|
||
|
|
json=_letterhead_payload(),
|
||
|
|
headers=ORIGIN_HEADER,
|
||
|
|
)
|
||
|
|
lh_id = resp.json()["id"]
|
||
|
|
|
||
|
|
# minimal valid PNG (1x1 transparent)
|
||
|
|
png_bytes = bytes.fromhex(
|
||
|
|
"89504e470d0a1a0a0000000d49484452000000010000000108060000001f15c489"
|
||
|
|
"0000000d49444154789c626001000000ffff03000006000557bfabd40000000049454e44ae426082"
|
||
|
|
)
|
||
|
|
resp = await docs_client.post(
|
||
|
|
f"{DOC_BASE}/letterheads/{lh_id}/assets",
|
||
|
|
files={"file": ("logo.png", png_bytes, "image/png")},
|
||
|
|
headers=ORIGIN_HEADER,
|
||
|
|
)
|
||
|
|
assert resp.status_code == 201, f"{resp.status_code} {resp.text}"
|
||
|
|
body = resp.json()
|
||
|
|
assert body["filename"] == "logo.png"
|
||
|
|
assert body["mime_type"] == "image/png"
|
||
|
|
assert body["id"]
|
||
|
|
|
||
|
|
async def test_upload_non_image_422(
|
||
|
|
self, docs_client: AsyncClient, db_session
|
||
|
|
):
|
||
|
|
await seed_tenant_and_users(db_session)
|
||
|
|
await login_client(docs_client, "admin@tenanta.com")
|
||
|
|
resp = await docs_client.post(
|
||
|
|
f"{DOC_BASE}/letterheads",
|
||
|
|
json=_letterhead_payload(),
|
||
|
|
headers=ORIGIN_HEADER,
|
||
|
|
)
|
||
|
|
lh_id = resp.json()["id"]
|
||
|
|
resp = await docs_client.post(
|
||
|
|
f"{DOC_BASE}/letterheads/{lh_id}/assets",
|
||
|
|
files={"file": ("notes.txt", b"not an image", "text/plain")},
|
||
|
|
headers=ORIGIN_HEADER,
|
||
|
|
)
|
||
|
|
assert resp.status_code == 422
|
||
|
|
|
||
|
|
async def test_upload_to_foreign_letterhead_404(
|
||
|
|
self, docs_client: AsyncClient, db_session
|
||
|
|
):
|
||
|
|
await seed_tenant_and_users(db_session)
|
||
|
|
await login_client(docs_client, "admin@tenanta.com")
|
||
|
|
resp = await docs_client.post(
|
||
|
|
f"{DOC_BASE}/letterheads/{uuid_mod.uuid4()}/assets",
|
||
|
|
files={"file": ("x.png", b"x", "image/png")},
|
||
|
|
headers=ORIGIN_HEADER,
|
||
|
|
)
|
||
|
|
assert resp.status_code == 404
|
||
|
|
|
||
|
|
|
||
|
|
# ─── Contacts Contract Contribution (Unit) ───────────────────────────────────
|
||
|
|
|
||
|
|
|
||
|
|
@pytest.mark.asyncio
|
||
|
|
class TestContactsDocumentContribution:
|
||
|
|
async def test_document_placeholders_contact(self):
|
||
|
|
from app.plugins.builtins.contacts.contracts import ContactsContract
|
||
|
|
|
||
|
|
placeholders = ContactsContract.document_placeholders("contact")
|
||
|
|
assert isinstance(placeholders, list)
|
||
|
|
keys = [p["key"] for p in placeholders]
|
||
|
|
assert "firstname" in keys and "email" in keys
|
||
|
|
|
||
|
|
async def test_document_placeholders_unknown_entity_empty(self):
|
||
|
|
from app.plugins.builtins.contacts.contracts import ContactsContract
|
||
|
|
|
||
|
|
assert ContactsContract.document_placeholders("warehouse") == []
|
||
|
|
|
||
|
|
async def test_document_data_loads_entity(self, db_session):
|
||
|
|
seed = await seed_tenant_and_users(db_session)
|
||
|
|
from app.plugins.builtins.contacts.contracts import ContactsContract
|
||
|
|
|
||
|
|
data = await ContactsContract.document_data(
|
||
|
|
db_session,
|
||
|
|
seed["tenant_a"].id,
|
||
|
|
seed["company_a"].id,
|
||
|
|
"company",
|
||
|
|
)
|
||
|
|
assert data["name"] == "Company Alpha"
|
||
|
|
|
||
|
|
async def test_document_data_unknown_entity_empty(self, db_session):
|
||
|
|
seed = await seed_tenant_and_users(db_session)
|
||
|
|
from app.plugins.builtins.contacts.contracts import ContactsContract
|
||
|
|
|
||
|
|
data = await ContactsContract.document_data(
|
||
|
|
db_session,
|
||
|
|
seed["tenant_a"].id,
|
||
|
|
uuid_mod.uuid4(),
|
||
|
|
"company",
|
||
|
|
)
|
||
|
|
assert data == {}
|