analysis/tests/test_core.py

# Licensed to the Apache Software Foundation (ASF) under one
# or more contributor license agreements.  See the NOTICE file
# distributed with this work for additional information
# regarding copyright ownership.  The ASF licenses this file
# to you under the Apache License, Version 2.0 (the
# "License"); you may not use this file except in compliance
# with the License.  You may obtain a copy of the License at
#
#   http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing,
# software distributed under the License is distributed on an
# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
# KIND, either express or implied.  See the License for the
# specific language governing permissions and limitations
# under the License.

"""What the analysis commands share (analysis/_core.py), and `prompts report` end to end."""

from __future__ import annotations

import datetime as dt
import json
import subprocess
import sys

import _core
import pytest
from conftest import ANALYSIS

PROMPTS = ANALYSIS / "prompts"


def test_a_script_finds_the_log_directory_in_config_or_stops(tmp_path):
    cfg = tmp_path / "config.yaml"
    with pytest.raises(SystemExit, match="not found"):
        _core.log_dir(cfg)
    cfg.write_text("prompt_log_dir: /x1/llmao/prompt-log\n")
    assert _core.log_dir(cfg).as_posix() == "/x1/llmao/prompt-log"
    cfg.write_text("prompt_log_dir: ./prompt-log\n")
    assert _core.log_dir(cfg) == tmp_path / "prompt-log"  # relative to the config, not the cwd
    cfg.write_text("other: 1\n")
    with pytest.raises(SystemExit, match="prompt_log_dir is required"):
        _core.log_dir(cfg)


def test_every_analysis_setting_is_required_and_checked_for_shape(tmp_path):
    cfg = tmp_path / "config.yaml"
    cfg.write_text("prompt_log_dir: ./prompt-log\n")
    for key, read in [
        ("analysis_dir", _core.analysis_dir),
        ("analysis_retention_days", _core.retention_days),
        ("analysis_tiers", _core.tiers),
        ("analysis_models", _core.judge_models),
        ("analysis_key", _core.judge_settings),
    ]:
        with pytest.raises(SystemExit, match=f"{key} is required"):
            _ = read(cfg)
    cfg.write_text("analysis_models: qwen3-8b\nanalysis_tiers: [qwen3-8b]\n")
    with pytest.raises(SystemExit, match="analysis_models should be a list"):
        _ = _core.judge_models(cfg)
    with pytest.raises(SystemExit, match="analysis_tiers should be a map"):
        _ = _core.tiers(cfg)
    cfg.write_text(
        "analysis_dir: ./analysis-data\nanalysis_retention_days: 30\nanalysis_gateway: https://llm.example.org/\n"
        "analysis_models: [qwen3-8b]\nanalysis_key: sk-test\nanalysis_tiers: {low: [qwen3-8b], high: [big]}\n"
    )
    assert _core.analysis_dir(cfg) == tmp_path / "analysis-data"
    assert _core.retention_days(cfg) == 30
    assert _core.tiers(cfg) == {"low": ["qwen3-8b"], "medium": [], "high": ["big"]}
    assert _core.judge_settings(cfg) == {"gateway": "https://llm.example.org", "models": ["qwen3-8b"], "key": "sk-test"}


def test_a_span_is_the_last_closed_days_or_from_to():
    today = dt.datetime.now(dt.UTC).date()
    last = _core.span(days=3)
    assert last == [today - dt.timedelta(days=3), today - dt.timedelta(days=2), today - dt.timedelta(days=1)]
    assert _core.span(start=dt.date(2026, 11, 1), end=dt.date(2026, 11, 2)) == [
        dt.date(2026, 11, 1),
        dt.date(2026, 11, 2),
    ]
    assert _core.span(days=2, end=dt.date(2026, 11, 3)) == [dt.date(2026, 11, 2), dt.date(2026, 11, 3)]
    try:
        _core.span(start=dt.date(2026, 11, 3), end=dt.date(2026, 11, 2))
    except ValueError as e:
        assert "after it ends" in str(e)
    else:
        raise AssertionError("a start after its end must be an error")


