Files
eco/tests/test_logs_commands.py
lemke_handClaude Sonnet 5 2043340467 Add eco.logs.recent_commands/frequent_commands
The command-history data sources for eco.ipymagic's command-picker mode
("...h") -- merged across every kernel-log session on disk (same source
eco.logs.widget()'s timeline already reads), deduplicated by keeping each
distinct command's most-recent occurrence (recent_commands) or ranking by
use count with a recency tie-break (frequent_commands), rather than
letting a command run every few minutes crowd out everything else.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
2026-09-12 17:05:44 +02:00

87 lines
2.6 KiB
Python

"""eco.logs.recent_commands / frequent_commands -- the command-history
data sources for eco.ipymagic's command-picker mode. Real KernelSession
files (not hand-written JSONL) so this also incidentally exercises the
real on-disk format recent_commands/frequent_commands actually parse."""
import pytest
from eco.widgets import kernel_registry
@pytest.fixture(autouse=True)
def _isolated_log_dir(tmp_path, monkeypatch):
monkeypatch.setattr(kernel_registry, "DEFAULT_LOG_DIR", tmp_path)
monkeypatch.setattr(kernel_registry, "_registry", [])
return tmp_path
def _session(tmp_path, kind="console", label="bernina"):
return kernel_registry.register(
kernel_registry.KernelSession(kind=kind, label=label, log_dir=tmp_path)
)
def test_recent_commands_are_most_recent_first_and_deduplicated(tmp_path):
from eco import logs
session = _session(tmp_path)
session.log_input("daq.compare_channels()")
session.log_input("mono.mv(5)")
session.log_input("daq.compare_channels()") # repeat -- should move to front, not duplicate
assert logs.recent_commands() == ["daq.compare_channels()", "mono.mv(5)"]
def test_recent_commands_ignores_blank_input(tmp_path):
from eco import logs
session = _session(tmp_path)
session.log_input(" ")
session.log_input("mono.mv(5)")
assert logs.recent_commands() == ["mono.mv(5)"]
def test_recent_commands_merges_across_sessions(tmp_path):
from eco import logs
s1 = _session(tmp_path, label="bernina")
s1.log_input("mono.mv(5)")
s2 = _session(tmp_path, label="bernina2")
s2.log_input("daq.compare_channels()")
assert set(logs.recent_commands()) == {"mono.mv(5)", "daq.compare_channels()"}
def test_recent_commands_respects_n(tmp_path):
from eco import logs
session = _session(tmp_path)
for i in range(5):
session.log_input(f"cmd_{i}()")
assert len(logs.recent_commands(n=2)) == 2
def test_frequent_commands_ranks_by_use_count(tmp_path):
from eco import logs
session = _session(tmp_path)
for _ in range(3):
session.log_input("daq.compare_channels()")
session.log_input("mono.mv(5)")
assert logs.frequent_commands()[:2] == ["daq.compare_channels()", "mono.mv(5)"]
def test_frequent_commands_ties_break_by_recency(tmp_path):
import time
from eco import logs
session = _session(tmp_path)
session.log_input("older_but_tied()")
time.sleep(0.01) # guarantee a distinguishable timestamp, not just a distinguishable call
session.log_input("newer_and_tied()")
assert logs.frequent_commands()[:2] == ["newer_and_tied()", "older_but_tied()"]