ed
0f5290f038
Migrates 5 SILENT_SWALLOW sites to full Result[T] pattern: session_logger.py (4 sites): 1. log_api_hook - returns Result[bool] (was None) 2. log_comms - returns Result[bool] (was None) 3. log_tool_call - returns Result[Optional[str]] (was Optional[str]) 4. log_cli_call - returns Result[bool] (was None) file_cache.py (1 site): - L98: removed dead code (try/except StopIteration around next(iter(_ast_cache)) is unreachable because we just checked len(_ast_cache) >= 10) Updates tests/test_session_logger_optimization.py to extract result.data from the new Result-based API. All callers of these log_* functions previously ignored the return value; they continue to ignore the new Result return value (backwards-compatible).
98 lines
3.6 KiB
Python
98 lines
3.6 KiB
Python
import pytest
|
|
from pathlib import Path
|
|
from typing import Generator
|
|
from src import session_logger
|
|
from src import paths
|
|
|
|
@pytest.fixture
|
|
def temp_session_setup(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> Generator[tuple[Path, Path], None, None]:
|
|
# Ensure session is closed and state is reset
|
|
session_logger.close_session()
|
|
monkeypatch.setattr(session_logger, "_comms_fh", None)
|
|
monkeypatch.setattr(session_logger, "_session_dir", None)
|
|
monkeypatch.setattr(session_logger, "_seq", 0)
|
|
monkeypatch.setattr(session_logger, "_output_seq", 0)
|
|
|
|
log_dir = tmp_path / "logs"
|
|
scripts_dir = tmp_path / "scripts" / "generated"
|
|
log_dir.mkdir(parents=True, exist_ok=True)
|
|
scripts_dir.mkdir(parents=True, exist_ok=True)
|
|
|
|
monkeypatch.setattr(paths, "get_logs_dir", lambda: log_dir)
|
|
monkeypatch.setattr(paths, "get_scripts_dir", lambda: scripts_dir)
|
|
|
|
yield log_dir, scripts_dir
|
|
|
|
# Cleanup
|
|
session_logger.close_session()
|
|
|
|
def test_session_directory_and_subdirectories_creation(temp_session_setup: tuple[Path, Path]) -> None:
|
|
log_dir, _ = temp_session_setup
|
|
session_logger.open_session(label="opt-test")
|
|
|
|
# Find the session directory
|
|
session_dirs = [d for d in log_dir.iterdir() if d.is_dir()]
|
|
assert len(session_dirs) == 1
|
|
session_dir = session_dirs[0]
|
|
|
|
assert (session_dir / "scripts").exists()
|
|
assert (session_dir / "outputs").exists()
|
|
assert (session_dir / "comms.log").exists()
|
|
assert (session_dir / "toolcalls.log").exists()
|
|
|
|
def test_log_tool_call_saves_in_session_scripts(temp_session_setup: tuple[Path, Path]) -> None:
|
|
log_dir, _ = temp_session_setup
|
|
session_logger.open_session(label="tool-call-test")
|
|
|
|
# Find the session directory
|
|
session_dir = next(d for d in log_dir.iterdir() if d.is_dir())
|
|
scripts_subdir = session_dir / "scripts"
|
|
|
|
script_content = "Write-Host 'Hello from test'"
|
|
result_content = "Success"
|
|
|
|
# Call log_tool_call with script_path=None. Returns Result[Optional[str]].
|
|
ps1_result = session_logger.log_tool_call(script_content, result_content, None)
|
|
assert ps1_result.ok, f"log_tool_call failed: {ps1_result.errors}"
|
|
assert ps1_result.data is not None
|
|
|
|
ps1_path = Path(ps1_result.data)
|
|
assert ps1_path.parent == scripts_subdir
|
|
assert ps1_path.name == "script_0001.ps1"
|
|
assert ps1_path.read_text(encoding="utf-8") == script_content
|
|
|
|
# Verify second call increments sequence
|
|
ps1_result_2 = session_logger.log_tool_call("Get-Date", "2026-03-08", None)
|
|
assert ps1_result_2.ok, f"log_tool_call failed: {ps1_result_2.errors}"
|
|
assert ps1_result_2.data is not None
|
|
assert Path(ps1_result_2.data).name == "script_0002.ps1"
|
|
|
|
def test_log_tool_output_saves_in_session_outputs(temp_session_setup: tuple[Path, Path]) -> None:
|
|
log_dir, _ = temp_session_setup
|
|
session_logger.open_session(label="output-test")
|
|
|
|
# Find the session directory
|
|
session_dir = next(d for d in log_dir.iterdir() if d.is_dir())
|
|
outputs_subdir = session_dir / "outputs"
|
|
|
|
output_content = "This is some tool output content."
|
|
|
|
# Call log_tool_output
|
|
output_path_str = session_logger.log_tool_output(output_content)
|
|
assert output_path_str is not None
|
|
|
|
output_path = Path(output_path_str)
|
|
assert output_path.parent == outputs_subdir
|
|
assert output_path.name == "output_0001.txt"
|
|
assert output_path.read_text(encoding="utf-8") == output_content
|
|
|
|
# Verify second call increments sequence
|
|
output_path_str_2 = session_logger.log_tool_output("More content")
|
|
assert output_path_str_2 is not None
|
|
assert Path(output_path_str_2).name == "output_0002.txt"
|
|
|
|
def test_log_tool_output_returns_none_if_no_session(temp_session_setup: tuple[Path, Path]) -> None:
|
|
# We don't call open_session here
|
|
output_path_str = session_logger.log_tool_output("Should not save")
|
|
assert output_path_str is None
|