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

"""Atasamente Discord -> blocuri de continut pentru Claude.

Imaginile ajung blocuri `image` (base64), fisierele text sunt inserate in prompt,
restul sunt doar numite. Zero retea: `FakeAttachment.read()` returneaza octeti locali.
"""

from __future__ import annotations

import base64

import pytest

import bot
import runner as runner_mod
from tests.test_bot import (  # noqa: F401  (bridge/store/allowed sunt fixturi)
    FakeAttachment, FakeBrokenAttachment, FakeMessage, allowed, bridge,
)

PNG = b"\x89PNG\r\n\x1a\n" + b"x" * 40


def image(name="poza.png", data=PNG, content_type="image/png", size=None):
    return FakeAttachment(name, data, content_type, size)


async def build(content="", attachments=()):
    msg = FakeMessage(content, attachments=list(attachments))
    return await bot.build_user_content(msg, content)


def text_block(blocks) -> str:
    return blocks[-1]["text"]


# ------------------------------------------------------------------ normalizare
@pytest.mark.parametrize(
    "content_type,filename,expected",
    [
        ("image/png", "a.png", "image/png"),
        ("image/jpeg", "a.jpg", "image/jpeg"),
        ("image/jpg", "a.jpg", "image/jpeg"),           # nestandard, normalizat
        ("image/png; charset=utf-8", "a.png", "image/png"),
        ("IMAGE/PNG", "a.png", "image/png"),
        ("", "a.webp", "image/webp"),                    # rezerva pe extensie
        ("application/octet-stream", "a.gif", "image/gif"),
        ("image/svg+xml", "a.svg", None),                # nesuportat de API
        ("image/heic", "a.heic", None),
        ("application/pdf", "a.pdf", None),
        ("", "a.bin", None),
    ],
)
def test_tipul_imaginii_e_normalizat_si_filtrat(content_type, filename, expected):
    assert bot.image_media_type(content_type, filename) == expected


def test_fisierele_text_sunt_recunoscute_dupa_mime_sau_extensie():
    assert bot.is_text_attachment("text/plain", "x.txt")
    assert bot.is_text_attachment("application/json", "x.json")
    assert bot.is_text_attachment("", "eroare.log")
    assert bot.is_text_attachment("application/octet-stream", "script.py")
    assert not bot.is_text_attachment("application/pdf", "manual.pdf")
    assert not bot.is_text_attachment("", "arhiva.zip")


# ------------------------------------------------------------------ constructie
@pytest.mark.asyncio
async def test_fara_atasamente_ramane_text_simplu():
    assert await build("salut") == "salut"


@pytest.mark.asyncio
async def test_imaginea_devine_bloc_base64_iar_textul_e_ultimul():
    blocks = await build("ce vezi?", [image()])
    assert [b["type"] for b in blocks] == ["image", "text"]
    src = blocks[0]["source"]
    assert src["media_type"] == "image/png"
    assert base64.b64decode(src["data"]) == PNG
    assert text_block(blocks) == "ce vezi?"


@pytest.mark.asyncio
async def test_imagine_fara_text_primeste_instructiune_implicita():
    blocks = await build("", [image()])
    assert "Analizeaza imaginea" in text_block(blocks)


@pytest.mark.asyncio
async def test_maxim_patru_imagini_restul_sunt_raportate():
    atts = [image(f"p{i}.png") for i in range(6)]
    blocks = await build("hai", atts)
    assert sum(1 for b in blocks if b["type"] == "image") == bot.MAX_IMAGE_ATTACHMENTS
    assert "p4.png" in text_block(blocks) and "p5.png" in text_block(blocks)


@pytest.mark.asyncio
async def test_imaginea_prea_mare_e_sarita_fara_descarcare():
    att = image("uriasa.png", size=bot.MAX_IMAGE_BYTES + 1)
    blocks = await build("hai", [att])
    assert isinstance(blocks, str)          # nimic lizibil ramas
    assert "uriasa.png" in blocks
    assert att.reads == 0                   # nu descarcam ce stim ca nu incape


@pytest.mark.asyncio
async def test_marimea_reala_e_recontrolata_dupa_descarcare():
    """Discord poate declara mai putin decat trimite; contam octetii primiti."""
    att = image("mincinoasa.png", data=b"y" * (bot.MAX_IMAGE_BYTES + 10), size=10)
    blocks = await build("hai", [att])
    assert att.reads == 1
    assert isinstance(blocks, str) and "mincinoasa.png" in blocks


@pytest.mark.asyncio
async def test_fisierul_text_e_inserat_in_prompt():
    att = FakeAttachment("eroare.log", b"ORA-03113 end-of-file", "text/plain")
    blocks = await build("ce inseamna?", [att])
    body = text_block(blocks)
    assert "ce inseamna?" in body
    assert "eroare.log" in body and "ORA-03113" in body


