A 5e storytelling engine with an LLM DM
1"""Tests for the claude -p subprocess driver."""
2
3import json
4
5import pytest
6
7from storied.claude import (
8 Result,
9 TextDelta,
10 ToolInputDelta,
11 ToolStart,
12 ToolStop,
13 build_claude_args,
14 build_mcp_config,
15 format_user_message,
16 parse_event,
17)
18
19
20class TestBuildMcpConfig:
21 def test_produces_valid_json(self):
22 config = build_mcp_config("http://127.0.0.1:9999/sse")
23 parsed = json.loads(config)
24 assert "mcpServers" in parsed
25 assert "storied" in parsed["mcpServers"]
26
27 def test_url_set(self):
28 config = json.loads(build_mcp_config("http://127.0.0.1:8080/sse"))
29 server = config["mcpServers"]["storied"]
30 assert server["url"] == "http://127.0.0.1:8080/sse"
31 assert server["type"] == "sse"
32
33
34class TestBuildClaudeArgs:
35 def test_new_session_includes_system_prompt(self, monkeypatch):
36 monkeypatch.setattr("shutil.which", lambda _: "/usr/bin/claude")
37 args = build_claude_args("sonnet", "You are a DM", "{}")
38 assert "--system-prompt" in args
39 assert "You are a DM" in args
40 assert "--model" in args
41 assert "sonnet" in args
42
43 def test_resume_session_includes_system_prompt(self, monkeypatch):
44 monkeypatch.setattr("shutil.which", lambda _: "/usr/bin/claude")
45 args = build_claude_args(
46 "sonnet",
47 "You are a DM",
48 "{}",
49 resume_session_id="abc-123",
50 )
51 assert "--resume" in args
52 assert "abc-123" in args
53 assert "--system-prompt" in args
54 assert "You are a DM" in args
55 assert "--model" in args
56 assert "sonnet" in args
57
58 def test_no_session_persistence_flag(self, monkeypatch):
59 monkeypatch.setattr("shutil.which", lambda _: "/usr/bin/claude")
60 args = build_claude_args("sonnet", "sys", "{}", persist_session=False)
61 assert "--no-session-persistence" in args
62
63 def test_claude_not_found_raises(self, monkeypatch):
64 monkeypatch.setattr("shutil.which", lambda _: None)
65 with pytest.raises(FileNotFoundError, match="Claude Code CLI not found"):
66 build_claude_args("sonnet", "sys", "{}")
67
68 def test_tools_disabled(self, monkeypatch):
69 monkeypatch.setattr("shutil.which", lambda _: "/usr/bin/claude")
70 args = build_claude_args("sonnet", "sys", "{}")
71 idx = args.index("--tools")
72 assert args[idx + 1] == ""
73
74 def test_slash_commands_disabled(self, monkeypatch):
75 monkeypatch.setattr("shutil.which", lambda _: "/usr/bin/claude")
76 args = build_claude_args("sonnet", "sys", "{}")
77 assert "--disable-slash-commands" in args
78
79
80class TestFormatUserMessage:
81 def test_produces_ndjson(self):
82 msg = format_user_message("Hello world")
83 parsed = json.loads(msg)
84 assert parsed["type"] == "user"
85 assert parsed["message"]["role"] == "user"
86 content = parsed["message"]["content"]
87 assert len(content) == 1
88 assert content[0]["type"] == "text"
89 assert content[0]["text"] == "Hello world"
90
91 def test_ends_with_newline(self):
92 msg = format_user_message("test")
93 assert msg.endswith(b"\n")
94
95
96class TestParseEvent:
97 def test_text_delta(self):
98 line = json.dumps(
99 {
100 "type": "stream_event",
101 "event": {
102 "type": "content_block_delta",
103 "index": 0,
104 "delta": {"type": "text_delta", "text": "Hello"},
105 },
106 }
107 )
108 event = parse_event(line)
109 assert isinstance(event, TextDelta)
110 assert event.text == "Hello"
111
112 def test_tool_start(self):
113 line = json.dumps(
114 {
115 "type": "stream_event",
116 "event": {
117 "type": "content_block_start",
118 "index": 1,
119 "content_block": {
120 "type": "tool_use",
121 "id": "t1",
122 "name": "mcp__storied__roll",
123 },
124 },
125 }
126 )
127 event = parse_event(line)
128 assert isinstance(event, ToolStart)
129 assert event.name == "mcp__storied__roll"
130
131 def test_tool_input_delta(self):
132 line = json.dumps(
133 {
134 "type": "stream_event",
135 "event": {
136 "type": "content_block_delta",
137 "index": 1,
138 "delta": {
139 "type": "input_json_delta",
140 "partial_json": '{"notation"',
141 },
142 },
143 }
144 )
145 event = parse_event(line)
146 assert isinstance(event, ToolInputDelta)
147 assert event.json_fragment == '{"notation"'
148
149 def test_tool_stop(self):
150 line = json.dumps(
151 {
152 "type": "stream_event",
153 "event": {"type": "content_block_stop", "index": 1},
154 }
155 )
156 event = parse_event(line)
157 assert isinstance(event, ToolStop)
158
159 def test_result(self):
160 line = json.dumps(
161 {
162 "type": "result",
163 "session_id": "sess-1",
164 "usage": {"input_tokens": 100, "output_tokens": 50},
165 "duration_ms": 1234,
166 }
167 )
168 event = parse_event(line)
169 assert isinstance(event, Result)
170 assert event.session_id == "sess-1"
171 assert event.usage["input_tokens"] == 100
172
173 def test_empty_line(self):
174 assert parse_event("") is None
175 assert parse_event(" ") is None
176
177 def test_invalid_json(self):
178 assert parse_event("not json") is None
179
180 def test_unknown_event_type(self):
181 assert parse_event(json.dumps({"type": "unknown"})) is None
182
183 def test_text_block_start_ignored(self):
184 line = json.dumps(
185 {
186 "type": "stream_event",
187 "event": {
188 "type": "content_block_start",
189 "index": 0,
190 "content_block": {"type": "text"},
191 },
192 }
193 )
194 assert parse_event(line) is None