File size: 5,260 Bytes
63c75d5 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 | """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,
},
]
|