nix-estimator/tests/test_miner.py

"""Unit tests for the history miner — internal-json parsing, nix-log fallback, merge."""

import json

from nix_estimator import miner
from nix_estimator.miner import (
    merge_histories,
    median,
    mine_history,
    parse_internal_json,
    parse_nix_log,
)

DRV = "/nix/store/aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa-hello-2.12.drv"
DRV2 = "/nix/store/bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb-openssl-3.0.0.drv"


def _start(id_, drv, *, prefix=True, with_fields=True):
    obj = {
        "action": "start",
        "id": id_,
        "level": 4,
        "parent": 0,
        "type": miner.ACT_BUILD,
        "text": f"building '{drv}'",
    }
    if with_fields:
        obj["fields"] = [drv, "", 1, 1]
    line = json.dumps(obj)
    return (miner._NIX_PREFIX + line) if prefix else line


def _stop(id_, *, prefix=True):
    line = json.dumps({"action": "stop", "id": id_})
    return (miner._NIX_PREFIX + line) if prefix else line


def test_start_stop_pairing_yields_minutes():
    lines = [
        (0.0, _start(1, DRV)),
        (120.0, _stop(1)),  # 120s = 2 min
    ]
    hist = parse_internal_json(lines)
    assert hist == {"hello-2.12": 2.0}


def test_missing_stop_is_dropped():
    lines = [
        (0.0, _start(1, DRV)),
        (30.0, _start(2, DRV2)),
        (90.0, _stop(2)),  # only drv2 finished
    ]
    hist = parse_internal_json(lines)
    assert hist == {"openssl-3.0.0": 1.0}


def test_multiple_drvs():
    lines = [
        (0.0, _start(1, DRV)),
        (10.0, _start(2, DRV2)),
        (70.0, _stop(2)),  # 60s
        (300.0, _stop(1)),  # 300s
    ]
    hist = parse_internal_json(lines)
    assert hist == {"hello-2.12": 5.0, "openssl-3.0.0": 1.0}


def test_drv_from_text_when_no_fields():
    lines = [
        (0.0, _start(1, DRV, with_fields=False)),
        (60.0, _stop(1)),
    ]
    assert parse_internal_json(lines) == {"hello-2.12": 1.0}


def test_lines_without_nix_prefix_still_parse():
    lines = [
        (0.0, _start(1, DRV, prefix=False)),
        (60.0, _stop(1, prefix=False)),
    ]
    assert parse_internal_json(lines) == {"hello-2.12": 1.0}


def test_noise_and_non_build_activities_ignored():
    lines = [
        (0.0, "these are not the logs you are looking for"),
        (0.0, ""),
        (0.0, '@nix {"action":"msg","level":1,"msg":"hi"}'),
        (0.0, json.dumps({"action": "start", "id": 9, "type": 100, "text": "x"})),
        (1.0, _start(1, DRV)),
        (61.0, _stop(1)),
    ]
    assert parse_internal_json(lines) == {"hello-2.12": 1.0}


def test_repeated_name_reduced_with_max_by_default():
    lines = [
        (0.0, _start(1, DRV)),
        (60.0, _stop(1)),  # 1 min
        (60.0, _start(2, DRV)),
        (300.0, _stop(2)),  # 4 min
    ]
    assert parse_internal_json(lines) == {"hello-2.12": 4.0}


def test_raw_string_mode_uses_clock():
    ticks = iter([0.0, 120.0])
    lines = [_start(1, DRV), _stop(1)]
    hist = parse_internal_json(lines, clock=lambda: next(ticks))
    assert hist == {"hello-2.12": 2.0}


def test_raw_string_without_clock_raises():
    try:
        parse_internal_json([_start(1, DRV)])
    except ValueError:
        pass
    else:
        raise AssertionError("expected ValueError without clock")


def test_merge_histories_median():
    runs = [
        {"hello-2.12": 2.0, "openssl-3.0.0": 1.0},
        {"hello-2.12": 4.0, "openssl-3.0.0": 1.5},
        {"hello-2.12": 3.0},
    ]
    merged = merge_histories(runs)
    assert merged["hello-2.12"] == 3.0  # median(2,4,3)
    assert merged["openssl-3.0.0"] == 1.25  # median(1,1.5)


