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"]) == {}


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}