| """Shared test fixtures for Session Amplifier tests.""" |
| import json |
| import sqlite3 |
| import sys |
| import os |
| import pytest |
|
|
| |
| sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..")) |
|
|
| |
| 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, |
| }, |
| ] |
|
|