def _record(
    day: str, model: str = "qwen3-8b", text: str = "what does this function return when the list is empty"
) -> dict:
    return dict(
        ts=f"{day}T09:30:00.000Z",
        model=model,
        status="success",
        error=None,
        stream=False,
        finish_reason="stop",
        params={},
        tokens={"prompt": 500, "completion": 20, "total": 520},
        key={"project": "tooling", "user": "someone"},
        user_agent="OpenAI/Python 1.109.1",
        messages=[
            {"role": "system", "content": "You review the tooling project's Python. Answer briefly."},
            {"role": "user", "content": text},
        ],
    )


def test_an_exemplar_is_re_picked_when_its_day_is_pruned(tmp_path):
    def write(day: str, *texts: str) -> None:
        lines = [json.dumps(_record(day, text=t)) for t in texts]
        (tmp_path / f"prompts-{day}.jsonl").write_text("\n".join(lines) + "\n")

    # Three records on the first day, one on each of the next two: the
    # cluster never doubles, so only the prune can move the exemplar.
    write("2026-11-01", "what does this function return when the list is empty", "a", "b")
    write("2026-11-02", "why does this test fail only on the second run of the day")
    write("2026-11-03", "how should the retry loop back off when the gateway is busy")
    store = _core.Store(tmp_path)
    for d in (1, 2):
        _ = store.scan_day(tmp_path, dt.date(2026, 11, d))
    [c] = store.clusters_over([dt.date(2026, 11, 1), dt.date(2026, 11, 2)])
    assert (c.exemplar_picked, c.records) == ("2026-11-01", 4)
    first = c.exemplar_hash
    # The prune takes the first day, the exemplar's; the next scan re-picks.
    assert store.prune(1, today=dt.date(2026, 11, 3)) == 0
    _ = store.scan_day(tmp_path, dt.date(2026, 11, 3))
    [c] = store.clusters_over([dt.date(2026, 11, 2), dt.date(2026, 11, 3)])
    assert (c.exemplar_picked, c.records) == ("2026-11-03", 2)
    assert c.exemplar_hash != first
    store.close()


def test_a_rebuild_keeps_every_verdict(tmp_path, monkeypatch, capsys):
    import argparse

    import _clusters

    (tmp_path / "prompts-2026-11-01.jsonl").write_text(json.dumps(_record("2026-11-01")) + "\n")
    monkeypatch.setattr(_core, "retention_days", lambda: 90)
    ap = argparse.ArgumentParser()
    args = argparse.Namespace(
        days=7, start=dt.date(2026, 11, 1), end=dt.date(2026, 11, 1), log_dir=tmp_path, analysis_dir=tmp_path
    )
    args.rebuild = False
    assert _clusters.run_build(ap, args) == 0
    store = _core.Store(tmp_path)
    [c] = store.clusters_over([dt.date(2026, 11, 1)])
    store.write_verdict(c, "prompt-1", "judge-1", dict(asf_work="yes", tier="low", activity="other", reason="r"))
    store.close()
    args.rebuild = True
    assert _clusters.run_build(ap, args) == 0
    store = _core.Store(tmp_path)
    assert [tuple(r) for r in store.db.execute("SELECT fingerprint, model FROM verdicts")] == [
        (c.fingerprint, "judge-1")
    ]
    assert store.db.execute("SELECT count(*) FROM current_verdicts").fetchone()[0] == 1
    store.close()
    _ = capsys.readouterr()


def test_prompt_report_prints_the_reports_as_text_and_as_csv(tmp_path):
    (tmp_path / "prompts-2026-11-01.jsonl").write_text(json.dumps(_record("2026-11-01")) + "\n")
    (tmp_path / "prompts-2026-11-02.jsonl").write_text(json.dumps(_record("2026-11-02", "gemma4-26b")) + "\n")
    base = [sys.executable, str(PROMPTS), "report", "--log-dir", str(tmp_path), "--from", "2026-11-01"]
    base += ["--to", "2026-11-02"]
    text = subprocess.run(base, capture_output=True, text=True, check=True).stdout
    assert "2 requests over 2026-11-01 to 2026-11-02, 2 days with records" in text
    assert "By project" in text and "tooling" in text and "Messages per request" in text
    as_csv = subprocess.run([*base, "--csv"], capture_output=True, text=True, check=True).stdout
    assert "# By model\nname,requests,per day,prompt tokens\n" in as_csv
    assert "gemma4-26b,1,0.5,500\n" in as_csv