1  import pytest
 2  
 3  from gateway.config import GatewayConfig, Platform, PlatformConfig
 4  from gateway.platforms.base import MessageEvent, MessageType
 5  from gateway.run import GatewayRunner
 6  from gateway.session import SessionSource, build_session_key
 7  
 8  
 9  def _make_runner() -> GatewayRunner:
10      runner = GatewayRunner.__new__(GatewayRunner)
11      runner.config = GatewayConfig(
12          platforms={Platform.TELEGRAM: PlatformConfig(enabled=True, token="fake")},
13      )
14      runner.adapters = {}
15      runner._model = "openai/gpt-4.1-mini"
16      runner._base_url = None
17      runner._decide_image_input_mode = lambda: "native"
18      return runner
19  
20  
21  def _source(chat_id: str) -> SessionSource:
22      return SessionSource(
23          platform=Platform.TELEGRAM,
24          chat_id=chat_id,
25          chat_type="private",
26          user_name=f"user-{chat_id}",
27      )
28  
29  
30  def _image_event(source: SessionSource, path: str) -> MessageEvent:
31      return MessageEvent(
32          text="see image",
33          message_type=MessageType.PHOTO,
34          source=source,
35          media_urls=[path],
36          media_types=["image/png"],
37      )
38  
39  
40  @pytest.mark.asyncio
41  async def test_native_image_buffer_isolated_per_session():
42      runner = _make_runner()
43      source_a = _source("chat-a")
44      source_b = _source("chat-b")
45  
46      await runner._prepare_inbound_message_text(
47          event=_image_event(source_a, "/tmp/a.png"),
48          source=source_a,
49          history=[],
50      )
51      await runner._prepare_inbound_message_text(
52          event=_image_event(source_b, "/tmp/b.png"),
53          source=source_b,
54          history=[],
55      )
56  
57      assert runner._consume_pending_native_image_paths(build_session_key(source_a)) == ["/tmp/a.png"]
58      assert runner._consume_pending_native_image_paths(build_session_key(source_b)) == ["/tmp/b.png"]
59  
60  
61  @pytest.mark.asyncio
62  async def test_native_image_buffer_not_cleared_by_other_sessions_without_images():
63      runner = _make_runner()
64      source_a = _source("chat-a")
65      source_b = _source("chat-b")
66  
67      await runner._prepare_inbound_message_text(
68          event=_image_event(source_a, "/tmp/a.png"),
69          source=source_a,
70          history=[],
71      )
72      await runner._prepare_inbound_message_text(
73          event=MessageEvent(text="plain text", source=source_b),
74          source=source_b,
75          history=[],
76      )
77  
78      assert runner._consume_pending_native_image_paths(build_session_key(source_a)) == ["/tmp/a.png"]
79      assert runner._consume_pending_native_image_paths(build_session_key(source_b)) == []