@@ -1,4 +1,4 @@
|
||||
"""Reproduction test: list_sessions drops corrupt legacy-stem sessions during repair."""
|
||||
"""Regression tests for legacy-stem session handling."""
|
||||
import json
|
||||
from datetime import datetime
|
||||
from pathlib import Path
|
||||
@@ -38,3 +38,39 @@ def test_list_sessions_repairs_corrupt_legacy_stem(tmp_path: Path, monkeypatch)
|
||||
# actual legacy filename. The session is silently dropped.
|
||||
assert len(sessions) == 1, f"Expected 1 session, got {len(sessions)}"
|
||||
assert sessions[0]["key"] == "telegram:12345"
|
||||
|
||||
|
||||
def test_read_session_methods_fall_back_to_legacy_lossy_stem(
|
||||
tmp_path: Path,
|
||||
monkeypatch,
|
||||
) -> None:
|
||||
monkeypatch.setattr(
|
||||
"nanobot.session.manager.get_legacy_sessions_dir",
|
||||
lambda: tmp_path / "legacy_sessions",
|
||||
)
|
||||
manager = SessionManager(tmp_path / "workspace")
|
||||
session_id = "123e4567-e89b-12d3-a456-426614174000"
|
||||
key = f"websocket:{session_id}"
|
||||
legacy_path = manager._get_legacy_lossy_path(key)
|
||||
assert legacy_path.name == f"websocket_{session_id}.jsonl"
|
||||
|
||||
metadata = {
|
||||
"_type": "metadata",
|
||||
"key": key,
|
||||
"created_at": datetime(2025, 1, 1).isoformat(),
|
||||
"updated_at": datetime(2025, 1, 1).isoformat(),
|
||||
"metadata": {
|
||||
"workspace_scope": "project",
|
||||
"project_path": "/tmp/example-project",
|
||||
},
|
||||
}
|
||||
legacy_path.write_text(json.dumps(metadata) + "\n", encoding="utf-8")
|
||||
|
||||
metadata_result = manager.read_session_metadata(key)
|
||||
file_result = manager.read_session_file(key)
|
||||
|
||||
assert metadata_result is not None
|
||||
assert metadata_result["metadata"] == metadata["metadata"]
|
||||
assert file_result is not None
|
||||
assert file_result["metadata"] == metadata["metadata"]
|
||||
assert file_result["messages"] == []
|
||||
|
||||
Reference in New Issue
Block a user