def test_merge_histories_custom_reducer():
    runs = [{"a": 1.0}, {"a": 3.0}]
    assert merge_histories(runs, reducer=lambda v: sum(v) / len(v)) == {"a": 2.0}


def test_median_helper():
    assert median([1.0, 2.0, 3.0]) == 2.0
    assert median([1.0, 3.0]) == 2.0


def test_parse_nix_log_bracket_timestamps():
    lines = [
        "[10:00:00] building hello",
        "[10:00:30] linking",
        "[10:02:00] done",  # 120s span = 2 min
    ]
    assert parse_nix_log(DRV, lines) == {"hello-2.12": 2.0}


def test_parse_nix_log_iso_and_midnight_wrap():
    lines = [
        "2026-07-08T23:59:00 start",
        "2026-07-09T00:01:00 end",  # wraps midnight -> 120s
    ]
    assert parse_nix_log(DRV, lines) == {"hello-2.12": 2.0}


def test_parse_nix_log_no_timestamps_empty():
    assert parse_nix_log(DRV, ["no stamps here", "still none"]) == {}


# Real lines captured from `nix build --log-format internal-json` (stderr), Nix 2.x.
# Slice of /home/oleks/.claude/jobs/0a05a935/tmp/internal-json-sample.jsonl: a type-0
# source copy (must NOT count), a `result` event reusing `type:105` (resProgress — must
# NOT be treated as a build), and the one real type-105 actBuild (remote, on nixbuild).
_REAL_COPY_START = (
    '@nix {"action":"start","id":2290841066405888,"level":5,"parent":0,'
    '"text":"copying \'/nix/store/l61vfkyy0qrnz9bmgx84fa7z3bjzhyp4-source/'
    'pkgs/stdenv/generic/source-stdenv.sh\' to the store","type":0}'
)
_REAL_COPY_STOP = '@nix {"action":"stop","id":2290841066405888}'
_REAL_RESULT_105 = (
    '@nix {"action":"result","fields":[0,0,0,0],"id":2290957030522882,"type":105}'
)
_REAL_BUILD_START = (
    '@nix {"action":"start","fields":['
    '"/nix/store/504l8n0m8gvjha0m8la31ff01nz3w8z4-estimator-probe-15038.drv",'
    '"ssh://root@eu.nixbuild.net",1,1],"id":2290957030522890,"level":3,"parent":0,'
    '"text":"building \'/nix/store/504l8n0m8gvjha0m8la31ff01nz3w8z4-'
    "estimator-probe-15038.drv' on 'ssh://root@eu.nixbuild.net'\",\"type\":105}"
)
_REAL_BUILD_LOGLINE = (
    '@nix {"action":"result","fields":["[nixbuild.net] Build 10389885 queued"],'
    '"id":2290957030522890,"type":101}'
)
_REAL_BUILD_STOP = '@nix {"action":"stop","id":2290957030522890}'


def test_real_capture_only_actbuild_counted():
    """Feed real @nix lines: only the type-105 build is timed; copies/results are not."""
    lines = [
        (0.0, _REAL_COPY_START),  # type 0 source copy — not a build
        (0.5, _REAL_COPY_STOP),
        (10.0, _REAL_RESULT_105),  # result reusing type:105 (resProgress) — not a build
        (100.0, _REAL_BUILD_START),  # the real actBuild starts
        (105.0, _REAL_BUILD_LOGLINE),  # a build-log result line — ignored
        (280.0, _REAL_BUILD_STOP),  # 180s = 3 min
    ]
    hist = parse_internal_json(lines)
    assert hist == {"estimator-probe-15038": 3.0}


def test_real_capture_large_64bit_ids_pair():
    """Real activity ids are large 64-bit ints; start/stop must still pair on them."""
    hist = parse_internal_json([(0.0, _REAL_BUILD_START), (60.0, _REAL_BUILD_STOP)])
    assert hist == {"estimator-probe-15038": 1.0}


def test_mine_history_with_injected_runner():
    def fake_runner(installables):
        assert installables == [".#hello"]
        return [
            (0.0, _start(1, DRV)),
            (180.0, _stop(1)),  # 3 min
        ]

    hist = mine_history([".#hello"], runner=fake_runner)
    assert hist == {"hello-2.12": 3.0}