"""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"]