kamidso-back/tests/kamidso/test_event_adapter.py

from kamidso.integrations.event_adapter import EventAdapterState


def test_step_chunk_preserves_full_code_output():
    state = EventAdapterState("run_code", locale="fa-IR")
    events = []
    events.extend(state.map({"type": "step.start", "id": "s1", "title": "code_interpreter", "detail": ""}))
    code = "import pandas as pd\nprint(df.shape)"
    events.extend(
        state.map(
            {
                "type": "step.chunk",
                "id": "s1",
                "output_type": "code",
                "content": code,
            }
        )
    )
    events.extend(
        state.map(
            {
                "type": "step.chunk",
                "id": "s1",
                "output_type": "text",
                "content": "SHAPE: (789, 13)",
            }
        )
    )
    types = [item["type"] for item in events]
    assert "step.updated" not in types
    code_event = next(item for item in events if item["type"] == "step.output" and item["output_type"] == "code")
    assert code_event["content"] == code
    text_event = next(item for item in events if item["type"] == "step.output" and item["output_type"] == "text")
    assert text_event["content"] == "SHAPE: (789, 13)"


def test_maps_react_stream_to_kamidso_events():
    state = EventAdapterState("run_1", locale="fa-IR")
    events = []
    events.extend(state.map({"type": "step.start", "id": "s1", "title": "python", "detail": "inspect"}))
    events.extend(
        state.map(
            {
                "type": "step.chunk",
                "id": "s1",
                "output_type": "chart",
                "content": "plot",
            }
        )
    )
    events.extend(state.map({"type": "step.done", "id": "s1", "status": "done"}))
    events.extend(state.map({"type": "final", "content": "هزینه بیمه ۳۹۷۱"}))
    events.extend(state.map({"type": "done"}))

    types = [item["type"] for item in events]
    assert types[0] == "run.started"
    assert "step.started" in types
    assert "artifact.created" in types
    assert "step.completed" in types
    assert "answer.delta" in types
    assert types[-1] == "run.completed"
    started = next(item for item in events if item["type"] == "step.started")
    assert started["workspace"]["type"] == "step.start"
    assert "تحلیل" in started["title"] or started["title"]


def test_prefers_action_intention_over_generic_tool_label():
    state = EventAdapterState("run_intent", locale="fa-IR")
    events = state.map(
        {
            "type": "step.start",
            "id": "s1",
            "title": "شمارش تیکت‌ها",
            "action": "code_interpreter",
            "action_intention": "شمارش تیکت‌ها",
            "detail": "Thought/Action/Observation",
        }
    )
    started = next(item for item in events if item["type"] == "step.started")
    assert started["action"] == "code_interpreter"
    assert started["step_type"] == "code_execution"
    assert started["title"] == "شمارش تیکت‌ها"


def test_failed_step():
    state = EventAdapterState("run_2")
    events = state.map({"type": "step.done", "id": "s2", "status": "failed"})
    assert any(item["type"] == "step.failed" for item in events)


def test_html_chunk_emits_artifact_with_content():
    state = EventAdapterState("run_html", locale="fa-IR")
    html = "<html dir='rtl' lang='fa'><body><h1>گزارش</h1></body></html>"
    events = state.map(
        {
            "type": "step.chunk",
            "id": "s8",
            "output_type": "html",
            "content": html,
        }
    )
    output = next(item for item in events if item["type"] == "step.output")
    assert output["output_type"] == "html"
    assert output["content"] == html
    artifact = next(item for item in events if item["type"] == "artifact.created")
    assert artifact["artifact"]["type"] == "html"
    assert artifact["artifact"]["content"] == html


def test_report_chunk_emits_artifact_with_content():
    state = EventAdapterState("run_report", locale="fa-IR")
    report = {
        "version": 1,
        "title": "گزارش حقوق",
        "blocks": [{"type": "kpi", "label": "تعداد", "value": 12}],
    }
    events = state.map(
        {
            "type": "step.chunk",
            "id": "s9",
            "output_type": "report",
            "content": report,
        }
    )
    output = next(item for item in events if item["type"] == "step.output")
    assert output["output_type"] == "report"
    assert output["content"]["title"] == "گزارش حقوق"
    artifact = next(item for item in events if item["type"] == "artifact.created")
    assert artifact["artifact"]["type"] == "report"
    assert artifact["artifact"]["name"] == "گزارش حقوق"
    assert artifact["artifact"]["content"] == report


def test_file_chunk_emits_artifact_named_from_payload():
    state = EventAdapterState("run_file", locale="fa-IR")
    content = {"url": "/files/abc12345_buyers.xlsx", "name": "buyers.xlsx"}
    events = state.map(
        {
            "type": "step.chunk",
            "id": "s10",
            "output_type": "file",
            "content": content,
        }
    )
    artifact = next(item for item in events if item["type"] == "artifact.created")
    assert artifact["artifact"]["type"] == "file"
    assert artifact["artifact"]["name"] == "buyers.xlsx"
    assert artifact["artifact"]["content"] == content


def test_maps_question_asked_for_frontend():
    state = EventAdapterState("run_q")
    events = state.map(
        {
            "type": "question.asked",
            "request_id": "que_abc123",
            "conv_id": "conv_1",
            "questions": [
                {
                    "question": "Which file should I analyze?",
                    "header": "Clarification",
                    "options": [{"label": "Book1", "description": "Uploaded Excel"}],
                }
            ],
        }
    )
    assert len(events) == 2
    asked = next(item for item in events if item["type"] == "question.asked")
    assert asked["run_id"] == "run_q"
    assert asked["id"] == "que_abc123"
    assert asked["question_id"] == "que_abc123"
    assert asked["question"] == "Which file should I analyze?"
    assert asked["options"][0]["label"] == "Book1"