tests/conftest.py

"""Shared test fixtures for Session Amplifier tests."""
import json
import sqlite3
import sys
import os
import pytest

# Add project root to path so imports work as they do in production
sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))

# Override settings before importing anything that uses them
os.environ.setdefault("OPENCLAW_AGENTS_ROOT", os.path.join(os.path.dirname(__file__), "_fake_agents"))
os.environ.setdefault("OPENCLAW_STATE_DIR", os.path.join(os.path.dirname(__file__), "_fake_state"))
os.environ.setdefault("SESSION_AMPLIFIER_DB_PATH", ":memory:")


@pytest.fixture
def in_memory_db():
    """Create an in-memory SQLite DB with the session amplifier schema."""
    conn = sqlite3.connect(":memory:")
    conn.row_factory = sqlite3.Row
    conn.executescript("""
        CREATE TABLE IF NOT EXISTS spooled_entries (
            id INTEGER PRIMARY KEY,
            session_id TEXT NOT NULL,
            agent_id TEXT NOT NULL,
            entry_idx INTEGER NOT NULL,
            entry_type TEXT,
            role TEXT,
            timestamp TEXT,
            tool_name TEXT,
            clean_text TEXT,
            original_length INTEGER,
            preview TEXT,
            is_error INTEGER DEFAULT 0,
            indexed_at TEXT DEFAULT (datetime('now')),
            UNIQUE(session_id, entry_idx)
        );
        CREATE TABLE IF NOT EXISTS spool_state (
            key TEXT PRIMARY KEY,
            value TEXT
        );
        CREATE TABLE IF NOT EXISTS review_reports (
            id INTEGER PRIMARY KEY,
            review_id TEXT UNIQUE,
            generated_at TEXT,
            period_from TEXT,
            period_to TEXT,
            report_json TEXT
        );
    """)
    yield conn
    conn.close()


@pytest.fixture
def sample_transcript_lines():
    """Sample JSONL transcript lines for testing."""
    return [
        json.dumps({"type": "session", "id": "test-session-001"}),
        json.dumps({
            "type": "message",
            "timestamp": "2026-04-05T10:00:00Z",
            "message": {
                "role": "user",
                "content": [{"type": "text", "text": "Hello, can you help me?"}],
            },
        }),
        json.dumps({
            "type": "message",
            "timestamp": "2026-04-05T10:00:05Z",
            "message": {
                "role": "assistant",
                "content": [{"type": "text", "text": "Of course! Let me look into that for you."}],
            },
        }),
        json.dumps({
            "type": "message",
            "timestamp": "2026-04-05T10:00:10Z",
            "message": {
                "role": "toolResult",
                "content": [{"type": "text", "text": "file contents: line1\nline2\nline3"}],
                "toolName": "read_file",
            },
        }),
        json.dumps({
            "type": "message",
            "timestamp": "2026-04-05T10:00:15Z",
            "message": {
                "role": "toolResult",
                "content": [{"type": "text", "text": ""}],
                "toolName": "write_file",
                "isError": True,
            },
        }),
        json.dumps({
            "type": "message",
            "timestamp": "2026-04-05T10:00:20Z",
            "message": {
                "role": "assistant",
                "content": [{"type": "text", "text": "I'll use the read_file tool to check."}],
            },
        }),
    ]


@pytest.fixture
def sample_spooled_entries():
    """Pre-processed spooled entry dicts for reviewer tests."""
    return [
        {
            "session_id": "s1",
            "agent_id": "ops",
            "entry_idx": 1,
            "entry_type": "message",
            "role": "user",
            "timestamp": "2026-04-05T10:00:00Z",
            "tool_name": "",
            "clean_text": "Help me debug this issue",
            "original_length": 25,
            "preview": "Help me debug this issue",
            "is_error": 0,
        },
        {
            "session_id": "s1",
            "agent_id": "ops",
            "entry_idx": 2,
            "entry_type": "message",
            "role": "assistant",
            "timestamp": "2026-04-05T10:00:05Z",
            "tool_name": "",
            "clean_text": "I'll use the read_file tool to inspect the code.",
            "original_length": 48,
            "preview": "I'll use the read_file tool to inspect the code.",
            "is_error": 0,
        },
        {
            "session_id": "s1",
            "agent_id": "ops",
            "entry_idx": 3,
            "entry_type": "message",
            "role": "toolResult",
            "timestamp": "2026-04-05T10:00:10Z",
            "tool_name": "read_file",
            "clean_text": "x" * 5000,
            "original_length": 5000,
            "preview": "x" * 300,
            "is_error": 0,
        },
        {
            "session_id": "s1",
            "agent_id": "ops",
            "entry_idx": 4,
            "entry_type": "message",
            "role": "toolResult",
            "timestamp": "2026-04-05T10:00:15Z",
            "tool_name": "write_file",
            "clean_text": "",
            "original_length": 0,
            "preview": "",
            "is_error": 1,
        },
    ]