Zum Inhalt

tests/test_client.py

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
from types import SimpleNamespace

from touchstone.client import OpenAIStreamClient


def test_client_yields_reasoning_and_content_from_delta(monkeypatch):
    client = OpenAIStreamClient("http://localhost:9/v1")
    # chunk 1: only reasoning_content; chunk 2: only content; chunk 3: usage
    chunks = [
        SimpleNamespace(
            choices=[
                SimpleNamespace(delta=SimpleNamespace(content=None, reasoning_content="think"))
            ],
            usage=None,
        ),
        SimpleNamespace(
            choices=[SimpleNamespace(delta=SimpleNamespace(content="hi"))],
            usage=None,
        ),
        SimpleNamespace(
            choices=[],
            usage=SimpleNamespace(prompt_tokens=3, completion_tokens=1),
        ),
    ]
    monkeypatch.setattr(client._client.chat.completions, "create", lambda **kw: chunks)

    evs = list(client.stream(messages=[], model="m", max_tokens=1, temperature=0.0, seed=0))
    assert any(e.reasoning_text == "think" for e in evs)
    assert any(e.delta_text == "hi" for e in evs)
    assert any(e.completion_tokens == 1 for e in evs)


def test_client_supports_reasoning_field_alias(monkeypatch):
    client = OpenAIStreamClient("http://localhost:9/v1")
    chunk = SimpleNamespace(
        choices=[SimpleNamespace(delta=SimpleNamespace(content=None, reasoning="hmm"))],
        usage=None,
    )
    monkeypatch.setattr(client._client.chat.completions, "create", lambda **kw: [chunk])
    evs = list(client.stream(messages=[], model="m", max_tokens=1, temperature=0.0, seed=0))
    assert any(e.reasoning_text == "hmm" for e in evs)


def test_stream_forwards_extra_body_only_when_set(monkeypatch):
    from touchstone.client import OpenAIStreamClient

    captured = {}

    class _FakeCreate:
        def __call__(self, **kwargs):
            captured.update(kwargs)
            return iter([])  # empty stream

    class _FakeOpenAI:
        def __init__(self, **_):
            self.chat = type("C", (), {"completions": type("X", (), {"create": _FakeCreate()})()})()
            self.models = None

    monkeypatch.setattr("openai.OpenAI", _FakeOpenAI)
    c = OpenAIStreamClient("http://x/v1")

    list(
        c.stream(
            messages=[{"role": "user", "content": "hi"}],
            model="m",
            max_tokens=10,
            temperature=0.0,
            seed=42,
        )
    )
    assert "extra_body" not in captured  # not set → not forwarded (no SDK default override)

    captured.clear()
    list(
        c.stream(
            messages=[{"role": "user", "content": "hi"}],
            model="m",
            max_tokens=10,
            temperature=0.0,
            seed=42,
            extra_body={"chat_template_kwargs": {"enable_thinking": False}},
        )
    )
    assert captured["extra_body"] == {"chat_template_kwargs": {"enable_thinking": False}}


def test_stream_omits_max_tokens_when_none(monkeypatch):
    from touchstone.client import OpenAIStreamClient

    captured = {}

    class _FakeCreate:
        def __call__(self, **kwargs):
            captured.update(kwargs)
            return iter([])  # empty stream

    class _FakeOpenAI:
        def __init__(self, **_):
            self.chat = type("C", (), {"completions": type("X", (), {"create": _FakeCreate()})()})()
            self.models = None

    monkeypatch.setattr("openai.OpenAI", _FakeOpenAI)
    c = OpenAIStreamClient("http://x/v1")

    # max_tokens=None → no cap sent → the server answers freely (eval default)
    list(
        c.stream(
            messages=[{"role": "user", "content": "hi"}],
            model="m",
            max_tokens=None,
            temperature=0.0,
            seed=42,
        )
    )
    assert "max_tokens" not in captured

    captured.clear()
    list(
        c.stream(
            messages=[{"role": "user", "content": "hi"}],
            model="m",
            max_tokens=256,
            temperature=0.0,
            seed=42,
        )
    )
    assert captured["max_tokens"] == 256