kamidso-back/tests/kamidso/test_report_v2.py

import pytest

from kamidso.contracts.report import validate_report
from kamidso.contracts.user import CurrentUser
from kamidso.repositories.store import RepositoryStore
from kamidso.resources.reports import ReportService, bind_report_service


@pytest.mark.asyncio
async def test_report_service_create_and_get():
    store = RepositoryStore()
    service = ReportService(store)
    user = CurrentUser(
        id="usr_1",
        email="a@b.c",
        name="A",
        role="user",
        workspace_id="ws_1",
    )
    spec = validate_report(
        {
            "version": 2,
            "title": "فروش",
            "blocks": [{"type": "text", "content": "خلاصه"}],
        }
    )
    doc = await service.create(user, spec, conversation_id="conv_1")
    assert doc["_id"].startswith("rpt_")
    got = await service.get_report(user, doc["_id"])
    assert got["title"] == "فروش"
    assert got["report_spec"]["blocks"][0]["type"] == "text"
    public = ReportService.to_public(got)
    assert public["id"] == doc["_id"]
    assert public["report_spec"]["title"] == "فروش"


@pytest.mark.asyncio
async def test_report_autosave_from_conversation():
    store = RepositoryStore()
    service = ReportService(store)
    bind_report_service(service)
    await store.create_conversation(
        {
            "_id": "conv_auto",
            "workspace_id": "ws_auto",
            "created_by": "usr_auto",
            "title": "chat",
        }
    )
    spec = validate_report(
        {
            "version": 2,
            "title": "Auto",
            "blocks": [{"type": "kpi", "label": "A", "value": 1}],
        }
    )
    saved = await service.autosave_from_tool(spec, conversation_id="conv_auto")
    assert saved is not None
    assert saved["workspace_id"] == "ws_auto"
    assert saved["_id"].startswith("rpt_")
    bind_report_service(None)


@pytest.mark.asyncio
async def test_report_interpreter_flow_without_engine_import():
    """Mirror report_interpreter validate + autosave path without loading engine tools."""
    import json

    store = RepositoryStore()
    service = ReportService(store)
    bind_report_service(service)
    await store.create_conversation(
        {
            "_id": "conv_tool",
            "workspace_id": "ws_tool",
            "created_by": "usr_tool",
        }
    )
    validated = validate_report(
        {
            "version": 2,
            "title": "از ابزار",
            "datasets": {
                "m": [{"month": "A", "sales": 1, "profit": 2}],
            },
            "blocks": [
                {
                    "type": "chart",
                    "chartType": "line",
                    "dataset": "m",
                    "x": "month",
                    "series": [
                        {"field": "sales", "label": "فروش"},
                        {"field": "profit", "label": "سود"},
                    ],
                }
            ],
        }
    )
    saved = await service.autosave_from_tool(
        validated,
        workspace_id="ws_tool",
        user_id="usr_tool",
        conversation_id="conv_tool",
    )
    assert saved is not None
    validated["report_id"] = saved["_id"]
    chunks = [
        {"output_type": "text", "content": f"report_id={validated['report_id']}"},
        {"output_type": "report", "content": validated},
    ]
    payload = json.loads(json.dumps({"chunks": chunks}, ensure_ascii=False))
    report_chunk = next(c for c in payload["chunks"] if c["output_type"] == "report")
    assert report_chunk["content"]["version"] == 2
    assert report_chunk["content"]["report_id"].startswith("rpt_")
    assert len(report_chunk["content"]["blocks"][0]["series"]) == 2
    bind_report_service(None)


@pytest.mark.asyncio
async def test_shared_report_omits_workspace_and_conversation():
    store = RepositoryStore()
    service = ReportService(store)
    user = CurrentUser(
        id="usr_1",
        email="a@b.c",
        name="A",
        role="user",
        workspace_id="ws_1",
    )
    spec = validate_report(
        {
            "version": 2,
            "title": "فروش",
            "blocks": [{"type": "text", "content": "خلاصه"}],
        }
    )
    doc = await service.create(user, spec, conversation_id="conv_secret")
    shared = await service.get_shared_report(doc["_id"])
    public = ReportService.to_share_public(shared)
    assert public["id"] == doc["_id"]
    assert public["title"] == "فروش"
    assert "conversation_id" not in public
    assert "created_by" not in public
    assert public["report_spec"]["blocks"][0]["content"] == "خلاصه"

    await service.update_report(user, doc["_id"], status="archived")
    with pytest.raises(KeyError):
        await service.get_shared_report(doc["_id"])


@pytest.mark.asyncio
async def test_html_share_is_public_and_reused():
    store = RepositoryStore()
    service = ReportService(store)
    user = CurrentUser(
        id="usr_1",
        email="a@b.c",
        name="A",
        role="user",
        workspace_id="ws_1",
    )
    html = "<html dir='rtl' lang='fa'><head><title>فروش</title></head><body><h1>فروش</h1></body></html>"
    first = await service.create_html_share(user, html)
    second = await service.create_html_share(user, html, title="عنوان دیگر")
    assert first["_id"] == second["_id"]
    assert first["title"] == "فروش"
    assert "conversation_id" not in first or first.get("conversation_id") is None

    shared = await service.get_shared_report(first["_id"])
    public = ReportService.to_share_public(shared)
    assert public["kind"] == "html"
    assert public["html"] == html.strip()
    assert public["title"] == "فروش"
    assert "workspace_id" not in public
    assert "created_by" not in public
    assert "html" not in public["report_spec"]

    with pytest.raises(ValueError):
        await service.create_html_share(user, "   ")