Fjerkroa/discord_bot
1
1
Fork 0
Code Issues Pull Requests Releases Wiki Activity

url-09: fetched images become vision input — fetch_url og:image/body images ride as input_image, direct image urls ingested, data urls never in json tool text

Browse Source
This commit is contained in:
Oleksandr Kozachuk
2026-07-21 13:01:21 +02:00
parent 6f2b3bc040
commit f3c25de310
5 changed files with 222 additions and 22 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/test_spec_url.py
+120 -9
View File
@@ -1,11 +1,14 @@
"""Unit coverage for SPEC-011 URL reading (URL-01..07)."""
"""Unit coverage for SPEC-011 URL reading (URL-01..09)."""
import json
import unittest
from unittest.mock import AsyncMock, patch
from unittest.mock import AsyncMock, Mock, patch
from fjerkroa_bot.openai_responder import OpenAIResponder
from fjerkroa_bot.url_reader import FETCH_URL_TOOL, URLReader, guard_url
from .test_bdd_envelope import envelope
CONFIG = {"openai-token": "t", "model": "m", "system": "s", "history-limit": 5}
@@ -91,7 +94,7 @@ class TestMetaRefresh(unittest.IsolatedAsyncioTestCase):
async def fake_get(session, url, max_bytes):
calls.append(url)
return (url, stub if "stub" in url else article)
return (url, stub if "stub" in url else article, "text/html")
reader._get = fake_get # type: ignore
with patch("fjerkroa_bot.url_reader.guard_url", return_value=None):
@@ -117,7 +120,7 @@ class TestMetaRefresh(unittest.IsolatedAsyncioTestCase):
stub = b'<meta http-equiv="refresh" content="0; url=http://127.0.0.1/secret">Redirecting'
async def fake_get(session, url, max_bytes):
return (url, stub)
return (url, stub, "text/html")
reader._get = fake_get # type: ignore
import fjerkroa_bot.url_reader as ur
@@ -207,11 +210,11 @@ class TestBodyReadCollectsAllChunks(unittest.IsolatedAsyncioTestCase):
return FakeResp()
with patch("fjerkroa_bot.url_reader.guard_url", return_value=None):
_, body = await reader._get(FakeSession(), "http://safe.example.com", 1000)
_, body, _ = await reader._get(FakeSession(), "http://safe.example.com", 1000)
self.assertEqual(body, b"".join(chunks))
with patch("fjerkroa_bot.url_reader.guard_url", return_value=None):
_, body = await reader._get(FakeSession(), "http://safe.example.com", 20)
_, body, _ = await reader._get(FakeSession(), "http://safe.example.com", 20)
self.assertEqual(body, b"".join(chunks)[:20])
@@ -220,7 +223,7 @@ class TestFetchSanitizes(unittest.IsolatedAsyncioTestCase):
"""URL-05: fetch output is length-capped and @everyone-neutralized."""
reader = URLReader(lambda: {"url-max-chars": 50}, None)
payload = ("<p>@everyone " + "x" * 5000 + "</p>").encode()
with patch.object(reader, "_get", new=AsyncMock(return_value=("http://x.com", payload))):
with patch.object(reader, "_get", new=AsyncMock(return_value=("http://x.com", payload, "text/html"))):
result = await reader.fetch("http://x.com", "chat", "alice")
self.assertLessEqual(len(result["text"]), 50)
self.assertNotIn("@everyone", result["text"])
@@ -238,6 +241,8 @@ class TestImageIngest(unittest.IsolatedAsyncioTestCase):
"""URL-06: og:image + <img> ingested (cap honored), internal srcs skipped."""
cache = type("C", (), {})()
cache.ingest_url = AsyncMock(side_effect=["sha1", "sha2", "sha3"])
cache.recent = Mock(return_value=[{"sha256": "sha1", "ext": "jpg"}, {"sha256": "sha2", "ext": "png"}])
cache.data_url = Mock(side_effect=lambda sha, ext: f"data:image/{ext};base64,{sha}")
reader = URLReader(lambda: {"url-max-images": 2}, cache)
html = (
'<meta property="og:image" content="https://cdn.example.com/hero.jpg">'
@@ -249,8 +254,9 @@ class TestImageIngest(unittest.IsolatedAsyncioTestCase):
return "refused" if "127.0.0.1" in url else None
with patch("fjerkroa_bot.url_reader.guard_url", side_effect=fake_guard):
count = await reader._ingest_images(html, "https://example.com", "chat", "alice")
self.assertEqual(count, 2) # og:image + first public img, cap 2
data_urls = await reader._ingest_images(html, "https://example.com", "chat", "alice")
self.assertEqual(len(data_urls), 2) # og:image + first public img, cap 2
self.assertEqual(data_urls[0], "data:image/jpg;base64,sha1") # URL-09: data URLs for vision
ingested = [call.args[0] for call in cache.ingest_url.await_args_list]
self.assertNotIn("http://127.0.0.1/internal.png", ingested)
@@ -265,3 +271,108 @@ class TestPerUserCap(unittest.IsolatedAsyncioTestCase):
blocked = await responder._dispatch_tool("fetch_url", {"url": "http://x.com"}, "alice")
self.assertIn("error", blocked)
self.assertEqual(responder.url_reader.fetch.await_count, 2)
class TestFetchedImagesBecomeVision(unittest.IsolatedAsyncioTestCase):
"""URL-09: fetch results carry vision data URLs, direct image URLs are ingested."""
@staticmethod
def _cache():
cache = Mock()
cache.ingest_url = AsyncMock(return_value="abc123")
cache.ingest_bytes = Mock(return_value="abc123")
cache.recent = Mock(return_value=[{"sha256": "abc123", "ext": "png"}])
cache.data_url = Mock(return_value="data:image/png;base64,AAA")
return cache
@staticmethod
def _session_cm():
import fjerkroa_bot.url_reader as ur
class FakeCM:
async def __aenter__(self):
return object()
async def __aexit__(self, *a):
return False
return patch.object(ur.aiohttp, "ClientSession", return_value=FakeCM())
async def test_html_page_vision_data_urls(self):
"""URL-09: og:image lands in the result's vision list, count matches."""
reader = URLReader(lambda: {}, self._cache())
html = b'<meta property="og:image" content="https://x.com/c.png"><p>Comic of the day, longer text.</p>'
async def fake_get(session, url, max_bytes):
return (url, html, "text/html")
reader._get = fake_get # type: ignore
with patch("fjerkroa_bot.url_reader.guard_url", return_value=None):
with self._session_cm():
result = await reader.fetch("https://xkcd.com/1234", "chat", "alice")
self.assertEqual(result["vision"], ["data:image/png;base64,AAA"])
self.assertEqual(result["images_cached"], 1)
async def test_direct_image_url_ingested(self):
"""URL-09: content-type image/* -> direct ingest, text '(image)'."""
cache = self._cache()
reader = URLReader(lambda: {}, cache)
async def fake_get(session, url, max_bytes):
return (url, b"\x89PNG-bytes", "image/png")
reader._get = fake_get # type: ignore
with patch("fjerkroa_bot.url_reader.guard_url", return_value=None):
with self._session_cm():
result = await reader.fetch("https://imgs.xkcd.com/comics/x.png", "chat", "alice")
self.assertEqual(result["text"], "(image)")
self.assertEqual(result["vision"], ["data:image/png;base64,AAA"])
cache.ingest_bytes.assert_called_once()
async def test_capped_image_body_not_ingested(self):
"""URL-09: an image body at the byte cap may be truncated - not ingested."""
cache = self._cache()
reader = URLReader(lambda: {"url-max-bytes": 10}, cache)
async def fake_get(session, url, max_bytes):
return (url, b"0123456789", "image/png") # len == cap
reader._get = fake_get # type: ignore
with patch("fjerkroa_bot.url_reader.guard_url", return_value=None):
with self._session_cm():
result = await reader.fetch("https://x.com/big.png", "chat", "alice")
self.assertEqual(result["vision"], [])
cache.ingest_bytes.assert_not_called()
class TestLegacyPathVision(unittest.IsolatedAsyncioTestCase):
async def test_legacy_tool_loop_appends_image_message(self):
"""URL-09: legacy path - vision data URLs become an image_url user message; never JSON text."""
responder = OpenAIResponder(dict(CONFIG, **{"enable-url-reading": True}), "chat")
responder._dispatch_tool = AsyncMock(
return_value={"url": "u", "text": "t", "images_cached": 1, "vision": ["data:image/png;base64,AAA"]}
)
func = Mock()
func.name = "fetch_url"
func.arguments = json.dumps({"url": "https://xkcd.com/1"})
call = Mock(id="tc1", type="function", function=func)
first_msg = Mock(content=None, role="assistant", tool_calls=[call], refusal=None)
first = Mock(choices=[Mock(message=first_msg)], usage=None)
final_msg = Mock(content=envelope(answer="seen", answer_needed=True), role="assistant", tool_calls=None, refusal=None)
final = Mock(choices=[Mock(message=final_msg)], usage=None)
with patch("fjerkroa_bot.openai_responder.openai_chat", new_callable=AsyncMock) as chat_mock:
chat_mock.side_effect = [first, final]
answer, _ = await responder.chat([{"role": "user", "content": "look at this"}], 10)
self.assertEqual(json.loads(answer["content"])["answer"], "seen")
final_messages = chat_mock.await_args_list[1].kwargs["messages"]
image_parts = [
part
for msg in final_messages
if isinstance(msg.get("content"), list)
for part in msg["content"]
if part.get("type") == "image_url"
]
self.assertEqual(image_parts[0]["image_url"]["url"], "data:image/png;base64,AAA")
tool_texts = [msg["content"] for msg in final_messages if msg.get("role") == "tool"]
self.assertNotIn("data:image", tool_texts[0]) # data URL never in JSON tool text
self.assertNotIn("vision", tool_texts[0])