Zum Inhalt

tests/test_integration.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
"""End-to-end pipeline with a fake client and no real host sampling.

Exercises iter_cells → stream_once → record building → merge → report, proving
the whole chain produces report.md + raw.csv with the right row counts.
"""

from touchstone import report
from touchstone.config import Config, ModelSpec
from touchstone.runner import StreamEvent, run_benchmark


class FakeClient:
    engine = "fake"
    engine_version = "1.2.3"

    def stream(self, *, messages, model, max_tokens, temperature, seed):
        yield StreamEvent(delta_text="Hallo ")
        yield StreamEvent(delta_text="Welt")
        yield StreamEvent(prompt_tokens=123, completion_tokens=7)


class NoopSampler:
    def start(self):
        pass

    def stop(self):
        pass


def _cfg(tmp_path):
    return Config.model_validate(
        {
            "endpoint": {"base_url": "http://localhost:1234/v1"},
            "machine": "M1-test",
            "runs_per_cell": 3,
            "context_buckets": [4096],
            "scenarios": ["bodydouble", "rag_synth"],
            "models": [{"id": "qwen3-8b", "quant": "Q5"}],
            "power_check": False,
            "output_dir": str(tmp_path),
        }
    )


def test_run_benchmark_full_pipeline(tmp_path):
    cfg = _cfg(tmp_path)
    run_dir = tmp_path / "run1"
    records = run_benchmark(cfg, FakeClient(), run_dir=run_dir, sampler=NoopSampler())

    # cells: bodydouble (ctx 0) + rag_synth (ctx 4096) = 2 cells
    # per cell: 3 runs; plus 1 cold-start overall = 2*3 + 1 = 7 records
    assert len(records) == 7
    assert sum(1 for r in records if r.is_cold_start) == 1
    assert sum(1 for r in records if r.warmup) == 2  # one warmup per cell
    assert all(r.engine == "lm-studio" for r in records)
    assert all(r.engine_version == "1.2.3" for r in records)
    assert all(r.actual_prompt_tokens == 123 for r in records)

    # Write + read back the report artifacts.
    md_path, raw_path = report.write_report(records, run_dir, date_str="2026-06-15")
    assert md_path.exists() and raw_path.exists()
    md = md_path.read_text(encoding="utf-8")
    assert "Cold-Start" in md
    assert "M1-test" in md
    # engine resolved from the :1234 port, version taken from the client
    assert "lm-studio 1.2.3" in md

    reloaded = report.load_raw_csv(raw_path)
    assert len(reloaded) == 7


def test_run_benchmark_aggregates_exclude_warmup_and_cold(tmp_path, monkeypatch):
    # Make the test hermetic w.r.t. the host's power state: on battery the runner
    # would flag every run and aggregate_cells would exclude them (n_valid → 0).
    monkeypatch.setattr("touchstone.sampler.read_power_source", lambda: "ac")
    cfg = _cfg(tmp_path)
    records = run_benchmark(cfg, FakeClient(), run_dir=tmp_path / "run2", sampler=NoopSampler())
    cells = report.aggregate_cells(records)
    # 2 cells, each with 2 valid runs (3 runs − 1 warmup), cold excluded from cells
    assert len(cells) == 2
    assert all(c.n_valid == 2 for c in cells)


def test_latency_runner_forwards_model_extra_body(tmp_path):
    """`ModelSpec.extra_body` (z. B. `reasoning_effort`) muss AUCH im Latenz-Runner ankommen.
    Er hat es still verworfen — gefunden am 2026-09-20 beim Splash-Smoke: Das Denken blieb an,
    verbrauchte das Token-Budget, und `ttft_s` (= erster sichtbarer Inhalt) blieb `nan`,
    während der Eval-Pfad denselben Knopf korrekt durchreicht."""
    seen: list[dict | None] = []

    class SpyClient(FakeClient):
        def stream(self, *, messages, model, max_tokens, temperature, seed, extra_body=None):
            seen.append(extra_body)
            yield StreamEvent(delta_text="Hallo")
            yield StreamEvent(prompt_tokens=10, completion_tokens=2)

    cfg = _cfg(tmp_path).model_copy(
        update={
            "models": [
                ModelSpec.model_validate(
                    {"id": "qwen3-8b", "quant": "Q5", "extra_body": {"reasoning_effort": "none"}}
                )
            ]
        }
    )
    run_benchmark(cfg, SpyClient(), run_dir=tmp_path / "eb", sampler=NoopSampler())
    assert seen, "kein Request abgesetzt"
    assert all(e == {"reasoning_effort": "none"} for e in seen), seen


def test_latency_runner_omits_extra_body_when_unset(tmp_path):
    """Gegenprobe: ohne Knopf darf nichts mitgeschickt werden (sonst überschriebe ein leeres
    Dict Server-Defaults)."""
    seen: list[dict | None] = []

    class SpyClient(FakeClient):
        def stream(self, *, messages, model, max_tokens, temperature, seed, extra_body=None):
            seen.append(extra_body)
            yield StreamEvent(delta_text="Hallo")
            yield StreamEvent(prompt_tokens=10, completion_tokens=2)

    run_benchmark(_cfg(tmp_path), SpyClient(), run_dir=tmp_path / "nb", sampler=NoopSampler())
    assert seen and all(e is None for e in seen), seen