claude-code-chat-browser/tests/benchmarks/test_export_bench.py at 93cb445d5ab41e9b37ccd4dc83ea359538fb18ad · cppalliance/claude-code-chat-browser · GitHub

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
"""Benchmark run_bulk_export over 10, 50, and 100 session corpora."""

from __future__ import annotations

import io
import zipfile
from pathlib import Path

import pytest

from tests.benchmarks.conftest import TracemallocPeak
from utils.export_engine import NoopSink, ZipSink, run_bulk_export


def _bench_projects(export_corpus: Path) -> list[dict[str, str]]:
    return [{"name": "bench-project", "path": str(export_corpus), "display_name": "Bench"}]


@pytest.mark.benchmark(group="export")
@pytest.mark.parametrize(
    "export_corpus",
    [10, 50, 100],
    indirect=True,
    ids=["sessions-10", "sessions-50", "sessions-100"],
)
def test_bulk_export_session_count(
    benchmark,
    export_corpus: Path,
) -> None:
    projects = _bench_projects(export_corpus)

    def _run() -> object:
        # NoopSink + since="all" + empty last_export_sessions: no disk/state writes per round.
        return run_bulk_export(
            projects=projects,
            since="all",
            rules=[],
            last_export_sessions={},
            sink=NoopSink(),
            fmt="md",
            path_layout="api",
            manifest_style="api",
        )

    result = benchmark(_run)
    assert result.exported_session_count > 0


@pytest.mark.benchmark(group="export")
@pytest.mark.parametrize(
    "export_corpus",
    [10, 50, 100],
    indirect=True,
    ids=["sessions-10", "sessions-50", "sessions-100"],
)
def test_bulk_export_zip_peak_memory(
    benchmark,
    export_corpus: Path,
    tracemalloc_peak: TracemallocPeak,
) -> None:
    projects = _bench_projects(export_corpus)
    peaks: list[int] = []

    def _run() -> None:
        def _export() -> object:
            buf = io.BytesIO()
            with zipfile.ZipFile(buf, "w", zipfile.ZIP_DEFLATED) as zf:
                sink = ZipSink(zf)
                return run_bulk_export(
                    projects=projects,
                    since="all",
                    rules=[],
                    last_export_sessions={},
                    sink=sink,
                    fmt="md",
                    path_layout="api",
                    manifest_style="api",
                )

        result, peak = tracemalloc_peak.measure(_export)
        assert result.exported_session_count > 0
        peaks.append(peak)

    benchmark(_run)
    assert peaks, "benchmark produced no peak memory samples"
    benchmark.extra_info["peak_bytes"] = int(sum(peaks) / len(peaks))