@pytest.mark.asyncio
async def test_fisierul_text_prea_mare_e_trunchiat_nu_aruncat():
    data = b"a" * (bot.MAX_TEXT_BYTES + 500)
    blocks = await build("vezi", [FakeAttachment("mare.log", data, "text/plain")])
    body = text_block(blocks)
    assert "trunchiat" in body
    assert len(body) < len(data)


@pytest.mark.asyncio
async def test_text_binar_nu_arunca_exceptie():
    att = FakeAttachment("stricat.txt", b"\xff\xfe\x00binar", "text/plain")
    blocks = await build("vezi", [att])
    assert "stricat.txt" in text_block(blocks)


@pytest.mark.asyncio
async def test_imagini_si_text_impreuna():
    atts = [image(), FakeAttachment("note.md", b"# titlu", "text/markdown")]
    blocks = await build("analizeaza", atts)
    assert [b["type"] for b in blocks] == ["image", "text"]
    assert "# titlu" in text_block(blocks)


@pytest.mark.asyncio
async def test_atasamentul_nesuportat_e_numit_nu_ignorat_tacut():
    blocks = await build("uite", [FakeAttachment("manual.pdf", b"%PDF", "application/pdf")])
    assert isinstance(blocks, str)
    assert "manual.pdf" in blocks and "uite" in blocks


@pytest.mark.asyncio
async def test_eroarea_de_descarcare_nu_pica_turul():
    att = FakeBrokenAttachment("poza.png", PNG, "image/png")
    blocks = await build("vezi", [att])
    assert isinstance(blocks, str)
    assert "poza.png" in blocks


@pytest.mark.asyncio
async def test_o_imagine_buna_supravietuieste_langa_una_stricata():
    atts = [FakeBrokenAttachment("rea.png", PNG, "image/png"), image("buna.png")]
    blocks = await build("vezi", atts)
    assert sum(1 for b in blocks if b["type"] == "image") == 1
    assert "rea.png" in text_block(blocks)


# ------------------------------------------------------------------ transport
def test_blocurile_ajung_intacte_in_mesajul_de_pe_stdin():
    blocks = [
        {"type": "image", "source": {"type": "base64", "media_type": "image/png", "data": "QQ=="}},
        {"type": "text", "text": "ce vezi?"},
    ]
    import json

    payload = json.loads(runner_mod.user_message(blocks))
    assert payload["message"]["content"] == blocks


def test_textul_simplu_ramane_un_singur_bloc_text():
    import json

    payload = json.loads(runner_mod.user_message("salut"))
    assert payload["message"]["content"] == [{"type": "text", "text": "salut"}]


# ---------------------------------------------------- integrare prin punte
@pytest.mark.asyncio
async def test_mesaj_cu_imagine_trece_prin_punte_pana_la_cli(bridge, monkeypatch):
    """Blocurile construite in bot ajung neatinse la `proc.run_turn`."""
    vazut = {}

    async def fake_turn(prompt, on_event=None, timeout=None):
        vazut["prompt"] = prompt
        return runner_mod.TurnOutcome(
            result=bot.stream_mod.Result(0.01, 10, False, 1, text="ok")
        )

    msg = FakeMessage("ce vezi?", attachments=[image()])
    proc = bridge.runner.get(str(msg.channel.id))
    monkeypatch.setattr(proc, "run_turn", fake_turn)

    assert await bridge.handle_message(msg) == "ok"
    blocks = vazut["prompt"]
    assert [b["type"] for b in blocks] == ["image", "text"]
    assert base64.b64decode(blocks[0]["source"]["data"]) == PNG


@pytest.mark.asyncio
async def test_mesaj_doar_cu_atasament_nu_mai_e_gol(bridge):
    """Inainte, un mesaj fara text era respins ca 'empty' si poza se pierdea."""
    msg = FakeMessage("", attachments=[image()])
    assert await bridge.handle_message(msg) == "ok"


@pytest.mark.asyncio
async def test_mesaj_complet_gol_ramane_respins(bridge):
    assert await bridge.handle_message(FakeMessage("")) == "empty"


@pytest.mark.asyncio
async def test_steering_mid_tur_trimite_si_imaginea(bridge, monkeypatch):
    """O poza trimisa in timpul unui tur intra pe stdin, nu deschide tur nou."""
    trimis = {}
    tid = "200"
    proc = bridge.runner.get(tid)

    async def fake_send(content):
        trimis["content"] = content

    monkeypatch.setattr(proc, "send", fake_send)
    monkeypatch.setattr(type(proc), "alive", property(lambda self: True))
    proc.inflight = True

    msg = FakeMessage("uite eroarea", attachments=[image()])
    assert await bridge.handle_message(msg) == "steered"
    assert [b["type"] for b in trimis["content"]] == ["image", "text"]