(
process: subprocess.Popen[bytes],
responses: "queue.Queue[dict[str, Any]]",
scenario: str,
repo_path: str,
timeout: float,
)
| 192 | |
| 193 | |
| 194 | def run_scenario( |
| 195 | process: subprocess.Popen[bytes], |
| 196 | responses: "queue.Queue[dict[str, Any]]", |
| 197 | scenario: str, |
| 198 | repo_path: str, |
| 199 | timeout: float, |
| 200 | ) -> None: |
| 201 | request(process, responses, 1, "initialize", INITIALIZE_PARAMS, timeout) |
| 202 | send( |
| 203 | process, |
| 204 | {"jsonrpc": "2.0", "method": "notifications/initialized", "params": {}}, |
| 205 | ) |
| 206 | if scenario == "initialize": |
| 207 | return |
| 208 | if scenario == "invalid-index": |
| 209 | invalid_index_response = request( |
| 210 | process, |
| 211 | responses, |
| 212 | 2, |
| 213 | "tools/call", |
| 214 | { |
| 215 | "name": "index_repository", |
| 216 | "arguments": {"repo_path": repo_path, "mode": "fast"}, |
| 217 | }, |
| 218 | timeout, |
| 219 | accept_tool_error=True, |
| 220 | ) |
| 221 | invalid_index_result = invalid_index_response.get("result") |
| 222 | if ( |
| 223 | not isinstance(invalid_index_result, dict) |
| 224 | or invalid_index_result.get("isError") is not True |
| 225 | ): |
| 226 | raise SmokeFailure("index_repository unexpectedly accepted a nonexistent path") |
| 227 | request(process, responses, 3, "ping", {}, timeout) |
| 228 | return |
| 229 | index_response = request( |
| 230 | process, |
| 231 | responses, |
| 232 | 2, |
| 233 | "tools/call", |
| 234 | { |
| 235 | "name": "index_repository", |
| 236 | "arguments": {"repo_path": repo_path, "mode": "fast"}, |
| 237 | }, |
| 238 | timeout, |
| 239 | ) |
| 240 | index_result = index_response.get("result") |
| 241 | structured = index_result.get("structuredContent") if isinstance(index_result, dict) else None |
| 242 | project = structured.get("project") if isinstance(structured, dict) else None |
| 243 | if not isinstance(project, str) or not project: |
| 244 | raise SmokeFailure("index_repository response did not identify the indexed project") |
| 245 | if scenario == "roundtrip": |
| 246 | request( |
| 247 | process, |
| 248 | responses, |
| 249 | 3, |
| 250 | "tools/call", |
| 251 | { |
no test coverage detected