rp/tests/test_session.py

import json
import os

import pytest

from rp.core.session import SessionManager


@pytest.fixture
def temp_sessions_dir(tmp_path, monkeypatch):
    from rp.core import session

    original_dir = session.SESSIONS_DIR
    monkeypatch.setattr(session, "SESSIONS_DIR", str(tmp_path))
    # Clean any existing files
    import shutil

    if os.path.exists(str(tmp_path)):
        shutil.rmtree(str(tmp_path))
    os.makedirs(str(tmp_path), exist_ok=True)
    yield tmp_path
    monkeypatch.setattr(session, "SESSIONS_DIR", original_dir)


def test_session_manager_init(temp_sessions_dir):
    SessionManager()
    assert os.path.exists(temp_sessions_dir)


def test_save_and_load_session(temp_sessions_dir):
    manager = SessionManager()
    name = "test_session"
    messages = [{"role": "user", "content": "Hello"}]
    metadata = {"test": True}

    assert manager.save_session(name, messages, metadata)

    loaded = manager.load_session(name)
    assert loaded is not None
    assert loaded["name"] == name
    assert loaded["messages"] == messages
    assert loaded["metadata"] == metadata


def test_load_nonexistent_session(temp_sessions_dir):
    manager = SessionManager()
    loaded = manager.load_session("nonexistent")
    assert loaded is None


def test_list_sessions(temp_sessions_dir):
    manager = SessionManager()
    # Save a session
    manager.save_session("session1", [{"role": "user", "content": "Hi"}])
    manager.save_session("session2", [{"role": "user", "content": "Hello"}])

    sessions = manager.list_sessions()
    assert len(sessions) == 2
    assert sessions[0]["name"] == "session2"  # sorted by created_at desc


def test_delete_session(temp_sessions_dir):
    manager = SessionManager()
    name = "to_delete"
    manager.save_session(name, [{"role": "user", "content": "Test"}])

    assert manager.delete_session(name)
    assert manager.load_session(name) is None


def test_delete_nonexistent_session(temp_sessions_dir):
    manager = SessionManager()
    assert not manager.delete_session("nonexistent")


def test_export_session_json(temp_sessions_dir, tmp_path):
    manager = SessionManager()
    name = "export_test"
    messages = [{"role": "user", "content": "Export me"}]
    manager.save_session(name, messages)

    output_path = tmp_path / "exported.json"
    assert manager.export_session(name, str(output_path), "json")
    assert output_path.exists()

    with open(output_path) as f:
        data = json.load(f)
    assert data["name"] == name


def test_export_session_markdown(temp_sessions_dir, tmp_path):
    manager = SessionManager()
    name = "export_md"
    messages = [{"role": "user", "content": "Markdown export"}]
    manager.save_session(name, messages)

    output_path = tmp_path / "exported.md"
    assert manager.export_session(name, str(output_path), "markdown")
    assert output_path.exists()

    content = output_path.read_text()
    assert "# Session: export_md" in content


def test_export_session_txt(temp_sessions_dir, tmp_path):
    manager = SessionManager()
    name = "export_txt"
    messages = [{"role": "user", "content": "Text export"}]
    manager.save_session(name, messages)

    output_path = tmp_path / "exported.txt"
    assert manager.export_session(name, str(output_path), "txt")
    assert output_path.exists()

    content = output_path.read_text()
    assert "Session: export_txt" in content


def test_export_nonexistent_session(temp_sessions_dir, tmp_path):
    manager = SessionManager()
    output_path = tmp_path / "nonexistent.json"
    assert not manager.export_session("nonexistent", str(output_path), "json")


def test_export_unsupported_format(temp_sessions_dir, tmp_path):
    manager = SessionManager()
    name = "test"
    manager.save_session(name, [{"role": "user", "content": "Test"}])

    output_path = tmp_path / "test.unsupported"
    assert not manager.export_session(name, str(output_path), "unsupported")