ROMFASTSQL/proxmox/lxc171-claude-agent/discord-bridge/tests/test_stream.py

"""T6: parser tolerant."""

from __future__ import annotations

import json

import pytest

import stream


def P(**kw):
    return stream.StreamParser(**kw)


def test_system_init_aduce_session_id():
    p = P()
    (ev,) = p.feed_line(json.dumps({"type": "system", "subtype": "init",
                                    "session_id": "s1", "model": "sonnet", "cwd": "/workspace",
                                    "tools": ["Bash"]}))
    assert isinstance(ev, stream.SystemInit)
    assert ev.session_id == "s1" and ev.model == "sonnet" and ev.tools == ("Bash",)


def test_assistant_text_si_tool_use():
    p = P()
    evs = p.feed_line(json.dumps({"type": "assistant", "session_id": "s1", "message": {"content": [
        {"type": "text", "text": "salut"},
        {"type": "tool_use", "id": "tu1", "name": "Bash", "input": {"command": "ls"}}]}}))
    assert [type(e) for e in evs] == [stream.AssistantText, stream.ToolUse]
    assert evs[0].text == "salut"
    assert evs[1].name == "Bash" and evs[1].input["command"] == "ls"


def test_tool_result_din_user():
    p = P()
    (ev,) = p.feed_line(json.dumps({"type": "user", "message": {"content": [
        {"type": "tool_result", "tool_use_id": "tu1", "content": [{"type": "text", "text": "out"}]}]}}))
    assert isinstance(ev, stream.ToolResult) and ev.text == "out" and ev.is_error is False


def test_result_aduce_cost_si_durata():
    p = P()
    (ev,) = p.feed_line(json.dumps({"type": "result", "subtype": "success", "is_error": False,
                                    "total_cost_usd": 0.1547, "duration_ms": 34500,
                                    "num_turns": 3, "result": "gata"}))
    assert (ev.total_cost_usd, ev.duration_ms, ev.num_turns, ev.text) == (0.1547, 34500, 3, "gata")
    assert ev.is_error is False and p.saw_result


def test_result_cu_campuri_stricate_nu_crapa():
    p = P()
    (ev,) = p.feed_line(json.dumps({"type": "result", "total_cost_usd": "nu-i numar",
                                    "duration_ms": None, "num_turns": "x"}))
    assert ev.total_cost_usd == 0.0 and ev.duration_ms == 0 and ev.num_turns == 0


def test_result_de_eroare_marcat():
    p = P()
    (ev,) = p.feed_line(json.dumps({"type": "result", "subtype": "error_during_execution"}))
    assert ev.is_error is True


def test_tip_necunoscut_logat_o_singura_data():
    calls = []
    p = P(version_fn=lambda: (calls.append(1), "2.1.251")[1])
    line = json.dumps({"type": "tip_care_nu_exista", "detail": "x"})
    assert p.feed_line(line) == []
    assert p.feed_line(line) == []
    assert p.feed_line(json.dumps({"type": "alt_tip_nou"})) == []
    assert p.unknown_types == {"tip_care_nu_exista", "alt_tip_nou"}
    assert len(calls) == 2  # o data per tip, nu per linie


def test_linie_non_json_ignorata(caplog):
    p = P()
    with caplog.at_level("WARNING"):
        assert p.feed_line("Error: something went wrong") == []
        assert p.feed_line("[1, 2, 3]") == []
    assert p.bad_lines == 2
    assert any("non-JSON" in r.message or "nu e obiect" in r.message for r in caplog.records)


def test_linii_goale_ignorate():
    p = P()
    assert p.feed_line("") == [] and p.feed_line("   \n") == []


async def _aiter(lines):
    for x in lines:
        yield x


async def test_eof_inainte_de_result_da_eroare_explicita():
    p = P()
    got = []
    with pytest.raises(stream.StreamEOFError):
        async for ev in p.aiter_events(_aiter([
            json.dumps({"type": "system", "subtype": "init", "session_id": "s1"}),
            json.dumps({"type": "assistant", "message": {"content": [{"type": "text", "text": "hm"}]}}),
        ])):
            got.append(ev)
    assert len(got) == 2


async def test_aiter_se_opreste_la_result():
    p = P()
    evs = [e async for e in p.aiter_events(_aiter([
        json.dumps({"type": "system", "subtype": "init", "session_id": "s1"}),
        json.dumps({"type": "result", "total_cost_usd": 1.0}),
        json.dumps({"type": "assistant", "message": {"content": [{"type": "text", "text": "dupa"}]}}),
    ]))]
    assert [type(e) for e in evs] == [stream.SystemInit, stream.Result]


async def test_aiter_accepta_bytes_si_gunoi():
    p = P()
    evs = [e async for e in p.aiter_events(_aiter([
        b"gunoi binar\n",
        b'{"type":"result","total_cost_usd":0.5}\n',
    ]))]
    assert len(evs) == 1 and evs[0].total_cost_usd == 0.5


# ------------------------------------------ tool_progress si rate_limit_event
# Ambele apareau in bot.log ca "tip necunoscut ignorat" (claude 2.1.251).

def test_tool_progress_devine_eveniment():
    p = P()
    ev = p.feed_line(json.dumps({
        "type": "tool_progress",
        "tool_use_id": "toolu_01-heartbeat-0",
        "tool_name": "Bash",
        "parent_tool_use_id": "toolu_01",
        "elapsed_time_seconds": 30,
        "heartbeat": True,
    }))[0]
    assert isinstance(ev, stream.ToolProgress)
    assert (ev.tool_name, ev.elapsed_s, ev.heartbeat) == ("Bash", 30, True)
    assert p.unknown_types == set()


def test_tool_progress_cu_campuri_aiurea_nu_arunca():
    p = P()
    ev = p.feed_line(json.dumps({"type": "tool_progress", "elapsed_time_seconds": "x"}))[0]
    assert (ev.tool_name, ev.elapsed_s, ev.heartbeat) == ("?", 0, False)


def test_rate_limit_event_devine_eveniment():
    p = P()
    ev = p.feed_line(json.dumps({
        "type": "rate_limit_event",
        "rate_limit_info": {
            "status": "allowed", "resetsAt": 1788171600,
            "rateLimitType": "five_hour", "overageStatus": "rejected",
        },
    }))[0]
    assert isinstance(ev, stream.RateLimit)
    assert ev.ok and ev.limit_type == "five_hour" and ev.resets_at == 1788171600
    assert p.unknown_types == set()


def test_rate_limit_event_franat_nu_e_ok():
    p = P()
    ev = p.feed_line(json.dumps(
        {"type": "rate_limit_event", "rate_limit_info": {"status": "rejected"}}))[0]
    assert not ev.ok and ev.status == "rejected"


def test_rate_limit_event_fara_info_nu_arunca():
    p = P()
    ev = p.feed_line(json.dumps({"type": "rate_limit_event"}))[0]
    assert not ev.ok and ev.status == "necunoscut" and ev.resets_at == 0