gac bernina
c7f653ba7c
Previously only eco desktop's console had an activity log (via eco.widgets.console_kernel's ZMQ-message interception, which only works for a kernel with a real iopub channel a Qt widget attaches to) -- a plain terminal `eco console` session and JupyterLab-native/Voila kernels had none. eco.widgets.kernel_registry.install_shell_logger() closes that gap using IPython's pre_run_cell/post_run_cell events instead, which work identically for any real InteractiveShell. Wired into startup_inline.py (terminal), eco_cli.py's generated kernel profile (JupyterLab-native), and voila_app.ipynb/jupyterlab_app.ipynb (Voila / bare `eco jupyterlab`). Narrower than console_kernel's approach by design: only input plus the resulting value/error are captured, not raw stdout/stderr streaming. Also proactively set use_jedi=False for the terminal's `eco console` startup script, matching the reasoning already applied to desktop's console (see console_kernel.build_console_widget's docstring) -- Jedi's introspection was measured slower and less correct against eco's dynamic, lazily-resolving namespace. Every kernel log's entries are now tagged with which stream they came from (TimelineEntry.session, a "kind:label · HHMMSS" label derived from the log's own session_start record) so eco.logs's timeline keeps merging every session by default but LogTimelineQt gained a second, AND-combined "Stream:" filter row alongside the existing "Show:" kind row to separate them back out on demand. Fixed a same-second label collision the first version of the label had (two sessions of the same kind+label started within the same second, e.g. under test, produced identical labels) by falling back to a trailing id fragment from the log filename. Cross-user session browsing (the log_dir parameter already supports it) is deliberately deferred, not built this pass.
231 lines
8.5 KiB
Python
231 lines
8.5 KiB
Python
import json
|
|
|
|
import pytest
|
|
|
|
from eco.widgets import kernel_registry
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def _isolated_registry(monkeypatch):
|
|
# kernel_registry keeps its list of live sessions at module scope --
|
|
# reset it around every test so tests don't see each other's sessions.
|
|
monkeypatch.setattr(kernel_registry, "_registry", [])
|
|
|
|
|
|
def _make_session(tmp_path, **kw):
|
|
return kernel_registry.KernelSession(log_dir=tmp_path, **kw)
|
|
|
|
|
|
def test_kernel_session_creates_log_file_with_session_start(tmp_path):
|
|
session = _make_session(tmp_path, kind="console", label="bernina", pid=1234)
|
|
|
|
assert session.log_path.exists()
|
|
lines = session.log_path.read_text().splitlines()
|
|
assert len(lines) == 1
|
|
entry = json.loads(lines[0])
|
|
assert entry["event"] == "session_start"
|
|
assert entry["kind"] == "console"
|
|
assert entry["label"] == "bernina"
|
|
assert entry["pid"] == 1234
|
|
|
|
|
|
def test_kernel_session_defaults_label_to_kind(tmp_path):
|
|
session = _make_session(tmp_path, kind="desktop")
|
|
assert session.label == "desktop"
|
|
|
|
|
|
def test_log_input_and_log_output_append_jsonl_entries(tmp_path):
|
|
session = _make_session(tmp_path, kind="desktop")
|
|
session.log_input("namespace.motor1.get_current_value()")
|
|
session.log_output("result", text="1.23")
|
|
session.log_output("stream", name="stdout", text="hello\n")
|
|
|
|
lines = [json.loads(l) for l in session.log_path.read_text().splitlines()]
|
|
assert lines[1] == {"t": lines[1]["t"], "event": "input", "code": "namespace.motor1.get_current_value()"}
|
|
assert lines[2]["event"] == "result"
|
|
assert lines[2]["text"] == "1.23"
|
|
assert lines[3]["event"] == "stream"
|
|
assert lines[3]["name"] == "stdout"
|
|
|
|
|
|
def test_log_widget_control_appends_jsonl_entry(tmp_path):
|
|
session = _make_session(tmp_path, kind="desktop")
|
|
session.log_widget_control("cam_west.widget()")
|
|
|
|
lines = [json.loads(l) for l in session.log_path.read_text().splitlines()]
|
|
assert lines[1] == {"t": lines[1]["t"], "event": "widget_control", "code": "cam_west.widget()"}
|
|
|
|
|
|
def test_tail_returns_last_n_entries_oldest_first(tmp_path):
|
|
session = _make_session(tmp_path, kind="desktop")
|
|
for i in range(5):
|
|
session.log_input(f"cmd {i}")
|
|
|
|
tail = session.tail(n=2)
|
|
assert [e.get("code") for e in tail] == ["cmd 3", "cmd 4"]
|
|
|
|
|
|
def test_tail_on_nonexistent_log_returns_empty(tmp_path):
|
|
session = _make_session(tmp_path, kind="desktop")
|
|
session.log_path.unlink()
|
|
assert session.tail() == []
|
|
|
|
|
|
# -- install_shell_logger ------------------------------------------------
|
|
#
|
|
# Works for ANY real IPython InteractiveShell (a plain terminal `eco
|
|
# console` session, a JupyterLab-native kernel, Voila's kernel -- see the
|
|
# function's own docstring for why this needs a different mechanism than
|
|
# eco.widgets.console_kernel's ZMQ-message-based one). A fake shell with
|
|
# just enough of the real API (get_ipython(), .events.register(name, cb))
|
|
# is all that's needed to verify the wiring without a real IPython session.
|
|
|
|
|
|
class _FakeEvents:
|
|
def __init__(self):
|
|
self.registered = {}
|
|
|
|
def register(self, name, callback):
|
|
self.registered.setdefault(name, []).append(callback)
|
|
|
|
def fire(self, name, *args):
|
|
for cb in self.registered.get(name, []):
|
|
cb(*args)
|
|
|
|
|
|
class _FakeShell:
|
|
def __init__(self):
|
|
self.events = _FakeEvents()
|
|
|
|
|
|
class _FakeCellInfo:
|
|
def __init__(self, raw_cell):
|
|
self.raw_cell = raw_cell
|
|
|
|
|
|
class _FakeExecutionResult:
|
|
def __init__(self, result=None, error_in_exec=None, error_before_exec=None):
|
|
self.result = result
|
|
self.error_in_exec = error_in_exec
|
|
self.error_before_exec = error_before_exec
|
|
|
|
|
|
def test_install_shell_logger_returns_none_with_no_ipython_session(monkeypatch):
|
|
monkeypatch.setattr("IPython.get_ipython", lambda: None)
|
|
assert kernel_registry.install_shell_logger("console") is None
|
|
|
|
|
|
def test_install_shell_logger_registers_pre_and_post_run_cell(monkeypatch, tmp_path):
|
|
shell = _FakeShell()
|
|
monkeypatch.setattr("IPython.get_ipython", lambda: shell)
|
|
|
|
session = kernel_registry.install_shell_logger("console", label="bernina", log_dir=tmp_path)
|
|
|
|
assert "pre_run_cell" in shell.events.registered
|
|
assert "post_run_cell" in shell.events.registered
|
|
assert shell._eco_shell_logger_session is session
|
|
|
|
|
|
def test_install_shell_logger_is_idempotent(monkeypatch, tmp_path):
|
|
"""Calling it again on the same shell (e.g. a startup script that runs
|
|
more than once) must not double-register hooks or create a second
|
|
session."""
|
|
shell = _FakeShell()
|
|
monkeypatch.setattr("IPython.get_ipython", lambda: shell)
|
|
|
|
first = kernel_registry.install_shell_logger("console", label="bernina", log_dir=tmp_path)
|
|
second = kernel_registry.install_shell_logger("console", label="bernina", log_dir=tmp_path)
|
|
|
|
assert first is second
|
|
assert len(shell.events.registered["pre_run_cell"]) == 1
|
|
assert len(shell.events.registered["post_run_cell"]) == 1
|
|
|
|
|
|
def test_install_shell_logger_logs_input_and_result(monkeypatch, tmp_path):
|
|
shell = _FakeShell()
|
|
monkeypatch.setattr("IPython.get_ipython", lambda: shell)
|
|
session = kernel_registry.install_shell_logger("console", label="bernina", log_dir=tmp_path)
|
|
|
|
shell.events.fire("pre_run_cell", _FakeCellInfo("2 + 2"))
|
|
shell.events.fire("post_run_cell", _FakeExecutionResult(result=4))
|
|
|
|
lines = [json.loads(l) for l in session.log_path.read_text().splitlines()]
|
|
assert lines[1] == {"t": lines[1]["t"], "event": "input", "code": "2 + 2"}
|
|
assert lines[2] == {"t": lines[2]["t"], "event": "result", "text": "4"}
|
|
|
|
|
|
def test_install_shell_logger_does_not_log_a_result_for_a_bare_statement(monkeypatch, tmp_path):
|
|
"""print(...)/assignments don't produce a result.result -- must not
|
|
log a spurious empty "result" entry for those."""
|
|
shell = _FakeShell()
|
|
monkeypatch.setattr("IPython.get_ipython", lambda: shell)
|
|
session = kernel_registry.install_shell_logger("console", label="bernina", log_dir=tmp_path)
|
|
|
|
shell.events.fire("pre_run_cell", _FakeCellInfo('print("hi")'))
|
|
shell.events.fire("post_run_cell", _FakeExecutionResult(result=None))
|
|
|
|
lines = [json.loads(l) for l in session.log_path.read_text().splitlines()]
|
|
assert [l["event"] for l in lines] == ["session_start", "input"]
|
|
|
|
|
|
def test_install_shell_logger_logs_errors(monkeypatch, tmp_path):
|
|
shell = _FakeShell()
|
|
monkeypatch.setattr("IPython.get_ipython", lambda: shell)
|
|
session = kernel_registry.install_shell_logger("console", label="bernina", log_dir=tmp_path)
|
|
|
|
shell.events.fire("pre_run_cell", _FakeCellInfo("1 / 0"))
|
|
shell.events.fire("post_run_cell", _FakeExecutionResult(error_in_exec=ZeroDivisionError("division by zero")))
|
|
|
|
lines = [json.loads(l) for l in session.log_path.read_text().splitlines()]
|
|
assert lines[2] == {
|
|
"t": lines[2]["t"], "event": "error", "ename": "ZeroDivisionError", "evalue": "division by zero",
|
|
}
|
|
|
|
|
|
def test_install_shell_logger_ignores_blank_cells(monkeypatch, tmp_path):
|
|
shell = _FakeShell()
|
|
monkeypatch.setattr("IPython.get_ipython", lambda: shell)
|
|
session = kernel_registry.install_shell_logger("console", label="bernina", log_dir=tmp_path)
|
|
|
|
shell.events.fire("pre_run_cell", _FakeCellInfo(" \n "))
|
|
|
|
lines = session.log_path.read_text().splitlines()
|
|
assert len(lines) == 1 # just session_start -- no spurious blank input logged
|
|
|
|
|
|
def test_register_and_unregister_control_all_sessions(tmp_path):
|
|
s1 = _make_session(tmp_path, kind="desktop")
|
|
s2 = _make_session(tmp_path, kind="console")
|
|
|
|
kernel_registry.register(s1)
|
|
kernel_registry.register(s2)
|
|
assert kernel_registry.all_sessions() == [s1, s2]
|
|
|
|
kernel_registry.unregister(s1)
|
|
assert kernel_registry.all_sessions() == [s2]
|
|
|
|
|
|
def test_sessions_summary_shape(tmp_path):
|
|
session = _make_session(tmp_path, kind="console", label="alignment", pid=99)
|
|
kernel_registry.register(session)
|
|
|
|
[summary] = kernel_registry.sessions_summary()
|
|
assert summary["id"] == session.id
|
|
assert summary["kind"] == "console"
|
|
assert summary["label"] == "alignment"
|
|
assert summary["pid"] == 99
|
|
assert summary["log_path"] == str(session.log_path)
|
|
|
|
|
|
def test_find_all_logs_lists_jsonl_files(tmp_path):
|
|
_make_session(tmp_path, kind="desktop")
|
|
_make_session(tmp_path, kind="console")
|
|
logs = kernel_registry.find_all_logs(log_dir=tmp_path)
|
|
assert len(logs) == 2
|
|
assert all(p.suffix == ".jsonl" for p in logs)
|
|
|
|
|
|
def test_find_all_logs_missing_dir_returns_empty(tmp_path):
|
|
missing = tmp_path / "does_not_exist"
|
|
assert kernel_registry.find_all_logs(log_dir=missing) == []
|