trezy.codes
forked from
slices.network/quickslice
Auto-indexing service and GraphQL API for AT Protocol Records
chadtmiller.com
1import database/executor.{type Executor}
2import database/repositories/lexicons
3import gleam/http
4import gleam/option
5import gleam/string
6import gleeunit/should
7import handlers/mcp
8import lib/oauth/did_cache
9import test_helpers
10import wisp
11import wisp/simulate
12
13fn setup_test_ctx() -> #(Executor, mcp.McpContext) {
14 let assert Ok(exec) = test_helpers.create_test_db()
15 let assert Ok(_) = test_helpers.create_lexicon_table(exec)
16 let assert Ok(_) = test_helpers.create_record_table(exec)
17 let assert Ok(did_cache) = did_cache.start()
18
19 let ctx =
20 mcp.McpContext(
21 db: exec,
22 external_base_url: "https://example.com",
23 did_cache: did_cache,
24 signing_key: option.None,
25 plc_url: "https://plc.directory",
26 supported_scopes: ["atproto", "transition:generic"],
27 )
28
29 #(exec, ctx)
30}
31
32pub fn handle_initialize_test() {
33 let #(_db, ctx) = setup_test_ctx()
34
35 let body =
36 "{\"jsonrpc\":\"2.0\",\"method\":\"initialize\",\"params\":{},\"id\":1}"
37 let req =
38 simulate.request(http.Post, "/mcp")
39 |> simulate.header("content-type", "application/json")
40 |> simulate.string_body(body)
41
42 let response = mcp.handle(req, ctx)
43
44 response.status |> should.equal(200)
45 let assert wisp.Text(body_result) = response.body
46 string.contains(body_result, "quickslice") |> should.be_true
47}
48
49pub fn handle_tools_list_test() {
50 let #(_db, ctx) = setup_test_ctx()
51
52 let body = "{\"jsonrpc\":\"2.0\",\"method\":\"tools/list\",\"id\":2}"
53 let req =
54 simulate.request(http.Post, "/mcp")
55 |> simulate.header("content-type", "application/json")
56 |> simulate.string_body(body)
57
58 let response = mcp.handle(req, ctx)
59
60 response.status |> should.equal(200)
61 let assert wisp.Text(body_result) = response.body
62 string.contains(body_result, "list_lexicons") |> should.be_true
63}
64
65pub fn handle_method_not_allowed_test() {
66 let #(_db, ctx) = setup_test_ctx()
67
68 let req = simulate.request(http.Get, "/mcp")
69
70 let response = mcp.handle(req, ctx)
71
72 response.status |> should.equal(405)
73}
74
75pub fn handle_tools_call_get_lexicon_test() {
76 let #(exec, ctx) = setup_test_ctx()
77
78 // Insert a test lexicon
79 let lexicon_json =
80 "{\"lexicon\":1,\"id\":\"test.example.status\",\"defs\":{\"main\":{\"type\":\"record\"}}}"
81 let assert Ok(_) = lexicons.insert(exec, "test.example.status", lexicon_json)
82
83 let body =
84 "{\"jsonrpc\":\"2.0\",\"method\":\"tools/call\",\"params\":{\"name\":\"get_lexicon\",\"arguments\":{\"nsid\":\"test.example.status\"}},\"id\":3}"
85 let req =
86 simulate.request(http.Post, "/mcp")
87 |> simulate.header("content-type", "application/json")
88 |> simulate.string_body(body)
89
90 let response = mcp.handle(req, ctx)
91
92 response.status |> should.equal(200)
93 let assert wisp.Text(body_result) = response.body
94 string.contains(body_result, "test.example.status") |> should.be_true
95}
96
97pub fn handle_tools_call_execute_query_test() {
98 let #(exec, ctx) = setup_test_ctx()
99
100 // Insert a test lexicon so schema builds
101 let lexicon_json =
102 "{\"lexicon\":1,\"id\":\"test.example.status\",\"defs\":{\"main\":{\"type\":\"record\",\"key\":\"tid\",\"record\":{\"type\":\"object\",\"properties\":{\"text\":{\"type\":\"string\"}}}}}}"
103 let assert Ok(_) = lexicons.insert(exec, "test.example.status", lexicon_json)
104
105 let body =
106 "{\"jsonrpc\":\"2.0\",\"method\":\"tools/call\",\"params\":{\"name\":\"execute_query\",\"arguments\":{\"query\":\"{ __typename }\"}},\"id\":4}"
107 let req =
108 simulate.request(http.Post, "/mcp")
109 |> simulate.header("content-type", "application/json")
110 |> simulate.string_body(body)
111
112 let response = mcp.handle(req, ctx)
113
114 response.status |> should.equal(200)
115 let assert wisp.Text(body_result) = response.body
116 string.contains(body_result, "Query") |> should.be_true
117}